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 5f904cf217..faffb0c424 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 = ( @@ -617,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) @@ -630,6 +645,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 +683,31 @@ async def _handle_inner_agent( finally: if self._uses_hosted_responses_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) + + # 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 + 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 cd7cdef316..216a391c67 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,104 @@ 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_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, + ) -> 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)