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..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: diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 5b0dfda65c..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,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()