Merge #5677: resumable session SSE stream (#4812)

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

api/routes.py

@@ -28,7 +28,7 @@ import time
 import uuid
 import http.client
 import socket as _socket
-from collections import defaultdict
+from collections import defaultdict, deque
 from pathlib import Path
 from contextlib import closing
 from urllib.parse import parse_qs, quote, unquote, urljoin, urlsplit
@@ -1042,6 +1042,7 @@ def _skill_view_from_active_dir(name: str) -> dict:
 # Trivial; many production SSE deployments run 5-15s heartbeats specifically
 # to handle proxies and mobile NAT.
 _SSE_HEARTBEAT_INTERVAL_SECONDS = 5
+_SESSION_SSE_SENT_EVENT_ID_LIMIT = 4096
 
 
 def _normalize_messaging_source(raw_source) -> str:
@@ -9254,8 +9255,10 @@ from api.streaming import (
 )
 from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_enabled
 from api.run_journal import (
+    _parse_run_journal_event_id as _shared_parse_run_journal_event_id,
     find_run_summary,
     read_run_events,
+    read_session_run_events,
     stale_interrupted_event,
 )
 from api.todo_state import attach_todo_state
@@ -12900,6 +12903,10 @@ def handle_get(handler, parsed) -> bool:
     if parsed.path == '/api/sessions/events':
         return _handle_session_events_stream(handler)
 
+    session_events_session_id = _session_events_path_session_id(parsed.path)
+    if session_events_session_id is not None:
+        return _handle_session_sse_stream_for_session(handler, parsed, session_events_session_id)
+
     if parsed.path == "/api/media":
         return _handle_media(handler, parsed)
 
@@ -16399,19 +16406,46 @@ def _sse_with_id(handler, event, data, event_id=None):
     _sse(handler, event, data)
 
 
-def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]:
+def _session_events_path_session_id(path: str | None) -> str | None:
+    path = str(path or "")
+    parts = path.strip("/").split("/")
+    if len(parts) != 4:
+        return None
+    if parts[0] != "api" or parts[1] != "sessions" or parts[3] != "events":
+        return None
+    sid = str(parts[2] or "").strip()
+    return sid or None
+
+
+def _session_events_resume_event_id(handler, parsed) -> str | None:
+    headers = getattr(handler, "headers", None)
+    raw = None
+    if headers is not None:
+        try:
+            raw = headers.get("Last-Event-ID")
+        except Exception:
+            raw = None
     raw = str(raw or "").strip()
-    if not raw:
-        return None, None
-    if ":" in raw:
-        run_id, tail = raw.rsplit(":", 1)
-    else:
-        run_id, tail = None, raw
+    if raw:
+        return raw
+    qs = parse_qs(getattr(parsed, "query", "") or "")
+    raw = str(qs.get("after_event_id", [None])[0] or "").strip()
+    return raw or None
+
+
+def _session_snapshot_payload(session, *, active_stream_id: str | None = None) -> dict:
     try:
-        seq = max(0, int(tail))
-    except (TypeError, ValueError):
-        return run_id or None, None
-    return run_id or None, seq
+        payload = session.compact(
+            include_runtime=bool(active_stream_id),
+            active_stream_ids={active_stream_id} if active_stream_id else None,
+        )
+    except Exception:
+        payload = {"session_id": str(getattr(session, "session_id", "") or "")}
+    return {"session": payload}
+
+
+def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]:
+    return _shared_parse_run_journal_event_id(raw)
 
 
 def _parse_run_journal_after_seq(qs: dict, stream_id: str | None = None) -> int | None:
@@ -16825,6 +16859,142 @@ def _handle_sse_stream(handler, parsed):
     return True
 
 
+def _handle_session_run_journal_stream_for_session(handler, parsed, session_id):
+    if not _session_id_visible_to_request_profile(handler, session_id):
+        return True
+    try:
+        session = get_session(session_id, metadata_only=True)
+    except KeyError:
+        return j(handler, {"error": "Session not found"}, status=404)
+
+    handler.send_response(200)
+    handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
+    handler.send_header("Cache-Control", "no-cache")
+    handler.send_header("X-Accel-Buffering", "no")
+    # #3103: see _handle_gateway_sse_stream — `Connection: close` causes
+    # EventSource reconnect storms in browsers on long-lived SSE.
+    end_sse_headers(handler)
+    _sse_set_write_deadline(handler)
+
+    resume_event_id = _session_events_resume_event_id(handler, parsed)
+    active_stream_id = _active_run_stream_for_session(session_id)
+    subscriber = None
+    subscriber_stream = None
+    replay_cutoff_seq = None
+    sent_event_ids: set[str] = set()
+    sent_event_order = deque()
+
+    def note_sent_event_id(event_id):
+        if not event_id:
+            return
+        sent_event_ids.add(event_id)
+        sent_event_order.append(event_id)
+        while len(sent_event_order) > _SESSION_SSE_SENT_EVENT_ID_LIMIT:
+            sent_event_ids.discard(sent_event_order.popleft())
+
+    def attach_active_stream():
+        stream_id = _active_run_stream_for_session(session_id)
+        stream = STREAMS.get(stream_id) if stream_id else None
+        if stream is None:
+            return None, None, None, stream_id
+        if hasattr(stream, "subscribe_with_snapshot"):
+            queue_, snapshot = stream.subscribe_with_snapshot()
+        else:
+            queue_ = stream.subscribe() if hasattr(stream, "subscribe") else stream
+            snapshot = {}
+        return queue_, stream, snapshot, stream_id
+
+    def emit_replay(events, stream_id, cutoff_seq):
+        for entry in events:
+            event_id = str(entry.get("event_id") or "")
+            event_seq = _run_journal_same_run_seq(event_id, stream_id)
+            if cutoff_seq is not None and event_seq is not None and event_seq > cutoff_seq:
+                continue
+            if event_id and event_id in sent_event_ids:
+                continue
+            _sse_with_id(handler, entry.get("event") or entry.get("type") or "message", entry.get("payload"), event_id)
+            if event_id:
+                note_sent_event_id(event_id)
+
+    def emit_session_snapshot(active_stream_id):
+        try:
+            fresh_session = get_session(session_id, metadata_only=True)
+        except KeyError:
+            fresh_session = session
+        _sse(handler, "session_snapshot", _session_snapshot_payload(fresh_session, active_stream_id=active_stream_id))
+
+    try:
+        replay_events = []
+        replay_ok = False
+        if resume_event_id:
+            replay = read_session_run_events(session_id, after_event_id=resume_event_id)
+            if replay.get("status") != "ok":
+                emit_session_snapshot(active_stream_id)
+            else:
+                replay_ok = True
+                replay_events = replay.get("events") or []
+        subscriber, subscriber_stream, stream_snapshot, active_stream_id = attach_active_stream()
+        if subscriber is None:
+            if replay_ok:
+                emit_replay(replay_events, active_stream_id, None)
+            while True:
+                subscriber, subscriber_stream, stream_snapshot, active_stream_id = attach_active_stream()
+                if subscriber is not None:
+                    break
+                handler.wfile.write(b": keepalive\n\n")
+                handler.wfile.flush()
+                time.sleep(_SSE_HEARTBEAT_INTERVAL_SECONDS)
+        if subscriber is None:
+            return True
+        if replay_ok:
+            replay_cutoff_seq = _run_journal_same_run_seq(str(stream_snapshot.get("last_event_id") or ""), active_stream_id)
+            reconciled = read_session_run_events(session_id, after_event_id=resume_event_id)
+            if reconciled.get("status") == "ok":
+                emit_replay(reconciled.get("events") or [], active_stream_id, replay_cutoff_seq)
+            else:
+                emit_session_snapshot(active_stream_id)
+        try:
+            while True:
+                try:
+                    item = subscriber.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS)
+                except queue.Empty:
+                    handler.wfile.write(b": keepalive\n\n")
+                    handler.wfile.flush()
+                    continue
+                if len(item) >= 3:
+                    event, data, queued_event_id = item[0], item[1], item[2]
+                else:
+                    event, data = item
+                    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:
+                    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"):
+                    break
+        except _CLIENT_DISCONNECT_ERRORS:
+            pass
+    except _CLIENT_DISCONNECT_ERRORS:
+        pass
+    finally:
+        if subscriber is not None and subscriber is not subscriber_stream and hasattr(subscriber_stream, "unsubscribe"):
+            try:
+                subscriber_stream.unsubscribe(subscriber)
+            except Exception:
+                pass
+    return True
+
+
+_handle_session_sse_stream_for_session = _handle_session_run_journal_stream_for_session
+
+
 def _terminal_session_lookup(body_or_query):
     sid = str(body_or_query.get("session_id", "")).strip()
     if not sid:

api/run_journal.py

@@ -31,6 +31,9 @@ _TERMINAL_SSE_EVENTS = {"done", "cancel", "apperror", "error", "stream_end"}
 _FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC"
 _FSYNC_MODE_EAGER = "eager"
 _FSYNC_MODE_TERMINAL_ONLY = "terminal-only"
+_SESSION_REPLAY_MAX_BYTES = 4 * 1024 * 1024
+_SESSION_REPLAY_MAX_ROWS = 4096
+_SESSION_REPLAY_READ_CHUNK_BYTES = 64 * 1024
 
 
 def _default_session_dir() -> Path:
@@ -85,6 +88,21 @@ def _read_jsonl(path: Path) -> tuple[list[dict], list[dict]]:
     return events, malformed
 
 
+def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]:
+    raw = str(raw or "").strip()
+    if not raw:
+        return None, None
+    if ":" in raw:
+        run_id, tail = raw.rsplit(":", 1)
+    else:
+        run_id, tail = None, raw
+    try:
+        seq = max(0, int(tail))
+    except (TypeError, ValueError):
+        return run_id or None, None
+    return run_id or None, seq
+
+
 def _next_seq(path: Path) -> int:
     events, _malformed = _read_jsonl(path)
     seqs = [int(event.get("seq") or 0) for event in events if isinstance(event.get("seq"), int)]
@@ -181,6 +199,49 @@ def _fsync_parent_dir(path: Path) -> None:
         pass
 
 
+def _event_created_at(event: dict, *, fallback: float = 0.0) -> float:
+    try:
+        return float(event.get("created_at") or fallback)
+    except (TypeError, ValueError):
+        return fallback
+
+
+def _iter_bounded_raw_jsonl_lines(path: Path, *, max_bytes: int, retained_bytes: int = 0):
+    line_no = 0
+    buffered = bytearray()
+    total_bytes = int(retained_bytes)
+    try:
+        with path.open("rb") as fh:
+            while True:
+                chunk = fh.read(_SESSION_REPLAY_READ_CHUNK_BYTES)
+                if not chunk:
+                    if buffered:
+                        if total_bytes + len(buffered) > max_bytes:
+                            raise ValueError("replay_limit_bytes")
+                        line_no += 1
+                        total_bytes += len(buffered)
+                        yield line_no, bytes(buffered), total_bytes
+                    return
+                start = 0
+                while start < len(chunk):
+                    newline = chunk.find(b"\n", start)
+                    if newline == -1:
+                        buffered.extend(chunk[start:])
+                        if total_bytes + len(buffered) > max_bytes:
+                            raise ValueError("replay_limit_bytes")
+                        break
+                    buffered.extend(chunk[start : newline + 1])
+                    if total_bytes + len(buffered) > max_bytes:
+                        raise ValueError("replay_limit_bytes")
+                    line_no += 1
+                    total_bytes += len(buffered)
+                    yield line_no, bytes(buffered), total_bytes
+                    buffered.clear()
+                    start = newline + 1
+    except FileNotFoundError:
+        return
+
+
 def append_run_event(
     session_id: str,
     run_id: str,
@@ -319,6 +380,122 @@ def find_run_summary(run_id: str, *, session_dir: Path | None = None) -> dict |
     return None
 
 
+def read_session_run_events(
+    session_id: str,
+    *,
+    after_event_id: str | None = None,
+    session_dir: Path | None = None,
+    max_bytes: int = _SESSION_REPLAY_MAX_BYTES,
+    max_rows: int = _SESSION_REPLAY_MAX_ROWS,
+) -> dict:
+    """Replay durable run-journal rows for one session after an opaque cursor."""
+    sid = _validate_id(session_id, "session_id")
+    cursor_run_id, cursor_seq = _parse_run_journal_event_id(after_event_id)
+    raw_cursor = str(after_event_id or "").strip()
+    if raw_cursor and cursor_run_id is not None:
+        try:
+            cursor_run_id = _validate_id(cursor_run_id, "run_id")
+        except ValueError:
+            cursor_seq = None
+    if raw_cursor:
+        try:
+            if int(raw_cursor.rsplit(":", 1)[-1]) < 0:
+                cursor_seq = None
+        except (TypeError, ValueError):
+            pass
+    if raw_cursor and (cursor_run_id is None or cursor_seq is None or cursor_seq <= 0):
+        return {
+            "session_id": sid,
+            "cursor_run_id": cursor_run_id,
+            "cursor_seq": cursor_seq,
+            "status": "cursor_invalid",
+            "events": [],
+        }
+    if not raw_cursor:
+        return {
+            "session_id": sid,
+            "cursor_run_id": None,
+            "cursor_seq": None,
+            "status": "ok",
+            "events": [],
+        }
+    root = Path(session_dir) if session_dir is not None else _default_session_dir()
+    session_root = root / RUN_JOURNAL_DIR_NAME / sid
+    runs: list[tuple[float, str, list[dict]]] = []
+    retained_rows = 0
+    retained_bytes = 0
+    for path in sorted(session_root.glob("*.jsonl")) if session_root.exists() else []:
+        run_id = path.stem
+        try:
+            run_id = _validate_id(run_id, "run_id")
+        except ValueError:
+            continue
+        events: list[dict] = []
+        expected_seq = 1
+        try:
+            for _line_no, raw, total_bytes in _iter_bounded_raw_jsonl_lines(
+                path,
+                max_bytes=max_bytes,
+                retained_bytes=retained_bytes,
+            ):
+                retained_bytes = total_bytes
+                if not raw.strip():
+                    continue
+                try:
+                    event = json.loads(raw.decode("utf-8"))
+                    seq = int(event.get("seq")) if isinstance(event, dict) else 0
+                except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError):
+                    return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_malformed", "events": []}
+                if (
+                    seq != expected_seq
+                    or event.get("event_id") != f"{run_id}:{seq}"
+                    or event.get("run_id") != run_id
+                    or event.get("session_id") != sid
+                ):
+                    return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_noncontiguous", "events": []}
+                expected_seq += 1
+                retained_rows += 1
+                if retained_rows > max_rows:
+                    return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_rows", "events": []}
+                events.append(event)
+        except FileNotFoundError:
+            continue
+        except ValueError as exc:
+            if str(exc) == "replay_limit_bytes":
+                return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_bytes", "events": []}
+            raise
+        created_at = min((_event_created_at(event) for event in events), default=path.stat().st_mtime)
+        runs.append((created_at, run_id, events))
+    runs.sort(key=lambda run: (run[0], run[1]))
+    cursor_index = next((index for index, (_created_at, run_id, _events) in enumerate(runs) if run_id == cursor_run_id), None)
+    if cursor_index is None:
+        foreign_paths = root.joinpath(RUN_JOURNAL_DIR_NAME).glob(f"*/{cursor_run_id}.jsonl") if cursor_run_id else []
+        foreign_session_id = next((path.parent.name for path in foreign_paths if path.parent.name != sid), "")
+        status = "cursor_run_missing"
+        if foreign_session_id:
+            status = "cursor_session_mismatch"
+        return {
+            "session_id": sid,
+            "cursor_run_id": cursor_run_id,
+            "cursor_seq": cursor_seq,
+            "status": status,
+            "events": [],
+        }
+    cursor_events = runs[cursor_index][2]
+    if cursor_seq is None or cursor_seq > len(cursor_events):
+        return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "cursor_event_missing", "events": []}
+    replay_events = [event for event in cursor_events if event["seq"] > cursor_seq]
+    for _created_at, _run_id, events in runs[cursor_index + 1:]:
+        replay_events.extend(events)
+    return {
+        "session_id": sid,
+        "cursor_run_id": cursor_run_id,
+        "cursor_seq": cursor_seq,
+        "status": "ok",
+        "events": replay_events,
+    }
+
+
 def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> bool:
     """Remove the entire per-session run-journal directory (``_run_journal/{sid}/``).
 

docs/CONTRACTS.md

@@ -66,7 +66,9 @@ contributor guidance; it does not change runtime behavior or CI gates.
   stream `GET /api/sessions/{session_id}/events` (#4812). Distinct from the
   existing global session-list stream `GET /api/sessions/events`. Start here for
   any work that touches per-session SSE, `Last-Event-ID` replay, or session
-  lifecycle event delivery. Proposed RFCs do not authorize implementation.
+  lifecycle event delivery. The Phase 1 server contract is now shipped; the RFC
+  remains the vocabulary reference while broader client and platform claims stay
+  behind the recorded proof gates.
 
 When a change touches streaming, recovery, replay, compression, context
 reconstruction, cancellation, approval/clarify, session metadata, or run state,

tests/test_issue4812_session_sse_stream.py

@@ -0,0 +1,675 @@
+from __future__ import annotations
+
+import io
+import queue
+from types import SimpleNamespace
+from urllib.parse import urlparse
+
+from api.run_journal import append_run_event, read_session_run_events
+
+
+class _FakeHandler:
+    def __init__(self):
+        self.status = None
+        self.headers = {}
+        self.wfile = io.BytesIO()
+        self.command = "GET"
+        self.path = "/"
+        self.client_address = ("127.0.0.1", 12345)
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, key, value):
+        self.headers[key] = value
+
+    def end_headers(self):
+        pass
+
+
+def _capture(monkeypatch):
+    cap = {}
+
+    def _j(_h, obj, *_, **kwargs):
+        cap["ok"] = obj
+        cap["status"] = kwargs.get("status", 200)
+        return True
+
+    def _bad(_h, msg, code=400):
+        cap["bad"] = (msg, code)
+        return True
+
+    monkeypatch.setattr("api.routes.j", _j)
+    monkeypatch.setattr("api.routes.bad", _bad)
+    return cap
+
+
+def _stop_after_first_heartbeat(monkeypatch):
+    calls = {"count": 0}
+
+    def _sleep(_seconds):
+        calls["count"] += 1
+        raise BrokenPipeError("stop after the first heartbeat")
+
+    monkeypatch.setattr("api.routes.time.sleep", _sleep)
+    return calls
+
+
+def test_session_route_and_global_route_stay_separate(monkeypatch):
+    import api.routes as routes
+
+    calls = {"global": 0, "session": 0}
+
+    monkeypatch.setattr(routes, "_handle_extension_sidecar_proxy", lambda *_args, **_kwargs: False)
+    monkeypatch.setattr(
+        routes,
+        "_handle_session_events_stream",
+        lambda *_args, **_kwargs: calls.__setitem__("global", calls["global"] + 1) or True,
+    )
+    monkeypatch.setattr(
+        routes,
+        "_handle_session_sse_stream_for_session",
+        lambda *_args, **_kwargs: calls.__setitem__("session", calls["session"] + 1) or True,
+    )
+
+    assert routes._session_events_path_session_id("/api/sessions/events") is None
+    assert routes._session_events_path_session_id("/api/sessions/session_1/events") == "session_1"
+    assert routes._session_events_path_session_id("/api/sessions/session_1/events/extra") is None
+
+    handler = _FakeHandler()
+    routes.handle_get(handler, urlparse("/api/sessions/events"))
+    routes.handle_get(handler, urlparse("/api/sessions/session_1/events"))
+
+    assert calls == {"global": 1, "session": 1}
+
+
+def test_session_journal_replays_later_runs_after_cursor(tmp_path):
+    append_run_event(
+        "session_1",
+        "run_a",
+        "token",
+        {"text": "a1"},
+        session_dir=tmp_path,
+        seq=1,
+        created_at=100.0,
+    )
+    append_run_event(
+        "session_1",
+        "run_a",
+        "token",
+        {"text": "a2"},
+        session_dir=tmp_path,
+        seq=2,
+        created_at=101.0,
+    )
+    append_run_event(
+        "session_1",
+        "run_b",
+        "token",
+        {"text": "b1"},
+        session_dir=tmp_path,
+        seq=1,
+        created_at=200.0,
+    )
+    append_run_event(
+        "session_1",
+        "run_b",
+        "done",
+        {"session": {"session_id": "session_1"}},
+        session_dir=tmp_path,
+        seq=2,
+        created_at=201.0,
+    )
+    append_run_event(
+        "session_2",
+        "run_other",
+        "token",
+        {"text": "other"},
+        session_dir=tmp_path,
+        seq=1,
+        created_at=150.0,
+    )
+
+    replay = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path)
+
+    assert replay["status"] == "ok"
+    assert [event["event_id"] for event in replay["events"]] == ["run_a:2", "run_b:1", "run_b:2"]
+
+
+def test_session_journal_distinguishes_missing_and_foreign_cursor(tmp_path):
+    append_run_event(
+        "session_1",
+        "run_a",
+        "token",
+        {"text": "a1"},
+        session_dir=tmp_path,
+        seq=1,
+        created_at=100.0,
+    )
+    append_run_event(
+        "session_2",
+        "run_b",
+        "token",
+        {"text": "b1"},
+        session_dir=tmp_path,
+        seq=1,
+        created_at=200.0,
+    )
+
+    missing = read_session_run_events("session_1", after_event_id="run_missing:1", session_dir=tmp_path)
+    foreign = read_session_run_events("session_1", after_event_id="run_b:1", session_dir=tmp_path)
+
+    assert missing["status"] == "cursor_run_missing"
+    assert foreign["status"] == "cursor_session_mismatch"
+
+
+def test_session_journal_rejects_hostile_cursor_without_replay(tmp_path, monkeypatch):
+    append_run_event(
+        "session_1",
+        "run_a",
+        "token",
+        {"text": "a1"},
+        session_dir=tmp_path,
+        seq=1,
+        created_at=100.0,
+    )
+    monkeypatch.setattr(
+        "api.run_journal.find_run_summary",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("invalid cursor should not scan summaries")),
+    )
+
+    malformed = read_session_run_events("session_1", after_event_id="bad/run:1", session_dir=tmp_path)
+    negative = read_session_run_events("session_1", after_event_id="run_a:-5", session_dir=tmp_path)
+    zero = read_session_run_events("session_1", after_event_id="run_a:0", session_dir=tmp_path)
+
+    assert malformed["status"] == "cursor_invalid"
+    assert malformed["events"] == []
+    assert negative["status"] == "cursor_invalid"
+    assert negative["events"] == []
+    assert zero["status"] == "cursor_invalid"
+    assert zero["events"] == []
+
+
+def test_session_journal_blank_summary_session_stays_missing(tmp_path, monkeypatch):
+    monkeypatch.setattr("api.run_journal.find_run_summary", lambda *_args, **_kwargs: {"session_id": ""})
+
+    replay = read_session_run_events("session_1", after_event_id="run_missing:1", session_dir=tmp_path)
+
+    assert replay["status"] == "cursor_run_missing"
+
+
+def test_session_journal_rejects_bounds_and_invalid_contiguous_rows(tmp_path):
+    append_run_event("session_1", "run_a", "token", {"text": "one"}, session_dir=tmp_path, seq=1)
+    append_run_event("session_1", "run_a", "token", {"text": "two"}, session_dir=tmp_path, seq=2)
+
+    rows_limited = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_rows=1)
+    bytes_limited = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=1)
+    assert rows_limited["status"] != "ok" and rows_limited["events"] == []
+    assert bytes_limited["status"] != "ok" and bytes_limited["events"] == []
+
+    path = tmp_path / "_run_journal" / "session_1" / "run_a.jsonl"
+    path.write_text(path.read_text(encoding="utf-8").replace('"event_id":"run_a:2"', '"event_id":"wrong:2"'), encoding="utf-8")
+    invalid = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path)
+    assert invalid["status"] == "replay_noncontiguous"
+    assert invalid["events"] == []
+
+
+def test_session_journal_rejects_oversized_line_before_decode(tmp_path, monkeypatch):
+    path = tmp_path / "_run_journal" / "session_1" / "run_a.jsonl"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = '{"event_id":"run_a:1","seq":1,"run_id":"run_a","session_id":"session_1","event":"done","type":"done","payload":{"text":"' + ("x" * 5000) + '"}}\n'
+    path.write_bytes(payload.encode("utf-8"))
+    monkeypatch.setattr("api.run_journal.json.loads", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("oversized line should not decode")))
+
+    limited = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=1024)
+
+    assert limited["status"] == "replay_limit_bytes"
+    assert limited["events"] == []
+
+
+def test_session_journal_accepts_line_at_cap_and_rejects_next_byte(tmp_path):
+    path = tmp_path / "_run_journal" / "session_1" / "run_a.jsonl"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = '{"event_id":"run_a:1","seq":1,"run_id":"run_a","session_id":"session_1","event":"done","type":"done","payload":{"text":"ok"}}\n'
+    path.write_bytes(payload.encode("utf-8"))
+    cap = len(payload.encode("utf-8"))
+
+    at_cap = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=cap)
+    over = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=cap - 1)
+
+    assert at_cap["status"] == "ok"
+    assert at_cap["events"] == []
+    assert over["status"] == "replay_limit_bytes"
+    assert over["events"] == []
+
+
+def test_session_journal_requires_exact_cursor_and_contiguous_suffix(tmp_path):
+    append_run_event("session_1", "run_a", "token", {}, session_dir=tmp_path, seq=1)
+    exact_missing = read_session_run_events("session_1", after_event_id="run_a:2", session_dir=tmp_path)
+    append_run_event("session_1", "run_a", "token", {}, session_dir=tmp_path, seq=3)
+
+    gap = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path)
+
+    assert exact_missing["status"] != "ok"
+    assert exact_missing["events"] == []
+    assert gap["events"] == []
+
+
+def test_session_journal_accepts_final_cursor_and_replays_later_runs(tmp_path):
+    append_run_event("session_1", "run_a", "token", {}, session_dir=tmp_path, seq=1, created_at=100.0)
+    append_run_event("session_1", "run_b", "token", {}, session_dir=tmp_path, seq=1, created_at=200.0)
+
+    replay = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path)
+
+    assert replay["status"] == "ok"
+    assert [event["event_id"] for event in replay["events"]] == ["run_b:1"]
+
+
+def test_session_route_prefers_last_event_id_then_query_fallback(monkeypatch):
+    import api.routes as routes
+
+    captured = []
+    stop = _stop_after_first_heartbeat(monkeypatch)
+
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: 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 *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda _session_id, *, after_event_id=None, session_dir=None: captured.append(after_event_id) or {"status": "ok", "events": []},
+    )
+
+    handler = _FakeHandler()
+    handler.headers["Last-Event-ID"] = "run_a:2"
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_b:1"),
+        "session_1",
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_b:1"),
+        "session_1",
+    )
+
+    assert stop["count"] >= 2
+    assert captured == ["run_a:2", "run_b:1"]
+
+
+def test_session_route_arms_deadline_immediately_after_headers(monkeypatch):
+    import api.routes as routes
+
+    calls = []
+    stop = _stop_after_first_heartbeat(monkeypatch)
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {}))
+    monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(routes, "end_sse_headers", lambda _handler: calls.append("headers"))
+    monkeypatch.setattr(routes, "_sse_set_write_deadline", lambda _handler: calls.append("deadline"))
+
+    routes._handle_session_sse_stream_for_session(_FakeHandler(), urlparse("/api/sessions/session_1/events"), "session_1")
+
+    assert calls[:2] == ["headers", "deadline"]
+    assert stop["count"] == 1
+
+
+def test_session_route_emits_snapshot_without_id_for_missing_cursor_and_keepalive(monkeypatch):
+    import api.routes as routes
+
+    stop = _stop_after_first_heartbeat(monkeypatch)
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: 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 *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: {"status": "cursor_run_missing", "events": []},
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_missing:1"),
+        "session_1",
+    )
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert "event: session_snapshot\n" in body
+    assert ": keepalive\n\n" in body
+    assert "id: " not in body
+    assert stop["count"] == 1
+
+
+def test_session_route_blocks_hidden_sessions_before_replay_or_live_attach(monkeypatch):
+    import api.routes as routes
+
+    cap = _capture(monkeypatch)
+    monkeypatch.setattr(
+        routes,
+        "get_session",
+        lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, profile="other"),
+    )
+    monkeypatch.setattr(routes, "_get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("replay should not run")),
+    )
+    monkeypatch.setattr(
+        routes,
+        "_active_run_stream_for_session",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("live attach should not run")),
+    )
+
+    handler = _FakeHandler()
+    routes.handle_get(handler, urlparse("/api/sessions/hidden_session/events"))
+
+    assert cap["bad"] == ("Session not found", 404)
+
+
+def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch):
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            self.q.put_nowait(("token", {"text": "live replayed"}, "run_active:1"))
+            self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2"))
+            self.unsubscribed = False
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:1", "offline_buffered_events": 1}
+
+        def unsubscribe(self, q):
+            self.unsubscribed = q is self.q
+
+    stream = _FakeStream()
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: 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 *_args, **_kwargs: "run_active")
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: {
+            "status": "ok",
+            "events": [
+                {
+                    "run_id": "run_prev",
+                    "seq": 1,
+                    "event": "token",
+                    "payload": {"text": "replayed"},
+                    "event_id": "run_prev:1",
+                },
+                {
+                    "run_id": "run_active",
+                    "seq": 1,
+                    "event": "token",
+                    "payload": {"text": "replayed current"},
+                    "event_id": "run_active:1",
+                },
+            ],
+        },
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"),
+        "session_1",
+    )
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert "id: run_prev:1\n" in body
+    assert body.count("id: run_active:1\n") == 1
+    assert "id: run_active:2\n" in body
+    assert "event: stream_end\n" in body
+    assert stream.unsubscribed is True
+
+
+def test_session_route_unsubscribes_when_replay_disconnects(monkeypatch):
+    import api.routes as routes
+
+    class _DisconnectingWfile:
+        def write(self, _data):
+            raise BrokenPipeError("client closed")
+
+        def flush(self):
+            pass
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            self.unsubscribed = False
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:1"}
+
+        def unsubscribe(self, q):
+            self.unsubscribed = q is self.q
+
+    stream = _FakeStream()
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: 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 *_args, **_kwargs: "run_active")
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: {
+            "status": "ok",
+            "events": [
+                {
+                    "run_id": "run_prev",
+                    "seq": 1,
+                    "event": "token",
+                    "payload": {"text": "replayed"},
+                    "event_id": "run_prev:1",
+                },
+            ],
+        },
+    )
+
+    handler = _FakeHandler()
+    handler.wfile = _DisconnectingWfile()
+    assert routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"),
+        "session_1",
+    )
+    assert stream.unsubscribed is True
+
+
+def test_session_route_live_delivery_without_cursor_keeps_buffered_active_items(monkeypatch):
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            self.q.put_nowait(("token", {"text": "buffered"}, "run_active:1"))
+            self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2"))
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:1", "offline_buffered_events": 1}
+
+    stream = _FakeStream()
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: 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 *_args, **_kwargs: "run_active")
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("replay should not run without a cursor")),
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events"),
+        "session_1",
+    )
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert "id: run_active:1\n" in body
+    assert "id: run_active:2\n" in body
+    assert "event: stream_end\n" in body
+
+
+def test_session_route_reconciles_late_attach_cutoff_once(monkeypatch):
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            self.q.put_nowait(("token", {"text": "six"}, "run_active:6"))
+            self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:7"))
+            self.unsubscribed = 0
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:6"}
+
+        def unsubscribe(self, q):
+            self.unsubscribed += q is self.q
+
+    stream = _FakeStream()
+    available = {"value": False}
+    reads = []
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {}))
+    monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active" if available["value"] else None)
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: reads.append(1) or {"status": "ok", "events": [] if len(reads) == 1 else [{"event": "token", "payload": {"text": "six"}, "event_id": "run_active:6"}]},
+    )
+    monkeypatch.setattr(routes.time, "sleep", lambda _seconds: available.__setitem__("value", True))
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(handler, urlparse("/api/sessions/session_1/events?after_event_id=run_active:5"), "session_1")
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert body.count("id: run_active:6\n") == 1
+    assert "id: run_active:7\n" in body
+    assert stream.unsubscribed == 1
+
+
+def test_session_route_emits_snapshot_when_reconciliation_fails(monkeypatch):
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            self.q.put_nowait(("token", {"text": "buffered"}, "run_active:1"))
+            self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2"))
+
+        def subscribe_with_snapshot(self):
+            return self.q, {"last_event_id": "run_active:1"}
+
+    stream = _FakeStream()
+    reads = []
+    sessions = iter(
+        [
+            SimpleNamespace(session_id="session_1", compact=lambda **_kwargs: {"session_id": "session_1", "title": "stale"}),
+            SimpleNamespace(session_id="session_1", compact=lambda **_kwargs: {"session_id": "session_1", "title": "fresh"}),
+        ]
+    )
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(
+        routes,
+        "get_session",
+        lambda sid, metadata_only=False: next(sessions),
+    )
+    monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+    monkeypatch.setattr(
+        routes,
+        "read_session_run_events",
+        lambda *_args, **_kwargs: reads.append(1) or (
+            {"status": "ok", "events": [{"event": "token", "payload": {"text": "replayed"}, "event_id": "run_prev:1"}]}
+            if len(reads) == 1
+            else {"status": "replay_noncontiguous", "events": []}
+        ),
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(
+        handler,
+        urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"),
+        "session_1",
+    )
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert "event: session_snapshot\n" in body
+    assert '"title": "fresh"' in body
+    assert '"title": "stale"' not in body
+    assert "id: run_prev:1\n" not in body
+    assert body.count("id: run_active:1\n") == 0
+    assert "id: run_active:2\n" in body
+
+
+def test_session_route_bounds_sent_event_id_deduplication(monkeypatch):
+    import api.routes as routes
+
+    class _FakeStream:
+        def __init__(self):
+            self.q = queue.Queue()
+            for seq in (1, 2, 3, 1):
+                self.q.put_nowait(("token", {"seq": seq}, f"run_active:{seq}"))
+            self.q.put_nowait(("stream_end", {}, "run_active:4"))
+
+        def subscribe_with_snapshot(self):
+            return self.q, {}
+
+    stream = _FakeStream()
+    monkeypatch.setattr(routes, "_SESSION_SSE_SENT_EVENT_ID_LIMIT", 2)
+    monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {}))
+    monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active")
+    monkeypatch.setattr(routes, "STREAMS", {"run_active": stream})
+
+    handler = _FakeHandler()
+    routes._handle_session_sse_stream_for_session(handler, urlparse("/api/sessions/session_1/events"), "session_1")
+
+    body = handler.wfile.getvalue().decode("utf-8")
+    assert body.count("id: run_active:1\n") == 2
+    assert "id: run_active:4\n" in body