Skip to content

Python: AG-UI snapshot replay with service sessions blocks Foundry Hosted Agents #7710

Description

@djw-bsn

AgentFrameworkAgent currently accepts both use_service_session=True and a
snapshot_store.

The basic service-session path works correctly when the Foundry-generated
conv_* ID is used as threadId, snapshots are disabled, and the client sends
only incremental input. The issue is specifically introduced by combining that
working service-session path with AG-UI snapshot persistence.

On the second request, the AG-UI runner:

  1. reconstructs the complete stored transcript from the thread snapshot, and
  2. creates AgentSession(service_session_id=threadId).

The wrapped agent therefore receives both a provider continuation handle and
explicit prior conversation history. Some providers may tolerate this. Foundry
Hosted Agents do not accept the replayed assistant input, so this combination
breaks hosted-agent continuation through AG-UI.

This appears inconsistent with the package's documented “one State Authority”
model and is analogous to #4292, where AgentExecutor retaining
service_session_id during full-history replay creates competing continuation
sources. PR #7682 does not cover AG-UI because AG-UI invokes the agent through
agent_framework_ag_ui._agent_run.run_agent_stream.

Could the intended contract for this option combination be clarified? If
service-session mode is authoritative, AG-UI should likely pass only incremental
input to the provider while retaining complete snapshot messages for client
hydration. If snapshot replay is authoritative, AG-UI should clear the service
session ID or reject the ambiguous configuration.

Blocking integration scenario

This blocks multi-turn applications using this public integration stack:

AG-UI client
    -> Agent Framework Python AgentFrameworkAgent
    -> Agent Framework FoundryAgent
    -> Foundry Hosted Agent using the Responses protocol

It does not block applications that can use the Foundry-generated conv_* as
their public AG-UI thread ID, send only incremental turns, and do not require
Agent Framework snapshot persistence. Those applications can use
use_service_session=True without the snapshot store.

The blocked case needs both:

  • Foundry service-managed conversation history for hosted-agent provider calls;
  • Agent Framework AG-UI snapshots for backend-owned UI hydration, approvals, or
    other replayable thread state.

Foundry documents different responsibilities for hosted-agent conversations and
sessions:

  • a conversation ID is the durable conversation-history record;
  • a session ID identifies hosted compute and persisted filesystem state;
  • under the Responses protocol, conversation ID is the primary continuation
    concept and the platform associates compute session state with it.

The application therefore needs AG-UI to retain complete messages for UI
hydration while invoking the hosted agent incrementally through its Foundry
conversation. With use_service_session=True and snapshots enabled, the current
runner instead sends the conversation ID together with reconstructed prior
assistant history.

Observed compatibility differs by Foundry agent type:

  • Prompt-based Foundry agents accept replayed conversation messages and can
    continue under snapshot-authoritative history.
  • Hosted Foundry agents using Responses reject the reconstructed assistant
    message input with HTTP 400 schema validation.
  • Hosted continuation succeeds when a Foundry conversation is used with only
    incremental input.

No incorrect prompt-agent behavior was observed in these tests. The actionable
failure reported here is the hosted-agent integration; the framework-only spy
reproduction explains the AG-UI request shape that triggers it.

These results were repeated with the direct Azure AI Projects/OpenAI Responses
client using get_openai_client(agent_name=...), with no Agent Framework client
or message transformation. Direct prompt replay passed; direct hosted replay
failed. Therefore, the hosted replay limitation is not caused by
agent-framework-foundry.

Direct tests against the hosted-agent Responses endpoint showed:

  • as expected, agent_session_id alone reuses hosted compute/filesystem state
    without continuing conversation history, so it can be used for a fresh
    conversation on the same compute session;
  • previous_response_id and conversation=conv_* both continue history;
  • successful response/conversation continuation automatically preserves the
    same hosted session ID without explicitly resending it.

Consequently, replay authority is not a usable fallback for this hosted-agent
integration. Rejecting the snapshot/service option combination would make the
incompatibility clearer but would still leave the integration unsupported. A
service-authoritative mode that separates provider input from UI snapshot
history is needed for this use case.

This section describes integration impact, not proof that the hosted service is
defective. The framework-only reproduction above remains the issue's minimal
reproduction.

Temporary workaround

Hosted-agent continuation works when the application:

  1. creates a Foundry conversation and uses its conv_* ID as the AG-UI
    threadId;
  2. sets use_service_session=True;
  3. does not configure an Agent Framework snapshot store; and
  4. sends only the incremental user/tool input for each continuation request.

This uses the Foundry conversation as the sole provider-history authority. The
tradeoff is that Agent Framework cannot provide snapshot-backed message
hydration or related persisted thread state in this configuration. Applications
that require both currently need a custom boundary that retains complete
snapshots for the UI but passes only incremental input to the wrapped agent.

Removing assistant messages alone is not a complete workaround when no service
conversation is present, because information contained only in prior assistant
responses is then lost.

Package versions

  • agent-framework-core==1.14.0
  • agent-framework-ag-ui==1.1.0
  • Python 3.12
  • Linux

Minimal framework-only reproduction

from __future__ import annotations

from typing import Any

import pytest
from agent_framework import AgentResponseUpdate
from agent_framework.ag_ui import (
    AgentFrameworkAgent,
    InMemoryAGUIThreadSnapshotStore,
)


class InputSpyAgent:
    name = "spy"
    description = ""
    default_options: dict[str, Any] = {}
    context_providers: list[Any] = []

    def __init__(self) -> None:
        self.calls: list[dict[str, Any]] = []

    def run(self, messages, *, session, stream=False, **kwargs):
        async def updates():
            self.calls.append(
                {
                    "roles": [message.role for message in messages],
                    "service_session_id": session.service_session_id,
                }
            )
            yield AgentResponseUpdate(
                contents=["ACK"],
                role="assistant",
                response_id=f"response-{len(self.calls)}",
            )

        return updates()


async def drain(runner, body):
    return [event async for event in runner.run(body)]


@pytest.mark.asyncio
async def test_service_session_has_one_history_authority():
    agent = InputSpyAgent()
    runner = AgentFrameworkAgent(
        agent=agent,
        use_service_session=True,
        snapshot_store=InMemoryAGUIThreadSnapshotStore(),
    )

    first = {
        "threadId": "conv_PROVIDER_SESSION",
        "__ag_ui_snapshot_scope": "test-scope",
        "messages": [{"id": "u1", "role": "user", "content": "first"}],
    }
    second = {
        "threadId": "conv_PROVIDER_SESSION",
        "__ag_ui_snapshot_scope": "test-scope",
        "messages": [{"id": "u2", "role": "user", "content": "second"}],
    }

    await drain(runner, first)
    await drain(runner, second)

    assert agent.calls[1] == {
        "roles": ["user"],
        "service_session_id": "conv_PROVIDER_SESSION",
    }

Actual behavior

{
    "roles": ["user", "assistant", "user"],
    "service_session_id": "conv_PROVIDER_SESSION",
}

Expected contract

The invocation should have one conversation-history authority. If
use_service_session=True selects provider history, the likely expected shape
is:

{
    "roles": ["user"],
    "service_session_id": "conv_PROVIDER_SESSION",
}

If that is not the intended policy, the option combination should be rejected or
documented as replay-authoritative with service_session_id cleared.

Relevant Agent Framework code

agent_framework_ag_ui/_agent_run.py:

  • calls _reconstruct_messages_from_thread_snapshot() whenever a snapshot is
    present;
  • creates AgentSession(session_id=thread_id, service_session_id=supplied_thread_id) when use_service_session=True;
  • does not condition reconstruction on use_service_session.

agent_framework_ag_ui/_run_common.py:

  • _reconstruct_messages_from_thread_snapshot() prepends stored messages to
    the new request suffix.

agent_framework/_workflows/_agent_executor.py:

Questions for maintainers

  1. Is snapshot_store supported together with use_service_session=True?
  2. If supported, which source owns provider conversation history: snapshot
    messages or service_session_id?
  3. Should snapshot messages remain UI/hydration state only in service mode?
  4. How should incremental tool results, approval responses, and interrupt resumes
    be selected without replaying server-owned history?
  5. Should the API expose an explicit replay versus service continuation
    policy?

It would be helpful to document whether snapshots and service sessions are
intended to be used together and, if so, how prior history should be supplied to
the wrapped agent. Regression coverage for ordinary turns and tool/resume flows
would help keep that contract stable.

Metadata

Metadata

Labels

pythonUsage: [Issues, PRs], Target: PythonreproducedUsage: [Issues], Target: all issues that can be reproduced by the triage workflow

Type

Projects

Status
Done

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions