fix(#4812): break live SSE loop on a terminal event even when deduped at cutoff

Codex round-3 finding #2 [CORE]: when reconciliation replayed the active run's
terminal event at the subscribe cutoff, the live queue's copy hit the cutoff/dedup
'continue' BEFORE the terminal 'break' check — so the handler never broke, stayed
blocked on the dead run's queue, and missed subsequent session runs. Check the
terminal event type before the already-delivered continue: a terminal always ends
the loop (after emitting if not already sent). Regression test asserts the handler
unsubscribes (doesn't hang) when the terminal is deduped. 24/24 SSE tests green.

Co-authored-by: rodboev <rodboev@users.noreply.github.com>

api/routes.py

@@ -16985,16 +16985,26 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id):
                     queued_event_id = STREAM_LAST_EVENT_ID.get(active_stream_id)
                 event_id = queued_event_id or STREAM_LAST_EVENT_ID.get(active_stream_id)
                 event_seq = _run_journal_same_run_seq(event_id, active_stream_id)
-                if replay_cutoff_seq is not None and event_seq is not None and event_seq <= replay_cutoff_seq:
-                    continue
-                if event_id and event_id in sent_event_ids:
+                _is_terminal = event in ("stream_end", "error", "cancel")
+                _already_sent = (
+                    (replay_cutoff_seq is not None and event_seq is not None and event_seq <= replay_cutoff_seq)
+                    or (event_id and event_id in sent_event_ids)
+                )
+                if _already_sent:
+                    # Already delivered via replay/reconciliation (cutoff or dedup).
+                    # A terminal event still has to end this loop — otherwise, when
+                    # reconciliation replayed the active run's terminal at the cutoff,
+                    # the live copy would be skipped here and the handler would stay
+                    # blocked on a dead run's queue and miss subsequent session runs.
+                    if _is_terminal:
+                        break
                     continue
                 if event_id:
                     _sse_with_id(handler, event, data, event_id)
                     note_sent_event_id(event_id)
                 else:
                     _sse(handler, event, data)
-                if event in ("stream_end", "error", "cancel"):
+                if _is_terminal:
                     break
         except _CLIENT_DISCONNECT_ERRORS:
             pass

tests/test_issue4812_session_sse_stream.py

@@ -577,6 +577,67 @@ def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch
     assert stream.unsubscribed is True
 
 
+def test_session_route_breaks_on_terminal_even_when_reconciliation_replayed_it(monkeypatch):
+    """CORE regression (#5677 gate r3): if reconciliation replays the active run's
+    TERMINAL event at the subscribe cutoff, the live queue's copy is deduped — but
+    the loop must STILL break on it and unsubscribe. Otherwise the handler stays
+    blocked on the dead run's queue and misses subsequent session runs."""
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            # The live queue re-delivers the terminal event that reconciliation
+            # already replayed at/below the cutoff (same event_id) — it must be
+            # deduped for output yet still end the loop.
+            self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:1"))
+            self.unsubscribed = False
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:1", "offline_buffered_events": 0}
+
+        def unsubscribe(self, q):
+            self.unsubscribed = q is self.q
+
+    stream = _FakeStream()
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_a, **_k: True)
+    monkeypatch.setattr(
+        routes,
+        "get_session",
+        lambda sid, metadata_only=False: SimpleNamespace(
+            session_id=sid,
+            compact=lambda **_kwargs: {"session_id": sid, "title": "Session"},
+        ),
+    )
+    monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_a, **_k: "run_active")
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    # Reconciliation replays the active run's terminal event (run_active:1) — same id
+    # as the queued live copy, at the cutoff.
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_a, **_k: {
+            "status": "ok",
+            "events": [
+                {"run_id": "run_active", "seq": 1, "event": "stream_end", "payload": {"status": "done"}, "event_id": "run_active:1"},
+            ],
+        },
+    )
+
+    handler = _FakeHandler()
+    # If the fix regresses, the loop never breaks and blocks on subscriber.get();
+    # the _stop_after_first_heartbeat safety turns an unexpected wait into a failure.
+    _stop_after_first_heartbeat(monkeypatch)
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_active:0"),
+        "session_1",
+    )
+
+    # The handler must have exited cleanly and unsubscribed — not hung on the queue.
+    assert stream.unsubscribed is True, "handler must unsubscribe after a terminal, even when deduped"
+
+
 def test_session_route_unsubscribes_when_replay_disconnects(monkeypatch):
     import api.routes as routes