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..cb46ff5093 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -9,8 +9,9 @@ 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 contextvars import ContextVar from dataclasses import asdict, dataclass, is_dataclass from typing import Generic, Literal, TypeGuard, TypeVar, cast from urllib.parse import urlparse @@ -35,7 +36,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,6 +375,99 @@ def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None: # endregion Foundry Toolbox Auth integration +def _is_failed_stored_response(response: ResponseObject) -> bool: + """Return whether a persisted response envelope is a failed turn.""" + return response.get("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. 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_id: ContextVar[str | None]) -> None: + """Wrap ``inner`` so failed turns persist without input items.""" + self._inner = inner + self._failed_response_id = failed_response_id + + async def create_response( + self, + response: ResponseObject, + input_items: Iterable[OutputItem] | None, + history_item_ids: Iterable[str] | None, + *, + context: PlatformContext | None = None, + ) -> None: + """Persist ``response``, dropping input items when the turn failed.""" + response_id = response["id"] + 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) + + 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) + + 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) + + # region ResponsesHostServer class ResponsesHostServer(ResponsesAgentServerHost): """A responses server host for an agent.""" @@ -502,6 +598,28 @@ 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_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. + 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_id, + ) + 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 +707,61 @@ 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 ( + 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, + 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: + yield event + + async def _buffer_sync_response_events( + self, + events: AsyncIterable[ResponseStreamEvent | ResponseCheckpointEvent], + 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] = [] + 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( + isinstance(event, Mapping) and cast(Mapping[str, object], event).get("type") == "response.failed" + for event in buffered + ) + 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, + 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..36b115a6f6 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 @@ -56,10 +57,11 @@ InMemoryResponseProvider, ResponseContext, ResponseExitForRecovery, + ResponseObject, 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 @@ -70,8 +72,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 +1297,286 @@ def failing_run(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdat assert stored is not None assert stored.state["before_failure"] == "saved" + @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: + 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, history_source=history_source) + + 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) + 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() + store = _OmitFailedConversationInputProvider(inner, ContextVar("failed_response_id", default=None)) + 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( + cast(ResponseObject, {"id": "resp_failed", "status": "failed", "conversation": "conv-1", "output": []}), + [poison_item], + None, + ) + await store.create_response( + cast(ResponseObject, {"id": "resp_ok", "status": "completed", "conversation": "conv-1", "output": []}), + [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"] + 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)) + 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(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) + 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_id = ContextVar[str | None]("failed_response_id", default=None) + store = _OmitFailedConversationInputProvider(inner, failed_response_id) + poison_item = cast( + OutputItem, + { + "id": "item_poison", + "type": "function_call_output", + "call_id": "call_12345abc", + "output": "example function call output", + "status": "completed", + }, + ) + + 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_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, + 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(_build_text_workflow_agent("done"), store=InMemoryResponseProvider(), options=options) + async def test_run_save_failure_emits_failed_response(self) -> None: store = _FailingSessionStore() agent = _make_agent()