From 3a7e54a2f11a499ca9bbe42d7d2d82d03b54ab8f Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Sun, 20 Sep 2026 22:34:33 +0000 Subject: [PATCH 1/3] Python: Don't emit response.completed after a cancelled Foundry hosting run Fixes #8564. When cancellation_signal fires, _handle_inner_workflow and _handle_inner_agent stop their _SignalledIterator loop and return normally instead of raising, so the post-loop code in _handle_prepared_response treated the cancelled run as a success and still emitted response.completed (or response.incomplete). Add an early return once the inner generator drains so a cancelled run emits no terminal event, letting the cancel-aware host layer synthesize the cancelled terminal instead. --- .../_responses.py | 9 +++++ .../foundry_hosting/tests/test_responses.py | 34 ++++++++++++------- 2 files changed, 30 insertions(+), 13 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 6f7b9ebcf29..3a171920368 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -794,6 +794,15 @@ async def _handle_prepared_response( await inner.aclose() raise + if cancellation_signal.is_set(): + # A cancelled run drains the inner generator without raising (both + # ``_handle_inner_workflow`` and ``_handle_inner_agent`` stop their + # ``_SignalledIterator`` loop and return normally once the signal fires). + # Emit nothing here so a caller cannot mistake this for a normal + # completion; the host server's cancel-aware layer synthesizes the + # cancelled terminal when the handler returns without one. + return + for event in tracker.close(): yield event diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 756e8668739..f7dbaa09728 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -1502,8 +1502,9 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: assert stored is not None assert stored.state["started"] is True - async def test_cancellation_signal_stops_streaming_and_completes(self) -> None: - """Steering/explicit-cancel: the loop must break promptly, and the response still completes.""" + async def test_cancellation_signal_stops_streaming_without_completing(self) -> None: + """Steering/explicit-cancel: the loop must break promptly, and the handler must not emit a + ``response.completed`` terminal for a run it didn't finish (regression for #8564).""" store = SessionStore() agent = _make_agent( stream_updates=[ @@ -1536,7 +1537,8 @@ async def test_cancellation_signal_stops_streaming_and_completes(self) -> None: types = [event.get("type") for event in events if isinstance(event, Mapping)] assert types.count("response.output_text.delta") == 1 - assert types[-1] == "response.completed" + assert "response.completed" not in types + assert types[-1] == "response.output_text.delta" stored = await store.get("response-1") assert stored is not None @@ -1586,9 +1588,7 @@ async def _drain() -> list[Any]: # call instead of only being observed after it (eventually) produced an update. events = await asyncio.wait_for(_drain(), timeout=1.0) - types = [event.get("type") for event in events if isinstance(event, Mapping)] - assert "response.output_text.delta" not in types - assert types[-1] == "response.completed" + assert events == [] assert cleanup_called.is_set() async def test_consumer_failure_cancels_agent_stream_driver_task(self) -> None: @@ -5885,8 +5885,9 @@ async def test_basic_text_response_streaming(self) -> None: text_done = [e for e in events if e["event"] == "response.output_text.done"] assert any(e["data"]["text"] == "hello stream" for e in text_done) - async def test_cancellation_signal_stops_main_loop_and_completes(self) -> None: - """Explicit-cancel: the workflow's main loop must break promptly and still complete.""" + async def test_cancellation_signal_stops_main_loop_without_completing(self) -> None: + """Explicit-cancel: the workflow's main loop must break promptly, and the handler must not + emit a ``response.completed`` terminal for a run it didn't finish (regression for #8564).""" workflow_agent, inner = _build_multi_update_workflow_agent(["one", "two", "three"]) server = _make_server(workflow_agent) request = CreateResponse(model="m", input="hi", stream=True) @@ -5912,7 +5913,8 @@ async def test_cancellation_signal_stops_main_loop_and_completes(self) -> None: types = [event.get("type") for event in events if isinstance(event, Mapping)] assert types.count("response.output_text.delta") == 1 - assert types[-1] == "response.completed" + assert "response.completed" not in types + assert types[-1] == "response.output_text.delta" assert inner.run_count == 1 async def test_cancellation_signal_preempts_stuck_workflow_call(self) -> None: @@ -5940,8 +5942,13 @@ async def test_cancellation_signal_preempts_stuck_workflow_call(self) -> None: cancellation_signal.set() # Fires while the inner agent is stuck awaiting `gate`. async def _drain() -> list[Any]: - first = await pending - return [first, *[event async for event in handler]] + events: list[Any] = [] + try: + events.append(await pending) + except StopAsyncIteration: + return events + events.extend([event async for event in handler]) + return events # Bounded well below `gate` never being set: proves cancellation preempted the stuck # call instead of only being observed after it (eventually) produced an update. @@ -5950,7 +5957,7 @@ async def _drain() -> list[Any]: types = [event.get("type") for event in events if isinstance(event, Mapping)] assert "response.output_text.delta" not in types - assert types[-1] == "response.completed" + assert "response.completed" not in types assert inner.run_count == 1 async def test_shutdown_signal_preempts_stuck_workflow_call(self, tmp_path: Path) -> None: @@ -6020,7 +6027,8 @@ async def test_cancellation_signal_set_before_turn_skips_new_input(self) -> None types = [event.get("type") for event in events if isinstance(event, Mapping)] assert "response.output_text.delta" not in types - assert types[-1] == "response.completed" + assert "response.completed" not in types + assert types[-1] == "response.in_progress" # At most the restore-only replay call happened; the new-turn call (which would deliver # "hi again") must never fire. assert inner.run_count <= run_count_after_first_turn + 1 From d2813dd1e9e3490ec18922be2da0ae16cd33fa75 Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Sun, 20 Sep 2026 23:08:59 +0000 Subject: [PATCH 2/3] Python: Recheck cancellation after draining tracker.close() before terminal event Copilot review on #8565 pointed out that the cancellation_signal check added there runs before tracker.close(), but each event that loop yields suspends the handler. A cancellation arriving while those trailing *.done events are being consumed still fell through to emit_completed/emit_incomplete, reproducing the #8564 race for a narrower window. Recheck the signal once more immediately after draining tracker.close(), before selecting the terminal event. --- .../_responses.py | 8 +++++ .../foundry_hosting/tests/test_responses.py | 36 +++++++++++++++++++ 2 files changed, 44 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 3a171920368..e6c471b23bd 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -806,6 +806,14 @@ async def _handle_prepared_response( for event in tracker.close(): yield event + if cancellation_signal.is_set(): + # Draining ``tracker.close()`` yields events one at a time, and each + # ``yield`` above suspends this handler until the caller resumes it. + # A cancellation can arrive during that window, after the earlier check + # already passed, so it must be rechecked here, immediately before + # selecting the terminal event. + return + if tracker.oauth_consent_requested: yield response_event_stream.emit_incomplete(usage=tracker.usage) else: diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index f7dbaa09728..0f262501a9a 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -1543,6 +1543,42 @@ async def test_cancellation_signal_stops_streaming_without_completing(self) -> N stored = await store.get("response-1") assert stored is not None + async def test_cancellation_signal_during_close_drain_stops_completion(self) -> None: + """Regression for #8564 (Copilot follow-up): the cancellation recheck must happen *after* + draining ``tracker.close()``, not only before it. Each event that loop yields suspends the + handler, so a cancellation arriving mid-drain must still suppress ``response.completed``.""" + store = SessionStore() + agent = _make_agent( + stream_updates=[AgentResponseUpdate(contents=[Content.from_text("done")], role="assistant")] + ) + server = _make_server(agent, session_store=store) + request = CreateResponse(model="m", input="hi", stream=True) + context = ResponseContext(response_id="response-1", mode_flags=MagicMock()) + cancellation_signal = asyncio.Event() + + with ( + patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])), + patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])), + ): + handler = cast( + AsyncGenerator[Any, None], + server._handle_response(request, context, cancellation_signal), # pyright: ignore[reportPrivateUsage] + ) + events: list[Any] = [] + async for event in handler: + events.append(event) + # The inner agent stream has already finished draining (so the earlier check + # passed) by the time `tracker.close()` emits its first closing event; fire the + # cancellation exactly then, mid-drain, instead of before the drain starts. + if isinstance(event, Mapping) and event.get("type") == "response.output_text.done": + cancellation_signal.set() + break + events.extend([event async for event in handler]) + + types = [event.get("type") for event in events if isinstance(event, Mapping)] + assert "response.output_text.done" in types + assert "response.completed" not in types + async def test_cancellation_signal_preempts_stuck_agent_call(self) -> None: """Steering/explicit-cancel must interrupt an agent call stuck awaiting a slow model/tool response, not merely be checked between already-produced updates.""" From 7ae1fe55cb94112706aa7d76884c0ec9bbc63b16 Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Mon, 21 Sep 2026 01:17:06 +0000 Subject: [PATCH 3/3] Python: Gate cancellation-terminal suppression on client_cancelled Only skip the response.completed/incomplete terminal when the cancellation_signal cause is an explicit /cancel (or non-bg disconnect), i.e. context.client_cancelled is True. cancellation_signal also fires for steering pressure with no cause flag set, so without this gate a steered turn's partial output would be silently dropped instead of completing normally. Update the existing explicit-cancel tests to set client_cancelled alongside the signal, and add a regression test for the steering case. --- .../_responses.py | 12 ++-- .../foundry_hosting/tests/test_responses.py | 68 +++++++++++++++++-- 2 files changed, 69 insertions(+), 11 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 e6c471b23bd..155ef91272f 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -794,24 +794,28 @@ async def _handle_prepared_response( await inner.aclose() raise - if cancellation_signal.is_set(): + if cancellation_signal.is_set() and context.client_cancelled: # A cancelled run drains the inner generator without raising (both # ``_handle_inner_workflow`` and ``_handle_inner_agent`` stop their # ``_SignalledIterator`` loop and return normally once the signal fires). # Emit nothing here so a caller cannot mistake this for a normal # completion; the host server's cancel-aware layer synthesizes the - # cancelled terminal when the handler returns without one. + # cancelled terminal when the handler returns without one. Gated on + # ``client_cancelled`` (not just the signal) because steering pressure + # also sets ``cancellation_signal`` without that cause flag; a steered + # turn must still drain ``tracker.close()`` and emit its normal terminal + # below so its partial output is not misreported as a failure. return for event in tracker.close(): yield event - if cancellation_signal.is_set(): + if cancellation_signal.is_set() and context.client_cancelled: # Draining ``tracker.close()`` yields events one at a time, and each # ``yield`` above suspends this handler until the caller resumes it. # A cancellation can arrive during that window, after the earlier check # already passed, so it must be rechecked here, immediately before - # selecting the terminal event. + # selecting the terminal event. Same ``client_cancelled`` gate as above. return if tracker.oauth_consent_requested: diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 0f262501a9a..42cf749870e 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -1503,7 +1503,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: assert stored.state["started"] is True async def test_cancellation_signal_stops_streaming_without_completing(self) -> None: - """Steering/explicit-cancel: the loop must break promptly, and the handler must not emit a + """Explicit cancel: the loop must break promptly, and the handler must not emit a ``response.completed`` terminal for a run it didn't finish (regression for #8564).""" store = SessionStore() agent = _make_agent( @@ -1531,7 +1531,9 @@ async def test_cancellation_signal_stops_streaming_without_completing(self) -> N events.append(event) if isinstance(event, Mapping) and event.get("type") == "response.output_text.delta": break - # Cancellation arrives after the first delta; the loop must not process "two"/"three". + # Cancellation arrives after the first delta, via the explicit /cancel endpoint (both + # the signal and its cause flag fire together); the loop must not process "two"/"three". + context.client_cancelled = True cancellation_signal.set() events.extend([event async for event in handler]) @@ -1546,7 +1548,7 @@ async def test_cancellation_signal_stops_streaming_without_completing(self) -> N async def test_cancellation_signal_during_close_drain_stops_completion(self) -> None: """Regression for #8564 (Copilot follow-up): the cancellation recheck must happen *after* draining ``tracker.close()``, not only before it. Each event that loop yields suspends the - handler, so a cancellation arriving mid-drain must still suppress ``response.completed``.""" + handler, so an explicit cancel arriving mid-drain must still suppress ``response.completed``.""" store = SessionStore() agent = _make_agent( stream_updates=[AgentResponseUpdate(contents=[Content.from_text("done")], role="assistant")] @@ -1569,8 +1571,9 @@ async def test_cancellation_signal_during_close_drain_stops_completion(self) -> events.append(event) # The inner agent stream has already finished draining (so the earlier check # passed) by the time `tracker.close()` emits its first closing event; fire the - # cancellation exactly then, mid-drain, instead of before the drain starts. + # explicit cancel exactly then, mid-drain, instead of before the drain starts. if isinstance(event, Mapping) and event.get("type") == "response.output_text.done": + context.client_cancelled = True cancellation_signal.set() break events.extend([event async for event in handler]) @@ -1579,8 +1582,54 @@ async def test_cancellation_signal_during_close_drain_stops_completion(self) -> assert "response.output_text.done" in types assert "response.completed" not in types + async def test_steering_pressure_without_client_cancel_still_completes_normally(self) -> None: + """Steering: ``cancellation_signal`` also fires when a steerable conversation supersedes a + turn, but ``context.client_cancelled`` stays False for that cause (only the explicit + /cancel endpoint or a non-background disconnect sets it). A steered turn must still drain + ``tracker.close()`` and emit its normal terminal below so agentserver preserves the partial + output as ``response.completed`` instead of synthesizing ``response.failed`` for it.""" + store = SessionStore() + agent = _make_agent( + stream_updates=[ + AgentResponseUpdate(contents=[Content.from_text("one")], role="assistant"), + AgentResponseUpdate(contents=[Content.from_text("two")], role="assistant"), + ] + ) + server = _make_server(agent, session_store=store) + request = CreateResponse(model="m", input="hi", stream=True) + context = ResponseContext(response_id="response-1", mode_flags=MagicMock()) + cancellation_signal = asyncio.Event() + + with ( + patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])), + patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])), + ): + handler = cast( + AsyncGenerator[Any, None], + server._handle_response(request, context, cancellation_signal), # pyright: ignore[reportPrivateUsage] + ) + events: list[Any] = [] + async for event in handler: + events.append(event) + if isinstance(event, Mapping) and event.get("type") == "response.output_text.delta": + break + # Steering pressure supersedes the turn: the signal fires with no cause flag + # (``client_cancelled`` stays False), unlike an explicit /cancel. + assert context.client_cancelled is False + cancellation_signal.set() + events.extend([event async for event in handler]) + + types = [event.get("type") for event in events if isinstance(event, Mapping)] + assert types.count("response.output_text.delta") == 1 + assert "response.completed" in types + text_done = [e for e in events if isinstance(e, Mapping) and e.get("type") == "response.output_text.done"] + assert any(e.get("text") == "one" for e in text_done) + + stored = await store.get("response-1") + assert stored is not None + async def test_cancellation_signal_preempts_stuck_agent_call(self) -> None: - """Steering/explicit-cancel must interrupt an agent call stuck awaiting a slow model/tool + """Explicit cancel must interrupt an agent call stuck awaiting a slow model/tool response, not merely be checked between already-produced updates.""" store = SessionStore() gate = asyncio.Event() # Never set: simulates a model/tool call that never returns. @@ -1615,6 +1664,7 @@ def run_streaming(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpd ) await anext(handler) # response.created await anext(handler) # response.in_progress + context.client_cancelled = True cancellation_signal.set() # Fires while the agent is stuck awaiting `gate`. async def _drain() -> list[Any]: @@ -5943,7 +5993,9 @@ async def test_cancellation_signal_stops_main_loop_without_completing(self) -> N events.append(event) if isinstance(event, Mapping) and event.get("type") == "response.output_text.delta": break - # Cancellation arrives after the first delta; the loop must not process "two"/"three". + # Cancellation arrives after the first delta, via the explicit /cancel endpoint (both + # the signal and its cause flag fire together); the loop must not process "two"/"three". + context.client_cancelled = True cancellation_signal.set() events.extend([event async for event in handler]) @@ -5975,6 +6027,7 @@ async def test_cancellation_signal_preempts_stuck_workflow_call(self) -> None: await anext(handler) # response.in_progress async with _pending_workflow_event(handler, inner.started) as pending: + context.client_cancelled = True cancellation_signal.set() # Fires while the inner agent is stuck awaiting `gate`. async def _drain() -> list[Any]: @@ -6049,7 +6102,8 @@ async def test_cancellation_signal_set_before_turn_skips_new_input(self) -> None request = CreateResponse(model="m", input="hi again", stream=True) context = ResponseContext(response_id="response-2", mode_flags=MagicMock(), conversation_id="conv-1") cancellation_signal = asyncio.Event() - cancellation_signal.set() # Steering pressure already present before the turn even starts. + context.client_cancelled = True # Explicit cancel already present before the turn even starts. + cancellation_signal.set() with ( patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),