From c9b8abac57bc97ba7fe90966efa56dc81169daf1 Mon Sep 17 00:00:00 2001 From: atty57 <99388680+atty57@users.noreply.github.com> Date: Mon, 31 Aug 2026 00:06:54 -0400 Subject: [PATCH 1/2] Python: Give a hosted Foundry agent a single source of conversation history A hosted `/responses` agent fed the model the conversation transcript more than once per request, and the duplication compounded every turn: by the third turn the model saw the first turn three times. `_handle_inner_agent` gives `agent.run(...)` the full platform transcript from `context.get_history()` as `messages`, and separately the session loaded from the session store. The existing guard pops the transient history buffer out of `session.state`, but the session's `service_session_id` survives, and core resumes it. The run therefore continues a service-side thread that already holds the whole transcript while `messages` carries that transcript again, and the service appends the duplicated request to the thread, so each turn grows superlinearly. This is rarely seen because the default session store's reads currently fail and return None, so the service thread is never resumed. Any working session store exposes it. Turn server-side storage off for the run instead, so the platform record is the only history in play, mirroring the .NET behaviour from #7525 and #7572: - Set `store=False` on the run's chat options when hosting manages history. The chat client then keeps nothing of its own and reports no conversation id, so no second thread exists to resume. - If a service session id lands on the session anyway, the client stored the turn despite that setting and a second unreconciled record now exists. Fail the request and leave the session unsaved, so later turns cannot resume onto the duplicated thread. - Add `allow_stored_output_enabled` to `ResponsesHostServer` for containers that want the chat client left exactly as they configured it. Nothing is then overridden or checked, and reconciling the two records is theirs to own. Fixes #7955 --- .../_responses.py | 38 ++++++- .../foundry_hosting/tests/test_responses.py | 102 ++++++++++++++++++ 2 files changed, 139 insertions(+), 1 deletion(-) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 5f904cf217f..7a91f8315e7 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -336,6 +336,7 @@ def __init__( agent_session_store_provider: StoreProvider[SessionStore] | None = None, checkpoint_store_provider: ContextScopedStoreProvider[CheckpointStorage] | None = None, function_approval_store_provider: StoreProvider[FunctionApprovalStore] | None = None, + allow_stored_output_enabled: bool = False, **kwargs: Any, ) -> None: """Initialize a ResponsesHostServer. @@ -351,6 +352,12 @@ def __init__( If not provided, a default `CheckpointStoreProvider` will be used. function_approval_store_provider: Optional provider for function approval storage. If not provided, a default `FunctionApprovalStoreProvider` will be used. + allow_stored_output_enabled: Whether the agent's own chat client may store the + conversation server-side. Defaults to False, which turns storage off for every + run and fails the request if the client stored the turn anyway. Set to True to + leave the client exactly as the container configured it; nothing is then + overridden or checked, and reconciling the two records is the container's + responsibility. **kwargs: Additional keyword arguments. Note: @@ -416,6 +423,7 @@ def __init__( ) self._agent: SupportsAgentRun = agent + self._allow_stored_output_enabled = allow_stored_output_enabled # Storage providers self._checkpoint_storage_provider = ( @@ -630,6 +638,12 @@ async def _handle_inner_agent( "session": session, } chat_options, are_options_set = _to_chat_options(request) + if self._uses_hosted_responses_history and not self._allow_stored_output_enabled: + # The platform records this conversation and serves it back through + # `context.get_history()` above. Letting the agent's own service store it too would + # give the model the same transcript twice -- once as input, once as the resumed + # service thread -- compounding every turn. + chat_options["store"] = False if are_options_set and not isinstance(self._agent, RawAgent): logger.warning("Agent doesn't support runtime options. They will be ignored.") @@ -662,8 +676,30 @@ async def _handle_inner_agent( finally: if self._uses_hosted_responses_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) + + # A service session id on the session means the agent's chat client kept the turn + # despite `store=False`, so a second record of this conversation now exists that + # nothing here reconciles. Leave the session unsaved so later turns do not resume onto + # it, and report it: a container configured this way is a server fault, not a bad request. + stored_output_violation = ( + self._uses_hosted_responses_history + and not self._allow_stored_output_enabled + and session.service_session_id is not None + ) + if stored_output_violation: + misconfigured = RuntimeError( + "The agent's chat client stored this turn server-side while the hosting service " + "is managing the conversation, which would feed the model a duplicated transcript. " + "Configure the chat client so the underlying service does not store responses, or " + "construct ResponsesHostServer with allow_stored_output_enabled=True to own that " + "reconciliation yourself." + ) + logger.error("%s", misconfigured) + if request_failure is None and not request_interrupted: + request_failure = misconfigured try: - await session_storage.set(session_save_id, session) + if not stored_output_violation: + await session_storage.set(session_save_id, session) except Exception as save_error: save_failure = save_error if request_interrupted: diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index cd7cdef3160..1949b238d87 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -12,6 +12,7 @@ import asyncio import json +import logging import uuid from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass @@ -262,6 +263,44 @@ def _lookup_weather(location: str) -> str: return f"Weather in {location}: sunny" +class _ServiceStorageRecordingClient(BaseChatClient): + """A chat client whose service keeps the conversation, as OpenAI Responses does with ``store`` on.""" + + STORES_BY_DEFAULT = True + + def __init__(self, *, honours_store: bool = True) -> None: + super().__init__() + self._honours_store = honours_store + self.calls: list[list[Message]] = [] + self.store_options: list[Any] = [] + self.conversation_ids: list[Any] = [] + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + del kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." + self.calls.append(list(messages)) + self.store_options.append(options.get("store")) + self.conversation_ids.append(options.get("conversation_id")) + storing = options.get("store") is not False if self._honours_store else True + conversation_id = "svc-thread-1" if storing else None + + async def stream_response() -> AsyncIterator[ChatResponseUpdate]: + yield ChatResponseUpdate( + contents=[Content.from_text("recorded")], + role="assistant", + conversation_id=conversation_id, + ) + + return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) + + class _FailingSessionStore(SessionStore): def __init__(self) -> None: super().__init__() @@ -739,6 +778,69 @@ async def test_responses_history_is_not_duplicated_by_default_local_history(self assert InMemoryHistoryProvider.DEFAULT_SOURCE_ID not in stored.state assert "_foundry_responses_history" not in stored.state + async def test_responses_history_is_not_duplicated_when_the_client_stores_service_side(self) -> None: + """The platform transcript is the only history the model sees, even for a storing client. + + Without turning storage off downstream the session keeps a service session id, the next run + resumes that thread, and the model receives the transcript both as input and as the resumed + thread -- twice on the second turn and three times on the third. + """ + client = _ServiceStorageRecordingClient() + agent = Agent(client=client, name="Service Storage Agent") + store = SessionStore() + server = _make_server(agent, session_store=store) + + first = await _post(server, input_text="first") + second = await _post(server, input_text="second", previous_response_id=first.json()["id"]) + third = await _post(server, input_text="third", previous_response_id=second.json()["id"]) + + assert third.status_code == 200 + assert third.json()["status"] == "completed" + assert [[message.text for message in call] for call in client.calls] == [ + ["first"], + ["first", "recorded", "second"], + ["first", "recorded", "second", "recorded", "third"], + ] + # Storage is off downstream, so no service thread is ever resumed alongside that input. + assert client.store_options == [False, False, False] + assert client.conversation_ids == [None, None, None] + + stored = await store.get(first.json()["id"]) + assert stored is not None + assert stored.service_session_id is None + + async def test_client_that_stores_despite_disabled_storage_fails_and_leaves_session_unsaved( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + client = _ServiceStorageRecordingClient(honours_store=False) + agent = Agent(client=client, name="Ignores Store Agent") + store = SessionStore() + server = _make_server(agent, session_store=store) + + with caplog.at_level(logging.ERROR): + response = await _post(server, input_text="first") + + assert response.json()["status"] == "failed" + assert "stored this turn server-side" in caplog.text + # The session is not saved, so a later turn cannot resume onto the duplicated thread. + assert await store.get(response.json()["id"]) is None + + async def test_allow_stored_output_enabled_leaves_the_client_configuration_untouched(self) -> None: + client = _ServiceStorageRecordingClient() + agent = Agent(client=client, name="Container Managed Storage Agent") + store = SessionStore() + server = _make_server(agent, session_store=store, allow_stored_output_enabled=True) + + response = await _post(server, input_text="first") + + assert response.json()["status"] == "completed" + assert client.store_options == [None] + + stored = await store.get(response.json()["id"]) + assert stored is not None + assert stored.service_session_id == "svc-thread-1" + async def test_per_service_call_persistence_preserves_function_loop_history(self) -> None: provider = _PerServiceCallHistoryProvider() client = _FunctionLoopRecordingClient(provider) From 7dcf9cb466e51aa82ef2c5cb0d749c45466fb4d3 Mon Sep 17 00:00:00 2001 From: atty57 <99388680+atty57@users.noreply.github.com> Date: Mon, 31 Aug 2026 17:10:50 -0400 Subject: [PATCH 2/2] Python: Drop a stale service thread id from a hosted session on load A conversation that began before storage was disabled persisted a session naming the service thread that already holds its transcript. Core resumes that thread ahead of anything in the run options, so the model still saw the transcript twice, and the storage check then failed the turn even though the client had honoured store=False. Because a failed turn is not saved, the same stale session loaded again on every later turn. Clearing service_session_id where the transient history buffer is popped heals those conversations on their first post-upgrade turn, and makes the check exact: an id present afterwards can only have been set by this run. --- .../_responses.py | 16 ++++++--- .../foundry_hosting/tests/test_responses.py | 35 +++++++++++++++++++ 2 files changed, 47 insertions(+), 4 deletions(-) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 7a91f8315e7..faffb0c424e 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -625,6 +625,13 @@ async def _handle_inner_agent( try: if self._uses_hosted_responses_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) + if not self._allow_stored_output_enabled: + # A service session id on a loaded session points at a thread that already holds + # this conversation, written before storage was turned off. Core resumes it ahead + # of anything in the options, so it would duplicate the transcript the platform + # already supplies. Dropping it also makes the check in `finally` exact: an id + # present by then can only have been set by this run. + session.service_session_id = None input_items = await context.get_input_items() input_messages = await _items_to_messages(input_items, approval_storage=approval_storage) @@ -677,10 +684,11 @@ async def _handle_inner_agent( if self._uses_hosted_responses_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) - # A service session id on the session means the agent's chat client kept the turn - # despite `store=False`, so a second record of this conversation now exists that - # nothing here reconciles. Leave the session unsaved so later turns do not resume onto - # it, and report it: a container configured this way is a server fault, not a bad request. + # The session started this run without a service session id, so one now means the + # agent's chat client kept the turn despite `store=False`, and a second record of this + # conversation exists that nothing here reconciles. Leave the session unsaved so later + # turns do not resume onto it, and report it: a container configured this way is a + # server fault, not a bad request. stored_output_violation = ( self._uses_hosted_responses_history and not self._allow_stored_output_enabled diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 1949b238d87..216a391c67a 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -809,6 +809,41 @@ async def test_responses_history_is_not_duplicated_when_the_client_stores_servic assert stored is not None assert stored.service_session_id is None + async def test_session_carrying_a_pre_fix_service_thread_id_heals_instead_of_failing(self) -> None: + """A conversation started before storage was disabled continues cleanly after an upgrade. + + Its persisted session still names the service thread that holds the whole transcript, and + core resumes that thread ahead of anything in the options. Left in place it duplicates the + turn and trips the storage violation, and since a failed turn is never re-saved the same + stale session would load again on every later turn. + """ + client = _ServiceStorageRecordingClient() + agent = Agent(client=client, name="Upgraded Agent") + store = SessionStore() + server = _make_server(agent, session_store=store) + + first = await _post(server, input_text="first") + # What a pre-fix version left behind: the service thread that already holds turn 1. + upgraded = await store.get(first.json()["id"]) + assert upgraded is not None + upgraded.service_session_id = "svc-thread-1" + await store.set(first.json()["id"], upgraded) + + second = await _post(server, input_text="second", previous_response_id=first.json()["id"]) + + assert second.status_code == 200 + assert second.json()["status"] == "completed" + # The stale thread is neither resumed nor mistaken for a client that stored the turn. + assert client.conversation_ids == [None, None] + assert [[message.text for message in call] for call in client.calls] == [ + ["first"], + ["first", "recorded", "second"], + ] + + stored = await store.get(second.json()["id"]) + assert stored is not None + assert stored.service_session_id is None + async def test_client_that_stores_despite_disabled_storage_fails_and_leaves_session_unsaved( self, caplog: pytest.LogCaptureFixture,