Python: omit failed Foundry turns from conversation chat history

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -28,6 +28,7 @@
 from agent_framework.exceptions import AgentFrameworkException
 from azure.ai.agentserver.core import get_request_context
 from azure.ai.agentserver.responses import (
+    InMemoryResponseProvider,
     ResponseContext,
     ResponseProviderProtocol,
     ResponsesServerOptions,
@@ -167,6 +168,68 @@ def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None:
 # endregion Foundry Toolbox Auth integration
 
 
+def _response_field(response: Any, name: str) -> Any:
+    """Read a field from a mapping or attribute-bearing response envelope."""
+    if isinstance(response, Mapping):
+        return response.get(name)
+    return getattr(response, name, None)
+
+
+def _is_failed_stored_response(response: Any) -> bool:
+    """Return whether a persisted response envelope is a failed turn."""
+    return _response_field(response, "status") == "failed"
+
+
+class _OmitFailedConversationInputProvider:
+    """Omit failed-turn input from the Responses chat-history store.
+
+    The agentserver orchestrator persists input items for every stored response,
+    including ``status=failed``. Conversation history then replays those items on
+    the next turn, which is the #7630 failure mode. Azure OpenAI does not keep
+    failed input on the conversation.
+
+    Non-streaming ``store=true`` requests often ``create_response`` while still
+    ``in_progress`` and later ``update_response`` to ``failed``. The create path
+    therefore drops input only when the envelope is already failed; the update
+    path replaces a failed envelope so previously stored input is not replayed.
+    """
+
+    def __init__(self, inner: ResponseProviderProtocol) -> None:
+        """Wrap ``inner`` so failed turns persist without input items."""
+        self._inner = inner
+
+    async def create_response(
+        self,
+        response: Any,
+        input_items: Any,
+        history_item_ids: Any,
+        *,
+        context: Any = None,
+    ) -> None:
+        """Persist ``response``, dropping input items when the turn failed."""
+        if _is_failed_stored_response(response):
+            input_items = None
+        await self._inner.create_response(response, input_items, history_item_ids, context=context)
+
+    async def update_response(self, response: Any, *, context: Any = None) -> None:
+        """Update ``response``, dropping stored input when the turn failed."""
+        if not _is_failed_stored_response(response):
+            await self._inner.update_response(response, context=context)
+            return
+
+        response_id = _response_field(response, "id")
+        if response_id is not None:
+            try:
+                await self._inner.delete_response(str(response_id), context=context)
+            except (KeyError, ValueError):
+                pass
+        await self._inner.create_response(response, None, None, context=context)
+
+    def __getattr__(self, name: str) -> Any:
+        """Forward remaining provider methods to the wrapped store."""
+        return getattr(self._inner, name)
+
+
 # region ResponsesHostServer
 class ResponsesHostServer(ResponsesAgentServerHost):
     """A responses server host for an agent."""
@@ -205,7 +268,18 @@ def __init__(
                in memory, because the hosting environment may get deactivated between
                requests, and any in-memory context would be lost.
         """
-        super().__init__(prefix=prefix, options=options, store=store, **kwargs)
+        # Failed conversation turns must not enter chat history. Wrap every store
+        # (including the default in-memory provider) so the orchestrator's
+        # terminal create_response cannot replay poison input on the next turn.
+        history_store: ResponseProviderProtocol = (
+            InMemoryResponseProvider() if store is None else store
+        )
+        super().__init__(
+            prefix=prefix,
+            options=options,
+            store=_OmitFailedConversationInputProvider(history_store),
+            **kwargs,
+        )
 
         for provider in getattr(agent, "context_providers", []):
             if isinstance(provider, HistoryProvider) and provider.load_messages:

python/packages/foundry_hosting/tests/test_responses.py

@@ -62,6 +62,8 @@
 from agent_framework_foundry_hosting._responses import (
     CONSENT_ERROR_CODE,
     ConsentError,
+    _OmitFailedConversationInputProvider,  # pyright: ignore[reportPrivateUsage]
+    _is_failed_stored_response,  # pyright: ignore[reportPrivateUsage]
     _item_to_message,  # pyright: ignore[reportPrivateUsage]
     _output_item_to_message,  # pyright: ignore[reportPrivateUsage]
     consent_url_from_error,
@@ -639,6 +641,121 @@ def failing_run(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdat
         assert stored is not None
         assert stored.state["before_failure"] == "saved"
 
+    async def test_failed_conversation_input_is_not_in_subsequent_history(self) -> None:
+        """Failed conversation input must not be replayed on the next turn.
+
+        The agentserver store, not the MAF session, is what #7630 poisons:
+        a failed request still persisted input items onto the conversation.
+        """
+        recorded_messages: list[Sequence[Message]] = []
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("Hello!")])])
+        )
+        original_run = agent.run.side_effect
+
+        def run_dispatch(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
+            recorded_messages.append(cast(Sequence[Message], kwargs.get("messages") or []))
+            if len(recorded_messages) == 1:
+                return ResponseStream(
+                    _raising_updates("No tool call found for function call output with call_id call_12345abc."),
+                    finalizer=AgentResponse.from_updates,
+                )
+            return original_run(*args, **kwargs)
+
+        agent.run = MagicMock(side_effect=run_dispatch)
+        response_store = InMemoryResponseProvider()
+        server = _make_server(agent, response_store=response_store)
+
+        failed = await _post_json(
+            server,
+            {
+                "model": "test-model",
+                "conversation": "conv-failed",
+                "input": [
+                    {"role": "user", "content": "Hello, how are you?"},
+                    {
+                        "type": "function_call_output",
+                        "call_id": "call_12345abc",
+                        "output": "example function call output",
+                    },
+                ],
+            },
+        )
+        recovered = await _post(server, input_text="Hello, how are you?", conversation_id="conv-failed")
+
+        assert failed.json()["status"] == "failed"
+        assert recovered.json()["status"] == "completed"
+        assert len(recorded_messages) == 2
+        recovered_blob = json.dumps(
+            [
+                {
+                    "role": str(message.role),
+                    "contents": [getattr(content, "type", None) for content in message.contents],
+                    "text": [
+                        getattr(content, "text", None)
+                        for content in message.contents
+                        if getattr(content, "text", None)
+                    ],
+                    "call_ids": [
+                        getattr(content, "call_id", None)
+                        for content in message.contents
+                        if getattr(content, "call_id", None)
+                    ],
+                }
+                for message in recorded_messages[1]
+            ]
+        )
+        assert "call_12345abc" not in recovered_blob
+        assert "example function call output" not in recovered_blob
+        history_ids = await response_store.get_history_item_ids(None, "conv-failed", 100)
+        history_items = await response_store.get_items(history_ids)
+        history_blob = json.dumps(history_items)
+        assert "call_12345abc" not in history_blob
+
+    async def test_omit_failed_conversation_input_provider_drops_failed_input(self) -> None:
+        inner = InMemoryResponseProvider()
+        store = _OmitFailedConversationInputProvider(inner)
+        poison_item: dict[str, Any] = {
+            "id": "item_poison",
+            "type": "function_call_output",
+            "call_id": "call_12345abc",
+            "output": "example function call output",
+            "status": "completed",
+        }
+        ok_item: dict[str, Any] = {
+            "id": "item_ok",
+            "type": "message",
+            "role": "user",
+            "content": [{"type": "input_text", "text": "Hello, how are you?"}],
+            "status": "completed",
+        }
+
+        await store.create_response(
+            {"id": "resp_failed", "status": "failed", "conversation": "conv-1", "output": []},
+            [poison_item],
+            None,
+        )
+        await store.create_response(
+            {"id": "resp_ok", "status": "completed", "conversation": "conv-1", "output": []},
+            [ok_item],
+            None,
+        )
+
+        in_progress: dict[str, Any] = {
+            "id": "resp_stream_fail",
+            "status": "in_progress",
+            "conversation": "conv-1",
+            "output": [],
+        }
+        await store.create_response(in_progress, [poison_item], None)
+        await store.update_response({**in_progress, "status": "failed"})
+
+        history_ids = await store.get_history_item_ids(None, "conv-1", 100)
+        assert "item_poison" not in history_ids
+        assert "item_ok" in history_ids
+        assert _is_failed_stored_response({"status": "failed", "conversation": "conv-1"})
+        assert not _is_failed_stored_response({"status": "completed", "conversation": "conv-1"})
+
     async def test_run_save_failure_emits_failed_response(self) -> None:
         store = _FailingSessionStore()
         agent = _make_agent()