From f35601a79de640c77b05c27ec3c2d1384e589b08 Mon Sep 17 00:00:00 2001
From: Shikhar Goel <223222024+sgoel2be24-cyber@users.noreply.github.com>
Date: Sat, 22 Aug 2026 17:29:00 +0530
Subject: [PATCH 1/5] fix(python): omit failed Foundry turns from conversation
 history

---
 .../_responses.py                             | 118 ++++++++++++
 .../foundry_hosting/tests/test_responses.py   | 169 ++++++++++++++++++
 2 files changed, 287 insertions(+)

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 682103a24d..64f2d3ca63 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -372,6 +372,65 @@ 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 cast(Mapping[str, Any], 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.
+
+    For synchronous requests, the host knows the terminal status before the
+    provider sees the initial create. Failed responses therefore omit their
+    input items in the same operation that creates the response envelope.
+    Existing responses are updated in place: the provider protocol does not
+    expose an atomic operation for changing an envelope and its input references
+    together.
+    """
+
+    def __init__(self, inner: ResponseProviderProtocol, failed_response_ids: set[str]) -> None:
+        """Wrap ``inner`` so failed turns persist without input items."""
+        self._inner = inner
+        self._failed_response_ids = failed_response_ids
+
+    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."""
+        response_id = _response_field(response, "id")
+        known_failed = response_id is not None and str(response_id) in self._failed_response_ids
+        if _is_failed_stored_response(response) or known_failed:
+            input_items = None
+        await self._inner.create_response(response, input_items, history_item_ids, context=context)
+        if known_failed:
+            self._failed_response_ids.discard(str(response_id))
+
+    async def update_response(self, response: Any, *, context: Any = None) -> None:
+        """Update ``response`` without replacing the existing store entry."""
+        await self._inner.update_response(response, 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."""
@@ -502,6 +561,22 @@ def __init__(
         # No caller-owned agent state is mutated until all validation and base-host construction succeed.
         super().__init__(prefix=prefix, options=options, store=store, **kwargs)
 
+        # Let the base host resolve its hosted/local default and validate any
+        # explicitly supplied store before wrapping the resolved provider.
+        self._failed_sync_response_ids: set[str] = set()
+        orchestrator = self._orchestrator
+        if orchestrator is None:
+            raise RuntimeError("Responses host did not initialize its orchestrator.")
+        history_store = orchestrator._provider
+        wrapped_history_store = _OmitFailedConversationInputProvider(
+            history_store,
+            self._failed_sync_response_ids,
+        )
+        wrapped_provider = cast(ResponseProviderProtocol, wrapped_history_store)
+        orchestrator._provider = wrapped_provider
+        orchestrator._resilient_orchestrator._provider = wrapped_provider
+        self._endpoint._provider = wrapped_provider  # pyright: ignore[reportPrivateUsage]
+
         self._uses_agent_server_history = uses_agent_server_history
         self._client_stores_by_default = client_stores_by_default
         self._is_workflow_agent = is_workflow_agent
@@ -589,6 +664,49 @@ async def _handle_response(
         cancellation_signal: asyncio.Event,
     ) -> AsyncIterable[ResponseStreamEvent | ResponseCheckpointEvent]:
         """Handle the creation of a response."""
+        events = self._handle_response_events(request, context, cancellation_signal)
+        if not self._is_workflow_agent and request.get("stream") is not True and request.get("background") is not True:
+            events = self._buffer_sync_response_events(
+                events,
+                context.response_id,
+                store=request.get("store") is not False,
+            )
+        async with aclosing(events):
+            async for event in events:
+                yield event
+
+    async def _buffer_sync_response_events(
+        self,
+        events: AsyncIterable[ResponseStreamEvent | ResponseCheckpointEvent],
+        response_id: str,
+        *,
+        store: bool,
+    ) -> AsyncGenerator[ResponseStreamEvent | ResponseCheckpointEvent]:
+        """Know a synchronous turn's terminal status before its initial store write."""
+        buffered: list[ResponseStreamEvent | ResponseCheckpointEvent] = []
+        handler_error: Exception | None = None
+        try:
+            async for event in events:
+                buffered.append(event)
+        except Exception as ex:
+            handler_error = ex
+        failed = handler_error is not None or any(
+            _response_field(event, "type") == "response.failed" for event in buffered
+        )
+        if store and failed:
+            self._failed_sync_response_ids.add(response_id)
+        for event in buffered:
+            yield event
+        if handler_error is not None:
+            raise handler_error
+
+    async def _handle_response_events(
+        self,
+        request: CreateResponse,
+        context: ResponseContext,
+        cancellation_signal: asyncio.Event,
+    ) -> AsyncGenerator[ResponseStreamEvent | ResponseCheckpointEvent]:
+        """Produce response events for workflow and non-workflow agents."""
         # Common per-request setup shared by the workflow and non-workflow paths:
         # create the response stream and the streaming output-item tracker, emit
         # the opening lifecycle events, and convert any exception raised while
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 4d1b6c6497..c5c8611580 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -70,8 +70,10 @@
 from agent_framework_foundry_hosting._responses import (
     CONSENT_ERROR_CODE,
     ConsentError,
+    _is_failed_stored_response,  # pyright: ignore[reportPrivateUsage]
     _item_to_message,  # pyright: ignore[reportPrivateUsage]
     _json_safe_to_str,  # pyright: ignore[reportPrivateUsage]
+    _OmitFailedConversationInputProvider,  # pyright: ignore[reportPrivateUsage]
     _output_item_to_message,  # pyright: ignore[reportPrivateUsage]
     _OutputItemTracker,  # pyright: ignore[reportPrivateUsage]
     _stringify_mcp_output,  # pyright: ignore[reportPrivateUsage]
@@ -1293,6 +1295,173 @@ 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_terminal_failed_input(self) -> None:
+        inner = InMemoryResponseProvider()
+        store = _OmitFailedConversationInputProvider(inner, set())
+        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,
+        )
+
+        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_omit_failed_conversation_input_provider_updates_existing_response_in_place(self) -> None:
+        inner = InMemoryResponseProvider()
+        store = _OmitFailedConversationInputProvider(inner, set())
+        poison_item: dict[str, Any] = {
+            "id": "item_poison",
+            "type": "function_call_output",
+            "call_id": "call_12345abc",
+            "output": "example function call output",
+            "status": "completed",
+        }
+        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"})
+
+        persisted = await inner.get_response("resp_stream_fail")
+        input_items = await inner.get_input_items("resp_stream_fail", ascending=True)
+        assert persisted["status"] == "failed"
+        assert [item["id"] for item in input_items] == ["item_poison"]
+
+    async def test_omit_failed_conversation_input_provider_drops_known_failed_initial_input(self) -> None:
+        inner = InMemoryResponseProvider()
+        failed_response_ids = {"resp_failed"}
+        store = _OmitFailedConversationInputProvider(inner, failed_response_ids)
+        poison_item: dict[str, Any] = {
+            "id": "item_poison",
+            "type": "function_call_output",
+            "call_id": "call_12345abc",
+            "output": "example function call output",
+            "status": "completed",
+        }
+
+        await store.create_response(
+            {"id": "resp_failed", "status": "in_progress", "conversation": "conv-1", "output": []},
+            [poison_item],
+            None,
+        )
+
+        assert await inner.get_input_items("resp_failed") == []
+        assert failed_response_ids == set()
+
+    def test_default_response_store_is_resolved_before_wrapping(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+    ) -> None:
+        monkeypatch.setenv("AGENTSERVER_STATE_ROOT", str(tmp_path))
+
+        server = ResponsesHostServer(_make_agent())
+        wrapped = server._endpoint._provider  # pyright: ignore[reportPrivateUsage]
+
+        assert isinstance(wrapped, _OmitFailedConversationInputProvider)
+        assert type(wrapped._inner).__name__ == "FileResponseStore"  # pyright: ignore[reportPrivateUsage]
+
+    def test_explicit_volatile_store_still_fails_resilient_background_guard(self) -> None:
+        options = ResponsesServerOptions(resilient_background=True)
+
+        with pytest.raises(ValueError, match="resilient_background=True"):
+            ResponsesHostServer(_make_agent(), store=InMemoryResponseProvider(), options=options)
+
     async def test_run_save_failure_emits_failed_response(self) -> None:
         store = _FailingSessionStore()
         agent = _make_agent()

From c4c3399e8fd2685327d8a489f09770d102def3ab Mon Sep 17 00:00:00 2001
From: Shikhar Goel <223222024+sgoel2be24-cyber@users.noreply.github.com>
Date: Wed, 26 Aug 2026 17:51:31 +0530
Subject: [PATCH 2/5] fix(python): tighten failed-response provider types

---
 .../_responses.py                             | 42 ++++----
 .../foundry_hosting/tests/test_responses.py   | 99 +++++++++++--------
 2 files changed, 80 insertions(+), 61 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 64f2d3ca63..910e776b9c 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -9,7 +9,7 @@
 import logging
 import os
 import re
-from collections.abc import AsyncGenerator, AsyncIterable, AsyncIterator, Generator, Mapping, Sequence
+from collections.abc import AsyncGenerator, AsyncIterable, AsyncIterator, Generator, Iterable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, aclosing, suppress
 from dataclasses import asdict, dataclass, is_dataclass
 from typing import Generic, Literal, TypeGuard, TypeVar, cast
@@ -35,7 +35,9 @@
 from agent_framework.exceptions import AgentFrameworkException
 from azure.ai.agentserver.core import get_request_context
 from azure.ai.agentserver.responses import (
+    PlatformContext,
     ResponseContext,
+    ResponseObject,
     ResponseProviderProtocol,
     ResponsesServerOptions,
 )
@@ -372,16 +374,9 @@ 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 cast(Mapping[str, Any], response).get(name)
-    return getattr(response, name, None)
-
-
-def _is_failed_stored_response(response: Any) -> bool:
+def _is_failed_stored_response(response: ResponseObject) -> bool:
     """Return whether a persisted response envelope is a failed turn."""
-    return _response_field(response, "status") == "failed"
+    return response.get("status") == "failed"
 
 
 class _OmitFailedConversationInputProvider:
@@ -389,8 +384,7 @@ class _OmitFailedConversationInputProvider:
 
     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.
+    the next turn. Azure OpenAI does not keep failed input on the conversation.
 
     For synchronous requests, the host knows the terminal status before the
     provider sees the initial create. Failed responses therefore omit their
@@ -407,22 +401,27 @@ def __init__(self, inner: ResponseProviderProtocol, failed_response_ids: set[str
 
     async def create_response(
         self,
-        response: Any,
-        input_items: Any,
-        history_item_ids: Any,
+        response: ResponseObject,
+        input_items: Iterable[OutputItem] | None,
+        history_item_ids: Iterable[str] | None,
         *,
-        context: Any = None,
+        context: PlatformContext | None = None,
     ) -> None:
         """Persist ``response``, dropping input items when the turn failed."""
-        response_id = _response_field(response, "id")
-        known_failed = response_id is not None and str(response_id) in self._failed_response_ids
+        response_id = response["id"]
+        known_failed = response_id in self._failed_response_ids
         if _is_failed_stored_response(response) or known_failed:
             input_items = None
         await self._inner.create_response(response, input_items, history_item_ids, context=context)
         if known_failed:
-            self._failed_response_ids.discard(str(response_id))
+            self._failed_response_ids.discard(response_id)
 
-    async def update_response(self, response: Any, *, context: Any = None) -> None:
+    async def update_response(
+        self,
+        response: ResponseObject,
+        *,
+        context: PlatformContext | None = None,
+    ) -> None:
         """Update ``response`` without replacing the existing store entry."""
         await self._inner.update_response(response, context=context)
 
@@ -691,7 +690,8 @@ async def _buffer_sync_response_events(
         except Exception as ex:
             handler_error = ex
         failed = handler_error is not None or any(
-            _response_field(event, "type") == "response.failed" for event in buffered
+            isinstance(event, Mapping) and cast(Mapping[str, object], event).get("type") == "response.failed"
+            for event in buffered
         )
         if store and failed:
             self._failed_sync_response_ids.add(response_id)
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index c5c8611580..b6a87c67ff 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -56,6 +56,7 @@
     InMemoryResponseProvider,
     ResponseContext,
     ResponseExitForRecovery,
+    ResponseObject,
     ResponsesServerOptions,
 )
 from azure.ai.agentserver.responses.aio import ResponseEventStream
@@ -1365,28 +1366,34 @@ def run_dispatch(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdat
     async def test_omit_failed_conversation_input_provider_drops_terminal_failed_input(self) -> None:
         inner = InMemoryResponseProvider()
         store = _OmitFailedConversationInputProvider(inner, set())
-        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",
-        }
+        poison_item = cast(
+            OutputItem,
+            {
+                "id": "item_poison",
+                "type": "function_call_output",
+                "call_id": "call_12345abc",
+                "output": "example function call output",
+                "status": "completed",
+            },
+        )
+        ok_item = cast(
+            OutputItem,
+            {
+                "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": []},
+            cast(ResponseObject, {"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": []},
+            cast(ResponseObject, {"id": "resp_ok", "status": "completed", "conversation": "conv-1", "output": []}),
             [ok_item],
             None,
         )
@@ -1394,28 +1401,34 @@ async def test_omit_failed_conversation_input_provider_drops_terminal_failed_inp
         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"})
+        assert _is_failed_stored_response(cast(ResponseObject, {"id": "resp_failed", "status": "failed"}))
+        assert not _is_failed_stored_response(cast(ResponseObject, {"id": "resp_ok", "status": "completed"}))
 
     async def test_omit_failed_conversation_input_provider_updates_existing_response_in_place(self) -> None:
         inner = InMemoryResponseProvider()
         store = _OmitFailedConversationInputProvider(inner, set())
-        poison_item: dict[str, Any] = {
-            "id": "item_poison",
-            "type": "function_call_output",
-            "call_id": "call_12345abc",
-            "output": "example function call output",
-            "status": "completed",
-        }
-        in_progress: dict[str, Any] = {
-            "id": "resp_stream_fail",
-            "status": "in_progress",
-            "conversation": "conv-1",
-            "output": [],
-        }
+        poison_item = cast(
+            OutputItem,
+            {
+                "id": "item_poison",
+                "type": "function_call_output",
+                "call_id": "call_12345abc",
+                "output": "example function call output",
+                "status": "completed",
+            },
+        )
+        in_progress = cast(
+            ResponseObject,
+            {
+                "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"})
+        await store.update_response(cast(ResponseObject, {**in_progress, "status": "failed"}))
 
         persisted = await inner.get_response("resp_stream_fail")
         input_items = await inner.get_input_items("resp_stream_fail", ascending=True)
@@ -1426,16 +1439,22 @@ async def test_omit_failed_conversation_input_provider_drops_known_failed_initia
         inner = InMemoryResponseProvider()
         failed_response_ids = {"resp_failed"}
         store = _OmitFailedConversationInputProvider(inner, failed_response_ids)
-        poison_item: dict[str, Any] = {
-            "id": "item_poison",
-            "type": "function_call_output",
-            "call_id": "call_12345abc",
-            "output": "example function call output",
-            "status": "completed",
-        }
+        poison_item = cast(
+            OutputItem,
+            {
+                "id": "item_poison",
+                "type": "function_call_output",
+                "call_id": "call_12345abc",
+                "output": "example function call output",
+                "status": "completed",
+            },
+        )
 
         await store.create_response(
-            {"id": "resp_failed", "status": "in_progress", "conversation": "conv-1", "output": []},
+            cast(
+                ResponseObject,
+                {"id": "resp_failed", "status": "in_progress", "conversation": "conv-1", "output": []},
+            ),
             [poison_item],
             None,
         )

From 4719f0bcd8be0d5673195bd76de9081181aa2d81 Mon Sep 17 00:00:00 2001
From: Shikhar Goel <223222024+sgoel2be24-cyber@users.noreply.github.com>
Date: Thu, 3 Sep 2026 14:04:34 +0530
Subject: [PATCH 3/5] fix(python): respect Foundry hosting history ownership

---
 .../_responses.py                             | 34 ++++++++++++-------
 .../foundry_hosting/tests/test_responses.py   | 27 ++++++++++++---
 2 files changed, 44 insertions(+), 17 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 910e776b9c..a0a29c7a9e 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -563,18 +563,21 @@ def __init__(
         # Let the base host resolve its hosted/local default and validate any
         # explicitly supplied store before wrapping the resolved provider.
         self._failed_sync_response_ids: set[str] = set()
-        orchestrator = self._orchestrator
-        if orchestrator is None:
-            raise RuntimeError("Responses host did not initialize its orchestrator.")
-        history_store = orchestrator._provider
-        wrapped_history_store = _OmitFailedConversationInputProvider(
-            history_store,
-            self._failed_sync_response_ids,
-        )
-        wrapped_provider = cast(ResponseProviderProtocol, wrapped_history_store)
-        orchestrator._provider = wrapped_provider
-        orchestrator._resilient_orchestrator._provider = wrapped_provider
-        self._endpoint._provider = wrapped_provider  # pyright: ignore[reportPrivateUsage]
+        if uses_agent_server_history and not is_workflow_agent:
+            # Agent-owned history does not replay this transcript. Preserve its
+            # protocol-level response storage without filtering failed inputs.
+            orchestrator = self._orchestrator
+            if orchestrator is None:
+                raise RuntimeError("Responses host did not initialize its orchestrator.")
+            history_store = orchestrator._provider
+            wrapped_history_store = _OmitFailedConversationInputProvider(
+                history_store,
+                self._failed_sync_response_ids,
+            )
+            wrapped_provider = cast(ResponseProviderProtocol, wrapped_history_store)
+            orchestrator._provider = wrapped_provider
+            orchestrator._resilient_orchestrator._provider = wrapped_provider
+            self._endpoint._provider = wrapped_provider  # pyright: ignore[reportPrivateUsage]
 
         self._uses_agent_server_history = uses_agent_server_history
         self._client_stores_by_default = client_stores_by_default
@@ -664,7 +667,12 @@ async def _handle_response(
     ) -> AsyncIterable[ResponseStreamEvent | ResponseCheckpointEvent]:
         """Handle the creation of a response."""
         events = self._handle_response_events(request, context, cancellation_signal)
-        if not self._is_workflow_agent and request.get("stream") is not True and request.get("background") is not True:
+        if (
+            self._uses_agent_server_history
+            and not self._is_workflow_agent
+            and request.get("stream") is not True
+            and request.get("background") is not True
+        ):
             events = self._buffer_sync_response_events(
                 events,
                 context.response_id,
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index b6a87c67ff..35b40d5bbd 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -1296,7 +1296,10 @@ 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:
+    @pytest.mark.parametrize("history_source", ["agent_server", "agent"])
+    async def test_failed_conversation_input_is_not_in_subsequent_history(
+        self, history_source: Literal["agent_server", "agent"]
+    ) -> None:
         """Failed conversation input must not be replayed on the next turn.
 
         The agentserver store, not the MAF session, is what #7630 poisons:
@@ -1319,7 +1322,7 @@ def run_dispatch(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdat
 
         agent.run = MagicMock(side_effect=run_dispatch)
         response_store = InMemoryResponseProvider()
-        server = _make_server(agent, response_store=response_store)
+        server = _make_server(agent, response_store=response_store, history_source=history_source)
 
         failed = await _post_json(
             server,
@@ -1361,7 +1364,23 @@ def run_dispatch(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdat
         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
+        if history_source == "agent_server":
+            assert "call_12345abc" not in history_blob
+        else:
+            # AgentServer still retains the protocol transcript in agent-history
+            # mode, but hosting does not replay it into the model.
+            assert "call_12345abc" in history_blob
+
+    async def test_agent_history_does_not_buffer_or_wrap_response_storage(self) -> None:
+        agent = _make_agent(response=AgentResponse(messages=[Message("assistant", ["done"])]))
+        response_store = InMemoryResponseProvider()
+        server = _make_server(agent, response_store=response_store, history_source="agent")
+
+        assert server._endpoint._provider is response_store  # pyright: ignore[reportPrivateUsage]
+        with patch.object(server, "_buffer_sync_response_events", side_effect=AssertionError("must not buffer")):
+            response = await _post(server, input_text="first", stream=False)
+
+        assert response.json()["status"] == "completed"
 
     async def test_omit_failed_conversation_input_provider_drops_terminal_failed_input(self) -> None:
         inner = InMemoryResponseProvider()
@@ -1479,7 +1498,7 @@ def test_explicit_volatile_store_still_fails_resilient_background_guard(self) ->
         options = ResponsesServerOptions(resilient_background=True)
 
         with pytest.raises(ValueError, match="resilient_background=True"):
-            ResponsesHostServer(_make_agent(), store=InMemoryResponseProvider(), options=options)
+            ResponsesHostServer(_build_text_workflow_agent("done"), store=InMemoryResponseProvider(), options=options)
 
     async def test_run_save_failure_emits_failed_response(self) -> None:
         store = _FailingSessionStore()

From aedebb27b98b8fbe986e8d0ab80a152e9b364ac7 Mon Sep 17 00:00:00 2001
From: sgoel2be24-cyber <223222024+sgoel2be24-cyber@users.noreply.github.com>
Date: Fri, 4 Sep 2026 11:55:59 +0530
Subject: [PATCH 4/5] fix(python): isolate failed response input filtering

Signed-off-by: sgoel2be24-cyber <223222024+sgoel2be24-cyber@users.noreply.github.com>
---
 .../_responses.py                             | 39 ++++++++-----
 .../foundry_hosting/tests/test_responses.py   | 58 ++++++++++++++-----
 2 files changed, 69 insertions(+), 28 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 a0a29c7a9e..1051b76b71 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -11,6 +11,7 @@
 import re
 from collections.abc import AsyncGenerator, AsyncIterable, AsyncIterator, Generator, Iterable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, aclosing, suppress
+from contextvars import ContextVar
 from dataclasses import asdict, dataclass, is_dataclass
 from typing import Generic, Literal, TypeGuard, TypeVar, cast
 from urllib.parse import urlparse
@@ -394,10 +395,10 @@ class _OmitFailedConversationInputProvider:
     together.
     """
 
-    def __init__(self, inner: ResponseProviderProtocol, failed_response_ids: set[str]) -> None:
+    def __init__(self, inner: ResponseProviderProtocol, failed_response_id: ContextVar[str | None]) -> None:
         """Wrap ``inner`` so failed turns persist without input items."""
         self._inner = inner
-        self._failed_response_ids = failed_response_ids
+        self._failed_response_id = failed_response_id
 
     async def create_response(
         self,
@@ -409,12 +410,13 @@ async def create_response(
     ) -> None:
         """Persist ``response``, dropping input items when the turn failed."""
         response_id = response["id"]
-        known_failed = response_id in self._failed_response_ids
-        if _is_failed_stored_response(response) or known_failed:
+        known_failed = self._failed_response_id.get() == response_id
+        belongs_to_history = (
+            response.get("conversation") is not None or response.get("previous_response_id") is not None
+        )
+        if belongs_to_history and (_is_failed_stored_response(response) or known_failed):
             input_items = None
         await self._inner.create_response(response, input_items, history_item_ids, context=context)
-        if known_failed:
-            self._failed_response_ids.discard(response_id)
 
     async def update_response(
         self,
@@ -562,7 +564,10 @@ def __init__(
 
         # Let the base host resolve its hosted/local default and validate any
         # explicitly supplied store before wrapping the resolved provider.
-        self._failed_sync_response_ids: set[str] = set()
+        self._failed_sync_response_id: ContextVar[str | None] = ContextVar(
+            f"failed_sync_response_id_{id(self)}",
+            default=None,
+        )
         if uses_agent_server_history and not is_workflow_agent:
             # Agent-owned history does not replay this transcript. Preserve its
             # protocol-level response storage without filtering failed inputs.
@@ -572,7 +577,7 @@ def __init__(
             history_store = orchestrator._provider
             wrapped_history_store = _OmitFailedConversationInputProvider(
                 history_store,
-                self._failed_sync_response_ids,
+                self._failed_sync_response_id,
             )
             wrapped_provider = cast(ResponseProviderProtocol, wrapped_history_store)
             orchestrator._provider = wrapped_provider
@@ -677,6 +682,8 @@ async def _handle_response(
                 events,
                 context.response_id,
                 store=request.get("store") is not False,
+                belongs_to_history=context.conversation_id is not None
+                or request.get("previous_response_id") is not None,
             )
         async with aclosing(events):
             async for event in events:
@@ -688,6 +695,7 @@ async def _buffer_sync_response_events(
         response_id: str,
         *,
         store: bool,
+        belongs_to_history: bool,
     ) -> AsyncGenerator[ResponseStreamEvent | ResponseCheckpointEvent]:
         """Know a synchronous turn's terminal status before its initial store write."""
         buffered: list[ResponseStreamEvent | ResponseCheckpointEvent] = []
@@ -701,12 +709,15 @@ async def _buffer_sync_response_events(
             isinstance(event, Mapping) and cast(Mapping[str, object], event).get("type") == "response.failed"
             for event in buffered
         )
-        if store and failed:
-            self._failed_sync_response_ids.add(response_id)
-        for event in buffered:
-            yield event
-        if handler_error is not None:
-            raise handler_error
+        token = self._failed_sync_response_id.set(response_id) if store and belongs_to_history and failed else None
+        try:
+            for event in buffered:
+                yield event
+            if handler_error is not None:
+                raise handler_error
+        finally:
+            if token is not None:
+                self._failed_sync_response_id.reset(token)
 
     async def _handle_response_events(
         self,
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 35b40d5bbd..a32cfa6f5c 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -15,6 +15,7 @@
 import logging
 import uuid
 from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Mapping, Sequence
+from contextvars import ContextVar
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Literal, cast, overload
@@ -60,7 +61,7 @@
     ResponsesServerOptions,
 )
 from azure.ai.agentserver.responses.aio import ResponseEventStream
-from azure.ai.agentserver.responses.models import CreateResponse, Item, OutputItem
+from azure.ai.agentserver.responses.models import CreateResponse, Item, OutputItem, ResponseStreamEvent
 from azure.ai.agentserver.responses.streaming._checkpoint import ResponseCheckpointEvent
 from mcp import McpError
 from mcp.types import ErrorData
@@ -1384,7 +1385,7 @@ async def test_agent_history_does_not_buffer_or_wrap_response_storage(self) -> N
 
     async def test_omit_failed_conversation_input_provider_drops_terminal_failed_input(self) -> None:
         inner = InMemoryResponseProvider()
-        store = _OmitFailedConversationInputProvider(inner, set())
+        store = _OmitFailedConversationInputProvider(inner, ContextVar("failed_response_id", default=None))
         poison_item = cast(
             OutputItem,
             {
@@ -1416,16 +1417,22 @@ async def test_omit_failed_conversation_input_provider_drops_terminal_failed_inp
             [ok_item],
             None,
         )
+        await store.create_response(
+            cast(ResponseObject, {"id": "resp_standalone", "status": "failed", "output": []}),
+            [poison_item],
+            None,
+        )
 
         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 [item["id"] for item in await store.get_input_items("resp_standalone")] == ["item_poison"]
         assert _is_failed_stored_response(cast(ResponseObject, {"id": "resp_failed", "status": "failed"}))
         assert not _is_failed_stored_response(cast(ResponseObject, {"id": "resp_ok", "status": "completed"}))
 
     async def test_omit_failed_conversation_input_provider_updates_existing_response_in_place(self) -> None:
         inner = InMemoryResponseProvider()
-        store = _OmitFailedConversationInputProvider(inner, set())
+        store = _OmitFailedConversationInputProvider(inner, ContextVar("failed_response_id", default=None))
         poison_item = cast(
             OutputItem,
             {
@@ -1456,8 +1463,8 @@ async def test_omit_failed_conversation_input_provider_updates_existing_response
 
     async def test_omit_failed_conversation_input_provider_drops_known_failed_initial_input(self) -> None:
         inner = InMemoryResponseProvider()
-        failed_response_ids = {"resp_failed"}
-        store = _OmitFailedConversationInputProvider(inner, failed_response_ids)
+        failed_response_id = ContextVar[str | None]("failed_response_id", default=None)
+        store = _OmitFailedConversationInputProvider(inner, failed_response_id)
         poison_item = cast(
             OutputItem,
             {
@@ -1469,17 +1476,40 @@ async def test_omit_failed_conversation_input_provider_drops_known_failed_initia
             },
         )
 
-        await store.create_response(
-            cast(
-                ResponseObject,
-                {"id": "resp_failed", "status": "in_progress", "conversation": "conv-1", "output": []},
-            ),
-            [poison_item],
-            None,
-        )
+        token = failed_response_id.set("resp_failed")
+        try:
+            await store.create_response(
+                cast(
+                    ResponseObject,
+                    {"id": "resp_failed", "status": "in_progress", "conversation": "conv-1", "output": []},
+                ),
+                [poison_item],
+                None,
+            )
+        finally:
+            failed_response_id.reset(token)
 
         assert await inner.get_input_items("resp_failed") == []
-        assert failed_response_ids == set()
+        assert failed_response_id.get() is None
+
+    async def test_failed_response_marker_is_cleared_when_persistence_stops_consumption(self) -> None:
+        server = _make_server(_make_agent())
+
+        async def failed_events() -> AsyncIterator[ResponseStreamEvent]:
+            yield cast(ResponseStreamEvent, {"type": "response.failed"})
+
+        buffered = server._buffer_sync_response_events(  # pyright: ignore[reportPrivateUsage]
+            failed_events(),
+            "resp_failed",
+            store=True,
+            belongs_to_history=True,
+        )
+
+        await anext(buffered)
+        assert server._failed_sync_response_id.get() == "resp_failed"  # pyright: ignore[reportPrivateUsage]
+
+        await buffered.aclose()
+        assert server._failed_sync_response_id.get() is None  # pyright: ignore[reportPrivateUsage]
 
     def test_default_response_store_is_resolved_before_wrapping(
         self,

From 5b181e848e7145d44c6a6c03f67b8ee7da627d03 Mon Sep 17 00:00:00 2001
From: Shikhar Goel <223222024+sgoel2be24-cyber@users.noreply.github.com>
Date: Fri, 4 Sep 2026 14:19:59 +0530
Subject: [PATCH 5/5] fix(python): filter failed standalone response history

Signed-off-by: Shikhar Goel <223222024+sgoel2be24-cyber@users.noreply.github.com>
---
 .../_responses.py                             | 36 ++++++++++++++
 .../foundry_hosting/tests/test_responses.py   | 47 +++++++++++++++++++
 2 files changed, 83 insertions(+)

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 1051b76b71..cb46ff5093 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -427,6 +427,42 @@ async def update_response(
         """Update ``response`` without replacing the existing store entry."""
         await self._inner.update_response(response, context=context)
 
+    async def get_history_item_ids(
+        self,
+        previous_response_id: str | None,
+        conversation_id: str | None,
+        limit: int,
+        *,
+        context: PlatformContext | None = None,
+    ) -> list[str]:
+        """Exclude a failed standalone response's inputs only when it becomes history."""
+        history_item_ids = await self._inner.get_history_item_ids(
+            previous_response_id,
+            conversation_id,
+            limit,
+            context=context,
+        )
+        if previous_response_id is None or not history_item_ids:
+            return history_item_ids
+
+        previous_response = await self._inner.get_response(previous_response_id, context=context)
+        is_failed_standalone = (
+            _is_failed_stored_response(previous_response)
+            and previous_response.get("conversation") is None
+            and previous_response.get("previous_response_id") is None
+        )
+        if not is_failed_standalone:
+            return history_item_ids
+
+        input_items = await self._inner.get_input_items(
+            previous_response_id,
+            limit=max(1, min(limit, 100)),
+            ascending=False,
+            context=context,
+        )
+        failed_input_ids = {item_id for item in input_items if isinstance((item_id := item.get("id")), str)}
+        return [item_id for item_id in history_item_ids if item_id not in failed_input_ids]
+
     def __getattr__(self, name: str) -> Any:
         """Forward remaining provider methods to the wrapped store."""
         return getattr(self._inner, name)
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index a32cfa6f5c..36b115a6f6 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -1427,9 +1427,56 @@ async def test_omit_failed_conversation_input_provider_drops_terminal_failed_inp
         assert "item_poison" not in history_ids
         assert "item_ok" in history_ids
         assert [item["id"] for item in await store.get_input_items("resp_standalone")] == ["item_poison"]
+        previous_history_ids = await store.get_history_item_ids("resp_standalone", None, 100)
+        assert "item_poison" not in previous_history_ids
         assert _is_failed_stored_response(cast(ResponseObject, {"id": "resp_failed", "status": "failed"}))
         assert not _is_failed_stored_response(cast(ResponseObject, {"id": "resp_ok", "status": "completed"}))
 
+    async def test_failed_standalone_input_is_not_replayed_by_previous_response_id(self) -> None:
+        recorded_messages: list[Sequence[Message]] = []
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("recovered")])])
+        )
+        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_standalone."),
+                    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",
+                "input": [
+                    {"role": "user", "content": "first"},
+                    {
+                        "type": "function_call_output",
+                        "call_id": "call_standalone",
+                        "output": "poison output",
+                    },
+                ],
+            },
+        )
+
+        recovered = await _post(server, input_text="second", previous_response_id=failed.json()["id"])
+
+        assert failed.json()["status"] == "failed"
+        assert recovered.json()["status"] == "completed"
+        recovered_blob = json.dumps([message.to_dict() for message in recorded_messages[1]])
+        assert "call_standalone" not in recovered_blob
+        assert "poison output" not in recovered_blob
+        stored_input_blob = json.dumps(await response_store.get_input_items(failed.json()["id"], ascending=True))
+        assert "call_standalone" in stored_input_blob
+        assert "poison output" in stored_input_blob
+
     async def test_omit_failed_conversation_input_provider_updates_existing_response_in_place(self) -> None:
         inner = InMemoryResponseProvider()
         store = _OmitFailedConversationInputProvider(inner, ContextVar("failed_response_id", default=None))