From 5e684a24795ad786b4f02627862a342e4c4a835b Mon Sep 17 00:00:00 2001 From: Anish Mehta Date: Thu, 17 Sep 2026 22:14:54 +0530 Subject: [PATCH 1/3] Python: mark hosted /responses incomplete on content_filter and length finish reasons The hosted /responses path iterated each AgentResponseUpdate's contents but never read its finish_reason, so a turn the model stopped early (Azure OpenAI content filter, token limit) ended with status "completed" and no trace of the cut-off. Callers could only detect a filtered turn by matching the canned refusal text. _OutputItemTracker now records truncating finish reasons from both the plain-agent and workflow update loops, and _handle_response ends the response with response.incomplete plus incomplete_details.reason ("content_filter" / "max_output_tokens") instead of response.completed, mirroring the OpenAI Responses shape. The refusal text is still delivered as an output item. Fixes #8475 Co-Authored-By: Claude Opus 5 (1M context) --- .../_responses.py | 30 ++++- .../foundry_hosting/tests/test_responses.py | 125 +++++++++++++++++- 2 files changed, 150 insertions(+), 5 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 6f7b9ebcf2..bdb37c56cb 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -65,6 +65,7 @@ OutputItem, OutputItemReasoningItem, OutputMessageContent, + ResponseIncompleteReason, ResponseStreamEvent, ResponseUsage, ResponseUsageInputTokensDetails, @@ -797,8 +798,9 @@ async def _handle_prepared_response( for event in tracker.close(): yield event - if tracker.oauth_consent_requested: - yield response_event_stream.emit_incomplete(usage=tracker.usage) + incomplete_reason = tracker.incomplete_reason + if tracker.oauth_consent_requested or incomplete_reason is not None: + yield response_event_stream.emit_incomplete(reason=incomplete_reason, usage=tracker.usage) else: yield response_event_stream.emit_completed(usage=tracker.usage) except Exception as ex: @@ -964,6 +966,7 @@ async def _handle_inner_agent( ) async with aclosing(agent_stream): async for update in agent_stream: + tracker.record_finish_reason(update.finish_reason) for content in update.contents: async for event in tracker.handle( content, message_id=update.message_id, approval_storage=approval_storage @@ -1187,6 +1190,7 @@ async def _handle_inner_workflow( ) yield response_event_stream.checkpoint() + tracker.record_finish_reason(update.finish_reason) for content in update.contents: async for event in tracker.handle( content, message_id=update.message_id, approval_storage=approval_storage @@ -1291,6 +1295,10 @@ def __init__(self, stream: ResponseEventStream) -> None: self._mcp_builder: OutputItemMcpCallBuilder | None = None self._outstanding_function_calls: dict[str, str | None] = {} self._oauth_consent_requests: set[tuple[str, str]] = set() + # Set when an agent update reports the model stopped early (content filter, token + # limit); the response then ends as ``incomplete`` instead of ``completed`` so callers + # can tell a cut-short turn from a successful one. + self._incomplete_reason: ResponseIncompleteReason | None = None for item in stream.response.get("output", []): if not isinstance(item, Mapping): continue @@ -1329,6 +1337,24 @@ def oauth_consent_requested(self) -> bool: """Return whether this response emitted an OAuth consent request.""" return bool(self._oauth_consent_requests) + @property + def incomplete_reason(self) -> ResponseIncompleteReason | None: + """Return why the turn was cut short, if any update reported a truncating finish reason.""" + return self._incomplete_reason + + def record_finish_reason(self, finish_reason: str | None) -> None: + """Note the finish reason of an agent update. + + Only finish reasons that mean the model stopped early are retained, mapped onto the + Responses ``incomplete_details.reason`` vocabulary. A content filter is kept in + preference to a token limit if both are seen during a multi-step turn, since it is the + more actionable signal for the caller. + """ + if finish_reason == "content_filter": + self._incomplete_reason = ResponseIncompleteReason.CONTENT_FILTER + elif finish_reason == "length" and self._incomplete_reason is None: + self._incomplete_reason = ResponseIncompleteReason.MAX_OUTPUT_TOKENS + async def handle( self, content: Content, diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 756e866873..813b7dc14e 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -5262,6 +5262,122 @@ async def test_connect_time_consent_rejects_unsafe_links(self) -> None: # region Error handling (response.failed surfacing) +class TestIncompleteFinishReasonSurfacing: + """A turn the model stopped early must end as ``incomplete`` with the reason, not ``completed``. + + Regression coverage for https://github.com/microsoft/agent-framework/issues/8475: the + underlying chat completion reported ``finish_reason="content_filter"`` but the hosted + ``/responses`` payload said ``status="completed"`` with no trace of the filter. + """ + + @staticmethod + def _filtered_agent(*, finish_reason: str) -> MagicMock: + return _make_agent( + stream_updates=[ + AgentResponseUpdate( + contents=[Content.from_text("I'm sorry, but I cannot assist with that request.")], + role="assistant", + finish_reason=finish_reason, # type: ignore[arg-type] + ) + ] + ) + + async def test_non_streaming_content_filter_marks_response_incomplete(self) -> None: + server = _make_server(self._filtered_agent(finish_reason="content_filter")) + + resp = await _post(server, input_text="hello", stream=False) + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "incomplete" + assert body["incomplete_details"] == {"reason": "content_filter"} + assert body.get("error") is None + + # The refusal text is still delivered so the caller can show it if it chooses to. + messages = [it for it in body["output"] if it["type"] == "message"] + assert len(messages) == 1 + assert messages[0]["content"][0]["text"] == "I'm sorry, but I cannot assist with that request." + + async def test_streaming_content_filter_emits_response_incomplete(self) -> None: + server = _make_server(self._filtered_agent(finish_reason="content_filter")) + + resp = await _post(server, input_text="hello", stream=True) + assert resp.status_code == 200 + events = _parse_sse_events(resp.text) + types = _sse_event_types(events) + + assert types[-1] == "response.incomplete" + assert "response.completed" not in types + incomplete = events[-1]["data"]["response"] + assert incomplete["status"] == "incomplete" + assert incomplete["incomplete_details"] == {"reason": "content_filter"} + # The text item itself still closes normally before the terminal event. + assert "response.output_text.done" in types + + async def test_length_finish_reason_maps_to_max_output_tokens(self) -> None: + server = _make_server(self._filtered_agent(finish_reason="length")) + + resp = await _post(server, input_text="hello", stream=False) + body = resp.json() + assert body["status"] == "incomplete" + assert body["incomplete_details"] == {"reason": "max_output_tokens"} + + async def test_normal_finish_reasons_still_complete(self) -> None: + agent = _make_agent( + stream_updates=[ + AgentResponseUpdate(contents=[Content.from_text("part one")], role="assistant"), + AgentResponseUpdate(contents=[Content.from_text(" part two")], role="assistant", finish_reason="stop"), + ] + ) + server = _make_server(agent) + + resp = await _post(server, input_text="hello", stream=False) + body = resp.json() + assert body["status"] == "completed" + assert body.get("incomplete_details") is None + + async def test_content_filter_persists_across_later_updates_in_the_turn(self) -> None: + """A filter mid-turn is not erased by a later update that finishes normally.""" + agent = _make_agent( + stream_updates=[ + AgentResponseUpdate( + contents=[Content.from_text("filtered")], role="assistant", finish_reason="content_filter" + ), + AgentResponseUpdate(contents=[Content.from_text("trailing")], role="assistant", finish_reason="stop"), + ] + ) + server = _make_server(agent) + + resp = await _post(server, input_text="hello", stream=False) + body = resp.json() + assert body["status"] == "incomplete" + assert body["incomplete_details"] == {"reason": "content_filter"} + + async def test_content_filter_takes_precedence_over_length(self) -> None: + agent = _make_agent( + stream_updates=[ + AgentResponseUpdate(contents=[Content.from_text("cut")], role="assistant", finish_reason="length"), + AgentResponseUpdate( + contents=[Content.from_text("filtered")], role="assistant", finish_reason="content_filter" + ), + ] + ) + server = _make_server(agent) + + resp = await _post(server, input_text="hello", stream=False) + body = resp.json() + assert body["incomplete_details"] == {"reason": "content_filter"} + + async def test_workflow_agent_content_filter_marks_response_incomplete(self) -> None: + workflow_agent = _build_text_workflow_agent("filtered by workflow", finish_reason="content_filter") + server = _make_server(workflow_agent) + + resp = await _post(server, input_text="hi", stream=False) + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "incomplete" + assert body["incomplete_details"] == {"reason": "content_filter"} + + class TestResponseFailedSurfacing: """Tests that exceptions raised by the hosted agent are converted into terminal ``response.failed`` events carrying the exception message, @@ -5600,15 +5716,16 @@ async def _iter() -> AsyncIterator[AgentResponseUpdate]: return ResponseStream(_iter(), finalizer=AgentResponse.from_updates) -def _build_text_workflow_agent(text: str) -> WorkflowAgent: +def _build_text_workflow_agent(text: str, *, finish_reason: str | None = None) -> WorkflowAgent: """Build a minimal ``WorkflowAgent`` whose inner agent emits a fixed text.""" class _TextAgent(SupportsAgentRun): - def __init__(self, name: str, text: str) -> None: + def __init__(self, name: str, text: str, finish_reason: str | None) -> None: self.id = str(uuid.uuid4()) self.name = name self.description: str | None = None self._text = text + self._finish_reason = finish_reason def create_session(self, **kwargs: Any) -> AgentSession: del kwargs @@ -5652,17 +5769,19 @@ def run( assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." text = self._text name = self.name + finish_reason = self._finish_reason async def _aiter() -> AsyncIterator[AgentResponseUpdate]: yield AgentResponseUpdate( contents=[Content.from_text(text=text)], role="assistant", author_name=name, + finish_reason=finish_reason, # type: ignore[arg-type] ) return ResponseStream(_aiter(), finalizer=AgentResponse.from_updates) - inner = _TextAgent("text-agent", text) + inner = _TextAgent("text-agent", text, finish_reason) @executor async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None: From d55233535a91a66afd7b958958453ebf7eebe702 Mon Sep 17 00:00:00 2001 From: Anish Mehta Date: Thu, 17 Sep 2026 22:37:56 +0530 Subject: [PATCH 2/3] Persist the incomplete reason across resilient checkpoint recovery Recovery rebuilds _OutputItemTracker from the persisted response, so a marker held only in memory was lost if the crash landed between a filtered update and a later one. Mirror it into the stream's internal_metadata (persisted with every checkpoint, like the last checkpoint id) and restore it in the tracker constructor. Co-Authored-By: Claude Opus 5 (1M context) --- .../_responses.py | 14 ++++++++++- .../foundry_hosting/tests/test_responses.py | 24 ++++++++++++++++++- 2 files changed, 36 insertions(+), 2 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 bdb37c56cb..3d2cb397d9 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -275,6 +275,9 @@ async def aclose(self) -> None: # checkpoint in storage (if any), or replay the original input if none exists as no output was ever # durably persisted. _LATEST_CHECKPOINT_ID_KEY = "_last_checkpoint_id" +# ``internal_metadata`` key carrying a truncating finish reason across resilient checkpoints, so a +# turn cut short before a crash still ends ``incomplete`` after recovery. +_INCOMPLETE_REASON_KEY = "_incomplete_reason" # Foundry Toolbox Auth integration @@ -1297,8 +1300,14 @@ def __init__(self, stream: ResponseEventStream) -> None: self._oauth_consent_requests: set[tuple[str, str]] = set() # Set when an agent update reports the model stopped early (content filter, token # limit); the response then ends as ``incomplete`` instead of ``completed`` so callers - # can tell a cut-short turn from a successful one. + # can tell a cut-short turn from a successful one. Mirrored into the stream's + # ``internal_metadata`` so it survives a resilient checkpoint/recovery cycle, which + # rebuilds this tracker from the persisted response. self._incomplete_reason: ResponseIncompleteReason | None = None + persisted_reason = stream.internal_metadata.get(_INCOMPLETE_REASON_KEY) + if isinstance(persisted_reason, str): + with suppress(ValueError): + self._incomplete_reason = ResponseIncompleteReason(persisted_reason) for item in stream.response.get("output", []): if not isinstance(item, Mapping): continue @@ -1354,6 +1363,9 @@ def record_finish_reason(self, finish_reason: str | None) -> None: self._incomplete_reason = ResponseIncompleteReason.CONTENT_FILTER elif finish_reason == "length" and self._incomplete_reason is None: self._incomplete_reason = ResponseIncompleteReason.MAX_OUTPUT_TOKENS + else: + return + self._stream.internal_metadata[_INCOMPLETE_REASON_KEY] = self._incomplete_reason.value async def handle( self, diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 813b7dc14e..642f525930 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -63,7 +63,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, ResponseIncompleteReason from azure.ai.agentserver.responses.streaming._checkpoint import ResponseCheckpointEvent from mcp import McpError from mcp.types import ErrorData @@ -72,6 +72,7 @@ from agent_framework_foundry_hosting import ResponsesHostServer from agent_framework_foundry_hosting._responses import ( + _INCOMPLETE_REASON_KEY, # pyright: ignore[reportPrivateUsage] CONSENT_ERROR_CODE, ConsentError, _item_to_message, # pyright: ignore[reportPrivateUsage] @@ -5367,6 +5368,27 @@ async def test_content_filter_takes_precedence_over_length(self) -> None: body = resp.json() assert body["incomplete_details"] == {"reason": "content_filter"} + async def test_incomplete_reason_survives_checkpoint_recovery(self) -> None: + """Resilient recovery rebuilds the tracker from the persisted response; the marker must ride along. + + A filtered update followed by a crash and a later ``stop`` update must still end ``incomplete``. + """ + stream = ResponseEventStream(response_id="resp_filtered") + stream.emit_created() + stream.emit_in_progress() + tracker = _OutputItemTracker(stream) + tracker.record_finish_reason("content_filter") + assert stream.internal_metadata[_INCOMPLETE_REASON_KEY] == "content_filter" + + # Simulate recovery: a fresh tracker over the checkpointed response snapshot. + recovered = _OutputItemTracker(stream) + assert recovered.incomplete_reason == ResponseIncompleteReason.CONTENT_FILTER + recovered.record_finish_reason("stop") + assert recovered.incomplete_reason == ResponseIncompleteReason.CONTENT_FILTER + + # A stream that was never marked restores nothing. + assert _OutputItemTracker(ResponseEventStream(response_id="resp_clean")).incomplete_reason is None + async def test_workflow_agent_content_filter_marks_response_incomplete(self) -> None: workflow_agent = _build_text_workflow_agent("filtered by workflow", finish_reason="content_filter") server = _make_server(workflow_agent) From 55f009d18da03325ce07cf4ce0fce0dea3fc83f2 Mon Sep 17 00:00:00 2001 From: Anish Mehta Date: Fri, 18 Sep 2026 13:15:50 +0530 Subject: [PATCH 3/3] Handle updates through one tracker entry point and pair snapshots with the checkpoint an update follows _OutputItemTracker.handle_update records the update's finish reason and then handles its contents, so the plain-agent and workflow loops share one ordering rule. The workflow loop snapshotted the response against the latest checkpoint in storage at the time an update was consumed. _SignalledIterator drives the workflow one update ahead, so by then the runner could already have checkpointed past the update; a crash after that snapshot resumed the workflow with the update never replayed and its output (and any incomplete reason) lost. _SignalledIterator now takes an optional stamp coroutine that the driver awaits right after each item is produced, before advancing the wrapped iterator again, and the loop pairs snapshots with that stamped checkpoint id. After the workflow completes, its final checkpoint is paired with the full output so recovery does not replay the last superstep. The real-crash recovery integration tests that were xfailed against #7809 pass with this. Co-Authored-By: Claude Opus 5 (1M context) --- .../_responses.py | 118 ++++++++++++------ .../foundry_hosting/tests/test_responses.py | 83 +++++++++++- 2 files changed, 164 insertions(+), 37 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 3d2cb397d9..cf0b46695f 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -172,15 +172,27 @@ class _SignalledIterator(Generic[_T]): task -- and the real agent/workflow run it's pumping -- would otherwise be silently abandoned. """ - def __init__(self, iterator: AsyncIterator[_T], *events: asyncio.Event) -> None: + def __init__( + self, + iterator: AsyncIterator[_T], + *events: asyncio.Event, + stamp: Callable[[], Awaitable[Any]] | None = None, + ) -> None: """Wrap an async iterator, stopping early if any of ``events`` fires. Args: iterator: The async iterator to wrap. events: One or more asyncio.Event objects to watch for. If any of them is set, iteration stops early. + stamp: Optional coroutine function the driver awaits right after the wrapped iterator produces an + item and before it is advanced again. Its result is exposed as :attr:`stamp` while that item + is the current one, which lets a consumer observe state (e.g. the latest persisted workflow + checkpoint) as it was when the item was produced rather than when it is consumed: the driver + runs one item ahead, so by consumption time the wrapped iterator may already have moved on. """ self._iterator = iterator self._events = events + self._stamp_fn = stamp + self._stamp: Any = None self._signalled = False # The queue is used to communicate items from the background driver task to the main iteration loop. self._queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=1) @@ -196,6 +208,11 @@ def signalled(self) -> bool: """ return self._signalled + @property + def stamp(self) -> Any: + """The ``stamp`` result taken when the current item was produced (``None`` without a ``stamp``).""" + return self._stamp + def __aiter__(self) -> _SignalledIterator[_T]: return self @@ -205,13 +222,14 @@ async def _drive(self) -> None: while True: try: item: Any = await self._iterator.__anext__() + stamp = await self._stamp_fn() if self._stamp_fn is not None else None except StopAsyncIteration: await self._queue.put(_STOP_SENTINEL) return except Exception as exc: await self._queue.put(exc) return - await self._queue.put(item) + await self._queue.put((item, stamp)) finally: iterator: AsyncIterator[_T] = self._iterator if isinstance(iterator, ResponseStream): @@ -252,6 +270,7 @@ async def __anext__(self) -> _T: raise StopAsyncIteration if isinstance(item, Exception): raise item + item, self._stamp = item return cast(_T, item) async def aclose(self) -> None: @@ -969,12 +988,8 @@ async def _handle_inner_agent( ) async with aclosing(agent_stream): async for update in agent_stream: - tracker.record_finish_reason(update.finish_reason) - for content in update.contents: - async for event in tracker.handle( - content, message_id=update.message_id, approval_storage=approval_storage - ): - yield event + async for event in tracker.handle_update(update, approval_storage=approval_storage): + yield event except (asyncio.CancelledError, GeneratorExit): request_interrupted = True raise @@ -1167,42 +1182,57 @@ async def _handle_inner_workflow( checkpoint_storage=checkpoint_storage, ) - main_iter = _SignalledIterator(run_stream, context.shutdown, cancellation_signal) + workflow_name = agent.workflow.name + + async def latest_checkpoint_id() -> str | None: + latest = await checkpoint_storage.get_latest(workflow_name=workflow_name) + return latest.checkpoint_id if latest is not None else None + + def snapshot_response( + checkpoint_id: str | None, + ) -> Generator[ResponseStreamEvent | ResponseCheckpointEvent, None, None]: + # Pair the response output emitted so far with the workflow checkpoint it corresponds + # to, so recovery from that checkpoint replays exactly the updates that came after it. + if checkpoint_id is None or checkpoint_id == response_event_stream.internal_metadata.get( + _LATEST_CHECKPOINT_ID_KEY + ): + return + yield from tracker.close() + response_event_stream.internal_metadata[_LATEST_CHECKPOINT_ID_KEY] = checkpoint_id + yield response_event_stream.checkpoint() + + main_iter = _SignalledIterator( + run_stream, + context.shutdown, + cancellation_signal, + # The runner creates a checkpoint at the end of each superstep, inside the generator + # that produces the updates (see RunnerImpl.run_until_convergence). Stamping each + # update with the latest checkpoint as it was produced tells which checkpoint the + # update follows; the driver runs one update ahead, so by the time an update is + # consumed the workflow may already have checkpointed past it. + stamp=latest_checkpoint_id if self._resilient_background else None, + ) async with aclosing(main_iter): async for update in main_iter: if self._resilient_background: - latest_checkpoint = await checkpoint_storage.get_latest(workflow_name=agent.workflow.name) - if ( - latest_checkpoint is not None - and latest_checkpoint.checkpoint_id - != response_event_stream.internal_metadata.get(_LATEST_CHECKPOINT_ID_KEY) - ): - # A new checkpoint is created when we pull the next item from the stream - # (see RunnerImpl.run_until_convergence). We only take a snapshot of the - # response (response_event_stream.checkpoint()) once the checkpoint is - # durably persisted. This means all items from the previous superstep - # has been pulled thus we can safely close the tracker. The latest checkpoint - # now reflects the state of the workflow that matches the response output. - # Note that if a workflow crashes before any update is created, no response - # snapshot is taken. However, upon recovery the workflow will still be resumed - # from the latest checkpoint. - for event in tracker.close(): - yield event - response_event_stream.internal_metadata[_LATEST_CHECKPOINT_ID_KEY] = ( - latest_checkpoint.checkpoint_id - ) - yield response_event_stream.checkpoint() - - tracker.record_finish_reason(update.finish_reason) - for content in update.contents: - async for event in tracker.handle( - content, message_id=update.message_id, approval_storage=approval_storage - ): + # Every update before this one belongs to the stamped checkpoint (or an + # earlier one), so the output so far can be snapshotted against it. If the + # workflow crashes before any update is produced, no snapshot is taken and + # recovery still resumes from the latest workflow checkpoint. + for event in snapshot_response(main_iter.stamp): yield event + + async for event in tracker.handle_update(update, approval_storage=approval_storage): + yield event # Cancellation needs no extra action here (the loop above already stopped); shutdown # does, but only if it's what actually stopped the loop, not a natural completion. if main_iter.signalled and context.shutdown.is_set(): await context.exit_for_recovery() + elif self._resilient_background and not main_iter.signalled: + # The workflow ran to completion: pair its final checkpoint with the full output, so + # recovery after this point does not replay the last superstep. + for event in snapshot_response(await latest_checkpoint_id()): + yield event except Exception: logger.exception("Failed to produce response for workflow agent") raise @@ -1367,6 +1397,22 @@ def record_finish_reason(self, finish_reason: str | None) -> None: return self._stream.internal_metadata[_INCOMPLETE_REASON_KEY] = self._incomplete_reason.value + async def handle_update( + self, + update: AgentResponseUpdate, + *, + approval_storage: FunctionApprovalStore | None = None, + ) -> AsyncGenerator[ResponseStreamEvent]: + """Process one agent update: note its finish reason, then handle each of its contents. + + This is the single entry point for both the plain-agent and the workflow loops, so the + finish reason cannot be forgotten on one of them. + """ + self.record_finish_reason(update.finish_reason) + for content in update.contents: + async for event in self.handle(content, message_id=update.message_id, approval_storage=approval_storage): + yield event + async def handle( self, content: Content, diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 642f525930..58d4322caa 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -11,12 +11,13 @@ from __future__ import annotations import asyncio +import copy import json import logging import os import uuid from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Generator, Mapping, Sequence -from contextlib import asynccontextmanager +from contextlib import aclosing, asynccontextmanager from dataclasses import dataclass from importlib import import_module from pathlib import Path @@ -73,12 +74,14 @@ from agent_framework_foundry_hosting import ResponsesHostServer from agent_framework_foundry_hosting._responses import ( _INCOMPLETE_REASON_KEY, # pyright: ignore[reportPrivateUsage] + _LATEST_CHECKPOINT_ID_KEY, # pyright: ignore[reportPrivateUsage] CONSENT_ERROR_CODE, ConsentError, _item_to_message, # pyright: ignore[reportPrivateUsage] _json_safe_to_str, # pyright: ignore[reportPrivateUsage] _output_item_to_message, # pyright: ignore[reportPrivateUsage] _OutputItemTracker, # pyright: ignore[reportPrivateUsage] + _SignalledIterator, # pyright: ignore[reportPrivateUsage] _stringify_mcp_output, # pyright: ignore[reportPrivateUsage] consent_url_from_error, ) @@ -6468,6 +6471,84 @@ async def test_workflow_yields_checkpoint_event_when_resilient_background(self, checkpoint_events = [e for e in events if isinstance(e, ResponseCheckpointEvent)] assert checkpoint_events, "expected at least one checkpoint event yielded for a resilient background run" + async def test_signalled_iterator_stamps_items_when_produced(self) -> None: + """The stamp reflects state as of production, not consumption. + + The driver runs one item ahead: once the consumer holds item k, the wrapped iterator may + already have resumed and created a checkpoint after it. The stamp taken right after item k + was produced must not see that later checkpoint. + """ + checkpoints: list[int] = [] + + async def produce() -> AsyncIterator[int]: + for k in range(1, 4): + yield k + # Runs when the iterator is resumed to produce the next item, i.e. after item k + # was handed over, mirroring the runner checkpointing at the end of a superstep. + checkpoints.append(k) + + async def stamp() -> int: + return len(checkpoints) + + seen: list[tuple[int, int, int]] = [] + it = _SignalledIterator(produce(), asyncio.Event(), stamp=stamp) + async with aclosing(it): + async for item in it: + # Give the driver every chance to run ahead before we look at the stamp. + for _ in range(5): + await asyncio.sleep(0) + seen.append((item, it.stamp, len(checkpoints))) + + assert [(item, stamped) for item, stamped, _ in seen] == [(1, 0), (2, 1), (3, 2)] + # Consumption-time state had already moved past the stamped one for every item. + assert all(consumed > stamped for _, stamped, consumed in seen) + + async def test_snapshots_pair_output_with_the_checkpoint_it_follows(self, tmp_path: Path) -> None: + """Every persisted snapshot must contain exactly the output emitted before it, and the final + snapshot must carry the full output and the incomplete reason. + """ + workflow_agent = _build_text_workflow_agent("filtered by workflow", finish_reason="content_filter") + server = _make_server( + workflow_agent, + response_store=FileResponseStore(storage_dir=tmp_path), + options=ResponsesServerOptions(resilient_background=True), + ) + request = CreateResponse(model="m", input="hi", background=True, stream=True, store=True) + context = ResponseContext(response_id="response-current", mode_flags=MagicMock()) + + emitted_text = "" + snapshots: list[tuple[str, dict[str, Any]]] = [] + async for event in server._handle_response( # pyright: ignore[reportPrivateUsage] + request, context, asyncio.Event() + ): + if isinstance(event, ResponseCheckpointEvent): + # The event references the live response; copy it as it is at persistence time. + snapshots.append((emitted_text, copy.deepcopy(dict(event.response)))) + elif isinstance(event, Mapping) and event.get("type") == "response.output_text.delta": + emitted_text += str(event["delta"]) + + assert emitted_text == "filtered by workflow" + assert snapshots, "expected the completed workflow to be snapshotted" + checkpoint_ids: list[str] = [] + for text_before, response in snapshots: + internal = json.loads(response["metadata"]["_internal_metadata"]) + checkpoint_ids.append(internal[_LATEST_CHECKPOINT_ID_KEY]) + snapshot_text = "".join( + part["text"] + for item in response["output"] + if item["type"] == "message" + for part in item["content"] + if part["type"] == "output_text" + ) + assert snapshot_text == text_before + assert len(set(checkpoint_ids)) == len(checkpoint_ids), "each checkpoint is snapshotted once" + + # The last snapshot is paired with the workflow's final checkpoint and carries everything. + final_text, final_response = snapshots[-1] + assert final_text == "filtered by workflow" + assert json.loads(final_response["metadata"]["_internal_metadata"])[_INCOMPLETE_REASON_KEY] == "content_filter" + assert final_response["status"] == "in_progress" + # endregion