[fix] Paused live streams carry the terminal stop reason in the finish frame

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

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)

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

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"