fix(streaming): default to official astream output, fix duplicated Send parallel updates, restore tool/replay/interrupt

scripts/verify_docker_api.py

@@ -234,8 +234,13 @@ def _assert_common_flow(base_url: str) -> None:
     assert isinstance(stream_body, str)
     assert "text/event-stream" in stream_content_type
     payloads = _stream_payloads(stream_body)
-    assert any(payload["event"] == "start" for payload in payloads)
-    assert any(payload["event"] == "end" and payload.get("status") == "success" for payload in payloads)
+    assert any(isinstance(payload, dict) and payload.get("event") == "start" for payload in payloads)
+    assert any(
+        isinstance(payload, dict)
+        and payload.get("event") == "end"
+        and payload.get("status") == "success"
+        for payload in payloads
+    )
 
     _, stateless_run, _ = _request(
         base_url=base_url,
@@ -326,7 +331,7 @@ def _assert_common_flow(base_url: str) -> None:
     end_statuses = {
         payload["status"]
         for payload in _stream_payloads(resumed_stream)
-        if payload["event"] == "end"
+        if isinstance(payload, dict) and payload.get("event") == "end"
     }
     assert "success" in end_statuses
     stress_waited, _ = _assert_sample_run(
@@ -358,7 +363,12 @@ def _assert_common_flow(base_url: str) -> None:
     react_output = react_waited["output"]
     assert isinstance(react_output, dict)
     assert "42" in str(react_output["final_text"])
-    assert any(payload["event"] == "tool_start" and payload["name"] == "lookup" for payload in react_payloads)
+    assert any(
+        isinstance(payload, dict)
+        and payload.get("event") == "tool-started"
+        and payload.get("tool_name") == "lookup"
+        for payload in react_payloads
+    )
 
     stress_tool_waited, stress_tool_payloads = _assert_sample_run(
         base_url=base_url,
@@ -372,7 +382,11 @@ def _assert_common_flow(base_url: str) -> None:
     tool_messages = [message for message in stress_tool_output["transcript"] if message["type"] == "ToolMessage"]
     assert len(tool_messages) == 3
     tool_starts = [
-        payload for payload in stress_tool_payloads if payload["event"] == "tool_start" and payload["name"] == "slow_process"
+        payload
+        for payload in stress_tool_payloads
+        if isinstance(payload, dict)
+        and payload.get("event") == "tool-started"
+        and payload.get("tool_name") == "slow_process"
     ]
     assert len(tool_starts) == 3
 
@@ -501,7 +515,7 @@ def _assert_resume_check(base_url: str, *, thread_id: str, run_id: str, resume:
     end_statuses = [
         payload["status"]
         for payload in _stream_payloads(run_stream)
-        if payload["event"] == "end"
+        if isinstance(payload, dict) and payload.get("event") == "end"
     ]
     assert "interrupted" in end_statuses
     assert "success" in end_statuses

src/agentseek_api/api/runs.py

@@ -324,6 +324,14 @@ def _protocol_channels_for_stream_modes(stream_modes: list[str]) -> list[str]:
     return channels
 
 
+# Channels replayed by the legacy GET /threads/{id}/runs/{run_id}/stream when no
+# ``stream_mode`` query is given. The astream migration no longer publishes
+# run-scoped stream events, so the run's protocol-v2 thread events (values /
+# updates / messages / tools / custom) are replayed instead, alongside the
+# run-scoped lifecycle records (start/end) still published by run_jobs.
+DEFAULT_RUN_STREAM_REPLAY_CHANNELS = ["values", "updates", "messages", "tools", "custom", "input"]
+
+
 async def _iter_persisted_protocol_run_events(
     *,
     thread_id: str,
@@ -400,6 +408,7 @@ def _is_block_message_event(event: dict[str, Any]) -> bool:
     async def _event_iter() -> AsyncIterator[str]:
         try:
             current_seq = after_seq
+            saw_interrupt = False
             if include_metadata:
                 yield _protocol_event_sse(event_name="metadata", data={"run_id": created.run_id, "attempt": 1})
 
@@ -417,10 +426,13 @@ async def _event_iter() -> AsyncIterator[str]:
                     continue
                 if suppress_block_messages and _is_block_message_event(event):
                     continue
+                event_data = event.get("params", {}).get("data", {})
+                if isinstance(event_data, dict) and "__interrupt__" in event_data:
+                    saw_interrupt = True
                 yield _protocol_event_sse(
                     seq=current_seq,
                     event_name=str(event.get("method", "message")),
-                    data=event.get("params", {}).get("data", {}),
+                    data=event_data,
                 )
 
             if _uses_redis_executor():
@@ -438,10 +450,13 @@ async def _event_iter() -> AsyncIterator[str]:
                     current_seq = max(current_seq, int(event.get("seq", 0)))
                     if suppress_block_messages and _is_block_message_event(event):
                         continue
+                    event_data = event.get("params", {}).get("data", {})
+                    if isinstance(event_data, dict) and "__interrupt__" in event_data:
+                        saw_interrupt = True
                     yield _protocol_event_sse(
                         seq=current_seq,
                         event_name=str(event.get("method", "message")),
-                        data=event.get("params", {}).get("data", {}),
+                        data=event_data,
                     )
             else:
                 async for event in iter_with_sse_keepalives(
@@ -462,10 +477,13 @@ async def _event_iter() -> AsyncIterator[str]:
                     current_seq = max(current_seq, int(event.get("seq", 0)))
                     if suppress_block_messages and _is_block_message_event(event):
                         continue
+                    event_data = event.get("params", {}).get("data", {})
+                    if isinstance(event_data, dict) and "__interrupt__" in event_data:
+                        saw_interrupt = True
                     yield _protocol_event_sse(
                         seq=current_seq,
                         event_name=str(event.get("method", "message")),
-                        data=event.get("params", {}).get("data", {}),
+                        data=event_data,
                     )
 
             final_run = (
@@ -482,7 +500,15 @@ async def _event_iter() -> AsyncIterator[str]:
                 )
                 return
             interrupt_event = _interrupt_stream_event_name(stream_modes)
-            if final_run.status == "interrupted" and final_run.interrupts and interrupt_event is not None:
+            if (
+                final_run.status == "interrupted"
+                and final_run.interrupts
+                and interrupt_event is not None
+                # The interrupt is delivered in-stream (values/updates carrying
+                # ``__interrupt__``); only emit a trailing event when this
+                # connection never saw it.
+                and not saw_interrupt
+            ):
                 current_seq += 1
                 yield _protocol_event_sse(
                     seq=current_seq,
@@ -894,8 +920,51 @@ async def _event_iter() -> AsyncIterator[str]:
             seq: payload for seq, payload in await load_run_stream_events(run_id, after_seq=after_seq)
         }
         records_by_seq.update({seq: payload for seq, payload in run_broker.snapshot_records(run_id, after_seq=after_seq)})
+        # Run-scoped lifecycle records (start/end) from run_jobs. The terminal
+        # "end" record is deferred until after the protocol thread events so
+        # the stream ends with the run's terminal status.
+        end_records: dict[int, dict[str, object]] = {
+            seq: event
+            for seq, event in records_by_seq.items()
+            if str(event.get("event")) == "end"
+        }
         for seq in sorted(records_by_seq):
             event = records_by_seq[seq]
+            if str(event.get("event")) == "end":
+                continue
+            current_seq = max(current_seq, seq)
+            event_name = str(event.get("event", "message"))
+            event_payload: dict[str, object] = {"run_id": run_id, **event}
+            payload = safe_json_dumps(event_payload)
+            yield f"id: {seq}\nevent: {event_name}\ndata: {payload}\n\n"
+
+        # Replay the run's protocol-v2 thread events so the default endpoint
+        # still returns the full stream (run-scoped stream events are no longer
+        # published by the astream migration).
+        try:
+            thread_events = await load_thread_stream_events(
+                thread_id,
+                channels=DEFAULT_RUN_STREAM_REPLAY_CHANNELS,
+                namespaces=None,
+                depth=None,
+                after_seq=0,
+            )
+        except Exception:  # noqa: BLE001 - replay is best-effort
+            thread_events = []
+        for event in thread_events:
+            if event.get("params", {}).get("run_id") != run_id:
+                continue
+            event_seq = int(event.get("seq", 0) or 0)
+            if event_seq <= after_seq:
+                continue
+            yield _protocol_event_sse(
+                seq=event_seq,
+                event_name=str(event.get("method", "message")),
+                data=event.get("params", {}).get("data", {}),
+            )
+
+        for seq in sorted(end_records):
+            event = end_records[seq]
             current_seq = max(current_seq, seq)
             event_name = str(event.get("event", "message"))
             event_payload: dict[str, object] = {"run_id": run_id, **event}

src/agentseek_api/api/streaming.py

@@ -11,6 +11,7 @@
 from agentseek_api.core.orm import Run, Thread
 from agentseek_api.models.auth import User
 from agentseek_api.models.protocol import ProtocolCommandRequest, ProtocolEventStreamRequest
+from agentseek_api.services.stream_modes import normalize_stream_modes
 from agentseek_api.services.run_preparation import (
     ActiveThreadRunConflictError,
     prepare_and_submit_run,
@@ -134,11 +135,18 @@ async def handle_protocol_command(
             )
 
         try:
+            run_kwargs: dict[str, Any] | None = None
+            if payload.params.get("stream_mode") is not None:
+                run_kwargs = {"stream_modes": normalize_stream_modes(payload.params.get("stream_mode"))}
+            if payload.params.get("stream_subgraphs"):
+                run_kwargs = run_kwargs or {}
+                run_kwargs["stream_subgraphs"] = True
             run = await prepare_and_submit_run(
                 thread_id=thread_id,
                 assistant_id=assistant_id,
                 payload=_coerce_protocol_input(payload.params.get("input")),
                 user=user,
+                kwargs=run_kwargs,
             )
         except ValueError as exc:
             return _protocol_error(request_id=payload.id, code="invalid_argument", message=str(exc), status_code=404)