Python: Give a hosted Foundry agent a single source of conversation history

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:

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)