From 9a2e1c9461d01a79641dca96158987bc26b3ad57 Mon Sep 17 00:00:00 2001 From: Shikhar Goel Date: Thu, 13 Aug 2026 01:24:41 +0530 Subject: [PATCH 1/2] fix(python): skip Foundry session persist after a failed conversation turn Saving the mutated session on failure poisons later requests that reuse the same conversation_id. Non-conversation runs still persist on failure. --- .../_responses.py | 10 +++++- .../foundry_hosting/tests/test_responses.py | 33 +++++++++++++++++++ 2 files changed, 42 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 6ff4550f7e..e32144b918 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -406,6 +406,10 @@ async def _handle_inner_agent( request_failure: Exception | None = None save_failure: Exception | None = None request_interrupted = False + # Conversation mode must not persist a failed turn: Azure OpenAI leaves + # failed input off the conversation, but saving here poisons every later + # request on the same conversation_id (see microsoft/agent-framework#7630). + persist_failed_conversation = context.conversation_id is None try: if self._uses_hosted_responses_history: @@ -455,8 +459,12 @@ async def _handle_inner_agent( finally: if self._uses_hosted_responses_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) + skip_persist = ( + request_failure is not None and not persist_failed_conversation + ) try: - await session_storage.set(context.conversation_id or context.response_id, session) + if not skip_persist: + await session_storage.set(context.conversation_id or context.response_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 5b0dfda65c..189583090d 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -639,6 +639,39 @@ 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_run_does_not_persist_session(self) -> None: + """Failed turns must not poison later requests on the same conversation_id. + + Azure OpenAI leaves failed input off the conversation. Foundry hosting + previously saved the mutated session anyway (microsoft/agent-framework#7630). + """ + store = SessionStore() + agent = _make_agent() + + def failing_run(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: + del _args + session = kwargs["session"] + assert isinstance(session, AgentSession) + + return ResponseStream( + _raising_updates( + "agent failed", + before_raise=lambda: session.state.__setitem__("poison", "yes"), + ), + finalizer=AgentResponse.from_updates, + ) + + agent.run = MagicMock(side_effect=failing_run) + server = _make_server(agent, session_store=store) + + response = await _post(server, conversation_id="conv-failed") + body = response.json() + + stored = await store.get("conv-failed") + + assert body["status"] == "failed" + assert stored is None + async def test_run_save_failure_emits_failed_response(self) -> None: store = _FailingSessionStore() agent = _make_agent() From 24ebe8b230c29d2ce0242e90998ce5882ef20db9 Mon Sep 17 00:00:00 2001 From: Shikhar Goel Date: Thu, 13 Aug 2026 16:36:19 +0530 Subject: [PATCH 2/2] fix(python): omit failed Foundry turns from conversation chat history Skip persisting input items when a stored response is failed so the next request on the same conversation does not replay the bad turn. --- .../_responses.py | 86 +++++++++-- .../foundry_hosting/tests/test_responses.py | 134 ++++++++++++++---- 2 files changed, 185 insertions(+), 35 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 e32144b918..34acdaa8c0 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/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: @@ -406,10 +480,6 @@ async def _handle_inner_agent( request_failure: Exception | None = None save_failure: Exception | None = None request_interrupted = False - # Conversation mode must not persist a failed turn: Azure OpenAI leaves - # failed input off the conversation, but saving here poisons every later - # request on the same conversation_id (see microsoft/agent-framework#7630). - persist_failed_conversation = context.conversation_id is None try: if self._uses_hosted_responses_history: @@ -459,12 +529,8 @@ async def _handle_inner_agent( finally: if self._uses_hosted_responses_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) - skip_persist = ( - request_failure is not None and not persist_failed_conversation - ) try: - if not skip_persist: - await session_storage.set(context.conversation_id or context.response_id, session) + await session_storage.set(context.conversation_id or context.response_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 189583090d..6cba55b5ad 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/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,38 +641,120 @@ 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_run_does_not_persist_session(self) -> None: - """Failed turns must not poison later requests on the same conversation_id. + async def test_failed_conversation_input_is_not_in_subsequent_history(self) -> None: + """Failed conversation input must not be replayed on the next turn. - Azure OpenAI leaves failed input off the conversation. Foundry hosting - previously saved the mutated session anyway (microsoft/agent-framework#7630). + The agentserver store, not the MAF session, is what #7630 poisons: + a failed request still persisted input items onto the conversation. """ - store = SessionStore() - agent = _make_agent() + 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) - def failing_run(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: - del _args - session = kwargs["session"] - assert isinstance(session, AgentSession) + agent.run = MagicMock(side_effect=run_dispatch) + response_store = InMemoryResponseProvider() + server = _make_server(agent, response_store=response_store) - return ResponseStream( - _raising_updates( - "agent failed", - before_raise=lambda: session.state.__setitem__("poison", "yes"), - ), - finalizer=AgentResponse.from_updates, - ) - - agent.run = MagicMock(side_effect=failing_run) - server = _make_server(agent, session_store=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") - response = await _post(server, conversation_id="conv-failed") - body = response.json() + 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", + } - stored = await store.get("conv-failed") + 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, + ) - assert body["status"] == "failed" - assert stored is 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()