From 28b32374539ef2b7fee6455e201a17e8a1a1cd65 Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Sat, 4 Jul 2026 16:54:53 +0200 Subject: [PATCH] fix(sdk): a paused live stream's finish frame carries the terminal stop reason The live runner emits its 'done' event with no stopReason (the engine settles paused-vs-ended after the event stream closes, onto the terminal result only), so a HITL-paused SSE stream ended with a generic finish frame while the batch envelope already carried stop_reason: paused. - agent_event_stream appends a corrective terminal 'done' when the result's stop_reason disagrees with the streamed one (the streaming analogue of agent_batch's fold(events, stop_reason=result.stop_reason)) - both vercel adapters prefer the last non-null stop reason, and the dev-only AgentStream variant now prefers the terminal result unconditionally (its old fallback-only read meant terminal-wins never applied) Display was never affected (the FE keys off the interaction_request part); this restores wire fidelity for headless/protocol consumers of the stream. --- .../sdk/agents/adapters/vercel/stream.py | 34 ++-- sdks/python/agenta/sdk/agents/handler.py | 17 ++ .../test_vercel_stream_finish_reason.py | 168 ++++++++++++++++++ .../adapters/test_vercel_stream_park.py | 74 ++++++++ 4 files changed, 282 insertions(+), 11 deletions(-) create mode 100644 sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_finish_reason.py diff --git a/sdks/python/agenta/sdk/agents/adapters/vercel/stream.py b/sdks/python/agenta/sdk/agents/adapters/vercel/stream.py index e553bce419..31213ce4ef 100644 --- a/sdks/python/agenta/sdk/agents/adapters/vercel/stream.py +++ b/sdks/python/agenta/sdk/agents/adapters/vercel/stream.py @@ -216,20 +216,25 @@ async def agent_run_to_vercel_parts( elif etype == "error": yield {"type": "error", "errorText": data.get("message", "")} elif etype == "done": - stop_reason = data.get("stopReason") + # Last non-null stop reason wins; see the routing-layer twin's `done` note. + reason = data.get("stopReason") + if reason is not None: + stop_reason = reason except Exception as exc: yield {"type": "error", "errorText": str(exc)} return - if usage is None or trace_id is None: - result = _safe_result(run) - if result is not None: - if usage is None: - usage = _usage_metadata(result.usage or {}) - if stop_reason is None: - stop_reason = result.stop_reason - if trace_id is None: - trace_id = result.trace_id + # The terminal AgentResult is authoritative. Its stop_reason wins over the `done` event's, + # which the live runner leaves empty on a HITL pause (mirrors fold's terminal-wins + # precedence); usage/trace_id fall back to it only when the stream carried neither. + result = _safe_result(run) + if result is not None: + if result.stop_reason is not None: + stop_reason = result.stop_reason + if usage is None: + usage = _usage_metadata(result.usage or {}) + if trace_id is None: + trace_id = result.trace_id yield {"type": "finish-step"} finish: Dict[str, Any] = {"type": "finish"} @@ -414,7 +419,14 @@ async def agent_stream_to_vercel_stream( elif etype == "error": yield {"type": "error", "errorText": data.get("message", "")} elif etype == "done": - stop_reason = data.get("stopReason") + # Prefer the LAST non-null stop reason. The handler appends a corrective + # terminal `done` after the runner's `done` when the authoritative result + # disagrees — the runner omits stopReason on a HITL pause — so the terminal + # value wins, while a null-carrying `done` never clobbers a real earlier one. + # Mirrors fold's terminal-wins precedence (agent_batch / fold.py). + reason = data.get("stopReason") + if reason is not None: + stop_reason = reason except Exception as exc: yield {"type": "error", "errorText": str(exc)} return diff --git a/sdks/python/agenta/sdk/agents/handler.py b/sdks/python/agenta/sdk/agents/handler.py index 0295a53bb2..8525d267cf 100644 --- a/sdks/python/agenta/sdk/agents/handler.py +++ b/sdks/python/agenta/sdk/agents/handler.py @@ -186,8 +186,25 @@ async def agent_event_stream( await harness.setup() run = await harness.stream(session_config, msgs) try: + event_stop_reason: Optional[str] = None async for event in run: + if event.type == "done": + event_stop_reason = (event.data or {}).get("stopReason") yield {"type": event.type, "data": event.data} + # The terminal result's stop_reason is authoritative: the runner's `done` event carries + # no stopReason for a HITL pause (the engine settles paused-vs-ended after the event + # stream closes, onto the terminal result only). When it disagrees with the streamed + # `done`, append a corrective terminal `done` so the egress finish frame can prefer it — + # the streaming analogue of agent_batch's `fold(events, stop_reason=result.stop_reason)`. + try: + terminal_stop_reason = run.result().stop_reason + except Exception: # result unavailable on a failed/aborted stream + terminal_stop_reason = None + if ( + terminal_stop_reason is not None + and terminal_stop_reason != event_stop_reason + ): + yield {"type": "done", "data": {"stopReason": terminal_stop_reason}} finally: try: record_usage(run.result().usage) diff --git a/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_finish_reason.py b/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_finish_reason.py new file mode 100644 index 0000000000..3d54781cdc --- /dev/null +++ b/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_finish_reason.py @@ -0,0 +1,168 @@ +"""Live streaming finishReason: the terminal result's stop_reason is authoritative. + +The live path is ``agent_stream_to_vercel_stream`` (routing projects the handler's neutral +``{type, data}`` event stream — it never sees the terminal ``AgentResult``). The live runner +settles paused-vs-ended AFTER the event stream closes, so its ``done`` event carries NO +stopReason; the authoritative value lands only on the terminal result record. The handler +(``agent_event_stream``) surfaces it by appending a corrective terminal ``done`` when the +result disagrees with the streamed ``done``; the adapter then honors last-non-null precedence +so the paused reason reaches the finish frame. This mirrors the batch +``fold(events, stop_reason=result.stop_reason)`` precedence. +""" + +from __future__ import annotations + +from typing import Any, AsyncIterator, Dict, List, Optional + +import pytest + +from agenta.sdk.agents.adapters.vercel.stream import agent_stream_to_vercel_stream +from agenta.sdk.agents.dtos import AgentResult, Event +from agenta.sdk.agents.handler import agent_event_stream +from agenta.sdk.agents.streaming import AgentStream + + +async def _events(items: List[Dict[str, Any]]) -> AsyncIterator[Dict[str, Any]]: + for item in items: + yield item + + +async def _collect(events: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + return [part async for part in agent_stream_to_vercel_stream(_events(events))] + + +# --- the adapter: last non-null stop reason wins ------------------------------ + + +@pytest.mark.asyncio +async def test_terminal_paused_done_wins_over_earlier_reasonless_done() -> None: + """(a) The runner's ``done`` omits stopReason; the handler's corrective terminal + ``done`` (paused) follows -> the finish frame carries the paused reason (AI SDK ``other``).""" + parts = await _collect( + [ + {"type": "message", "data": {"text": "one moment"}}, + {"type": "done", "data": {}}, # runner omits stopReason on a HITL pause + {"type": "done", "data": {"stopReason": "paused"}}, # handler's correction + ] + ) + finish = parts[-1] + assert finish["type"] == "finish" + assert finish["finishReason"] == "other" + + +@pytest.mark.asyncio +async def test_normal_turn_finish_reason_unchanged() -> None: + """(b) A normal completion (a single ``done`` with ``stop``) still finishes ``stop``.""" + parts = await _collect( + [ + {"type": "message", "data": {"text": "here you go"}}, + {"type": "done", "data": {"stopReason": "stop"}}, + ] + ) + finish = parts[-1] + assert finish["type"] == "finish" + assert finish["finishReason"] == "stop" + + +@pytest.mark.asyncio +async def test_reasonless_correction_does_not_clobber_real_reason() -> None: + """(c) A ``done`` WITH a stopReason and no corrective terminal value is still honored: + a later null-carrying ``done`` must never clobber a real earlier one (fallback).""" + parts = await _collect( + [ + {"type": "message", "data": {"text": "done thinking"}}, + {"type": "done", "data": {"stopReason": "end_turn"}}, + {"type": "done", "data": {}}, # a stray reasonless done must not win + ] + ) + finish = parts[-1] + assert finish["type"] == "finish" + assert finish["finishReason"] == "stop" # end_turn -> stop + + +# --- the handler: surfaces the terminal stop_reason into the stream ----------- + + +class _FakeHarness: + """Duck-typed harness for ``agent_event_stream``: setup/stream/cleanup + a run + whose terminal result carries the authoritative stop_reason.""" + + def __init__(self, records: List[Dict[str, Any]]) -> None: + self._records = records + self.cleaned_up = False + + async def setup(self) -> None: + return None + + async def cleanup(self) -> None: + self.cleaned_up = True + + async def stream(self, config, messages) -> AgentStream: + async def _gen() -> AsyncIterator[Dict[str, Any]]: + for record in self._records: + yield record + + return AgentStream(_gen()) + + +def _paused_records(done_stop_reason: Optional[str], result_stop_reason: Optional[str]): + done_event: Dict[str, Any] = {"type": "done"} + if done_stop_reason is not None: + done_event["stopReason"] = done_stop_reason + result: Dict[str, Any] = {"ok": True, "output": "", "sessionId": "c1"} + if result_stop_reason is not None: + result["stopReason"] = result_stop_reason + return [ + {"kind": "event", "event": {"type": "message", "text": "one moment"}}, + {"kind": "event", "event": done_event}, + {"kind": "result", "result": result}, + ] + + +@pytest.mark.asyncio +async def test_handler_appends_corrective_terminal_done_on_pause() -> None: + """The runner's ``done`` has no stopReason but the terminal result says ``paused`` -> + the handler appends a corrective terminal ``done`` carrying ``paused``.""" + harness = _FakeHarness( + _paused_records(done_stop_reason=None, result_stop_reason="paused") + ) + out = [event async for event in agent_event_stream(harness, object(), [])] + + dones = [e for e in out if e["type"] == "done"] + assert len(dones) == 2 + assert dones[0]["data"].get("stopReason") is None # the runner's reasonless done + assert dones[-1]["data"]["stopReason"] == "paused" # the handler's correction + assert harness.cleaned_up is True + + +@pytest.mark.asyncio +async def test_handler_does_not_duplicate_done_when_result_agrees() -> None: + """A normal turn (done ``stop`` == terminal ``stop``) is not corrected: no duplicate done.""" + harness = _FakeHarness( + _paused_records(done_stop_reason="stop", result_stop_reason="stop") + ) + out = [event async for event in agent_event_stream(harness, object(), [])] + + dones = [e for e in out if e["type"] == "done"] + assert len(dones) == 1 + assert dones[0]["data"]["stopReason"] == "stop" + + +@pytest.mark.asyncio +async def test_handler_stream_then_adapter_carries_paused_end_to_end() -> None: + """End to end: the handler's neutral stream fed straight into the live adapter yields a + finish frame with the paused reason, proving the two halves compose.""" + harness = _FakeHarness( + _paused_records(done_stop_reason=None, result_stop_reason="paused") + ) + neutral = [event async for event in agent_event_stream(harness, object(), [])] + parts = await _collect(neutral) + + finish = parts[-1] + assert finish["type"] == "finish" + assert finish["finishReason"] == "other" + + +# Guard the DTO import surface the fakes lean on (Event/AgentResult stay importable here). +def test_dto_imports_present() -> None: + assert Event is not None and AgentResult is not None diff --git a/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_park.py b/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_park.py index a2c7ff277a..c9204c038e 100644 --- a/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_park.py +++ b/sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_park.py @@ -641,3 +641,77 @@ async def test_repeat_tool_call_refreshes_input_without_a_second_start() -> None # The refresh precedes the result — input is settled before output arrives. types = [p.get("type") for p in parts] assert types.index("tool-input-available") < types.index("tool-output-available") + + +# --------------------------------------------------------------------------- +# finishReason precedence: the terminal result's stop_reason is authoritative. +# +# The live runner settles paused-vs-ended AFTER the event stream closes, so its `done` +# event carries NO stopReason — only the terminal result record does. The finish frame must +# therefore prefer the terminal result's stop_reason over the `done` event's, mirroring the +# batch `fold(events, stop_reason=result.stop_reason)` precedence. (adapters/vercel/stream.py) +# --------------------------------------------------------------------------- + + +def _run(events, result) -> AgentStream: + records = [{"kind": "event", "event": e} for e in events] + records.append({"kind": "result", "result": result}) + return AgentStream(_records(records)) + + +@pytest.mark.asyncio +async def test_terminal_paused_wins_when_done_has_no_stop_reason() -> None: + """(a) A paused turn whose `done` event omits stopReason (the real live-runner shape) but + whose terminal result says ``paused`` -> the finish frame carries the paused reason + (mapped to the AI SDK ``other``).""" + run = _run( + events=[ + {"type": "message", "text": "one moment"}, + {"type": "done"}, # runner omits stopReason on a HITL pause + ], + result={"ok": True, "output": "", "stopReason": "paused", "sessionId": "c1"}, + ) + parts = [part async for part in agent_run_to_vercel_parts(run)] + finish = parts[-1] + assert finish["type"] == "finish" + assert finish["finishReason"] == "other" + + +@pytest.mark.asyncio +async def test_normal_turn_finish_reason_unchanged() -> None: + """(b) A normal completion (done + terminal both ``stop``) still finishes ``stop``.""" + run = _run( + events=[ + {"type": "message", "text": "here you go"}, + {"type": "done", "stopReason": "stop"}, + ], + result={ + "ok": True, + "output": "here you go", + "stopReason": "stop", + "sessionId": "c1", + }, + ) + parts = [part async for part in agent_run_to_vercel_parts(run)] + finish = parts[-1] + assert finish["type"] == "finish" + assert finish["finishReason"] == "stop" + + +@pytest.mark.asyncio +async def test_done_stop_reason_honored_when_terminal_has_none() -> None: + """(c) When the terminal result carries NO stop_reason, the `done` event's value is the + fallback and is still honored — terminal-wins only applies when the terminal value exists.""" + run = _run( + events=[ + {"type": "message", "text": "done thinking"}, + {"type": "done", "stopReason": "end_turn"}, + ], + # terminal result omits stopReason -> stop_reason is None + result={"ok": True, "output": "done thinking", "sessionId": "c1"}, + ) + parts = [part async for part in agent_run_to_vercel_parts(run)] + finish = parts[-1] + assert finish["type"] == "finish" + # end_turn maps to the AI SDK `stop`. + assert finish["finishReason"] == "stop"