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,
@@ -888,48 +914,113 @@ async def stream_run(
             )
 
     async def _event_iter() -> AsyncIterator[str]:
-        current_seq = after_seq
-        use_redis_executor = _uses_redis_executor()
+        # Single monotonic SSE cursor. Run-scoped lifecycle records (start/end)
+        # and thread-protocol events each carry their own independent ``seq``
+        # domains; mixing them directly yields non-monotonic ``id:`` values
+        # (e.g. ``1, 3, ...25, 2``) which breaks Last-Event-ID resume.
+        #
+        # The replay set (run lifecycle + thread protocol events + terminal end)
+        # is a fixed, ordered list: we assign each frame a deterministic cursor
+        # equal to its 1-based position in that list, so a reconnect with
+        # ``Last-Event-ID`` can skip exactly the frames already delivered and
+        # never replays them. Live frames (after replay) continue from the end
+        # of the list with the same monotonic counter.
+        replay_frames: list[tuple[int, str, str]] = []  # (run_seq_or_0, event_name, body)
+
         records_by_seq: dict[int, dict[str, object]] = {
-            seq: payload for seq, payload in await load_run_stream_events(run_id, after_seq=after_seq)
+            seq: payload for seq, payload in await load_run_stream_events(run_id, after_seq=0)
+        }
+        records_by_seq.update({seq: payload for seq, payload in run_broker.snapshot_records(run_id, after_seq=0)})
+        # 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"
         }
-        records_by_seq.update({seq: payload for seq, payload in run_broker.snapshot_records(run_id, after_seq=after_seq)})
         for seq in sorted(records_by_seq):
             event = records_by_seq[seq]
-            current_seq = max(current_seq, seq)
+            if str(event.get("event")) == "end":
+                continue
+            event_name = str(event.get("event", "message"))
+            event_payload: dict[str, object] = {"run_id": run_id, **event}
+            replay_frames.append((seq, event_name, safe_json_dumps(event_payload)))
+
+        # 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
+            replay_frames.append((0, str(event.get("method", "message")), safe_json_dumps(event.get("params", {}).get("data", {}))))
+
+        for seq in sorted(end_records):
+            event = end_records[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_frames.append((seq, event_name, safe_json_dumps(event_payload)))
+
+        # Replay frames are emitted with a deterministic cursor equal to their
+        # 1-based position in the ordered replay set. A reconnect with
+        # ``Last-Event-ID`` skips the already-delivered frames (``idx <=
+        # after_seq``) and never replays them; live frames after the replay set
+        # continue numbering from the end of the set.
+        emit_seq = len(replay_frames)
+        current_run_seq = 0
+
+        for idx, (run_seq, event_name, body) in enumerate(replay_frames, start=1):
+            if run_seq:
+                current_run_seq = max(current_run_seq, run_seq)
+            if idx <= after_seq:
+                continue
+            yield f"id: {idx}\nevent: {event_name}\ndata: {body}\n\n"
 
+        use_redis_executor = _uses_redis_executor()
         if use_redis_executor:
             async for item in iter_with_sse_keepalives(
                 _iter_persisted_run_records(
                     run_id=run_id,
                     thread_id=thread_id,
-                    after_seq=current_seq,
+                    after_seq=current_run_seq,
                 )
             ):
                 if item is None:
                     yield sse_keepalive_comment()
                     continue
                 seq, event = item
                 event_name = str(event.get("event", "message"))
-                event_payload = {"run_id": run_id, **event}
-                yield f"id: {seq}\nevent: {event_name}\ndata: {safe_json_dumps(event_payload)}\n\n"
+                event_payload: dict[str, object] = {"run_id": run_id, **event}
+                emit_seq += 1
+                if emit_seq <= after_seq:
+                    continue
+                yield f"id: {emit_seq}\nevent: {event_name}\ndata: {safe_json_dumps(event_payload)}\n\n"
             return
 
         if row.status in TERMINAL_RUN_STATUSES:
             return
 
-        async for item in iter_with_sse_keepalives(run_broker.stream_records(run_id, after_seq=current_seq)):
+        async for item in iter_with_sse_keepalives(run_broker.stream_records(run_id, after_seq=current_run_seq)):
             if item is None:
                 yield sse_keepalive_comment()
                 continue
             seq, event = item
             event_name = str(event.get("event", "message"))
-            event_payload = {"run_id": run_id, **event}
-            yield f"id: {seq}\nevent: {event_name}\ndata: {safe_json_dumps(event_payload)}\n\n"
+            event_payload: dict[str, object] = {"run_id": run_id, **event}
+            emit_seq += 1
+            if emit_seq <= after_seq:
+                continue
+            yield f"id: {emit_seq}\nevent: {event_name}\ndata: {safe_json_dumps(event_payload)}\n\n"
 
     return StreamingResponse(
         _event_iter(),

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)