feat(#4812): add resumable session SSE stream

.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md

@@ -0,0 +1,10 @@
+| surface | command run | observed result | base/head |
+| --- | --- | --- | --- |
+| Path distinction between `/api/sessions/events` and `/api/sessions/{session_id}/events` | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `tests/test_issue4812_session_sse_stream.py::test_session_route_and_global_route_stay_separate PASSED` | base: `"/api/sessions/events"` still maps to the global invalidation handler; head: `"/api/sessions/session_1/events"` reaches `_handle_session_sse_stream_for_session`. |
+| Durable resume from `Last-Event-ID` or `after_event_id` across runs in one session | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `tests/test_issue4812_session_sse_stream.py::test_session_journal_replays_later_runs_after_cursor PASSED` | base: no session-scoped replay helper existed; head: replay after `run_a:1` returned `["run_a:2", "run_b:1", "run_b:2"]` across later runs in the same session. |
+| Snapshot fallback when replay is unavailable | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `tests/test_issue4812_session_sse_stream.py::test_session_route_emits_snapshot_without_id_for_missing_cursor_and_keepalive PASSED` | base: no `session_snapshot` fallback existed for the session-scoped stream; head: the missing cursor path emitted `event: session_snapshot` with no `id:` line and continued with `: keepalive`. |
+| Per-session auth and visibility enforcement | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `tests/test_issue4812_session_sse_stream.py::test_session_route_blocks_hidden_sessions_before_replay_or_live_attach PASSED` | base: there was no per-session route to guard; head: hidden sessions return `404` before replay or live attach. |
+| Live delivery for an active stream after replay cutoff | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `tests/test_issue4812_session_sse_stream.py::test_session_route_live_delivery_skips_replayed_active_run_items PASSED` | base: no session-scoped live bridge existed; head: replay emitted `id: run_active:1` once, the live queue skipped the duplicate, and `id: run_active:2` arrived from the active stream. |
+| Mode and state matrix boundary value `404` across idle, active, stale, foreign, and unauthorized session paths | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `7 passed in 4.87s`, `404` | base: no session-scoped route existed, so boundary value `404` and the idle / active / stale / foreign states could not be exercised; head: idle requests emit keepalive, active requests attach live after replay cutoff, stale or foreign cursors fall back to `session_snapshot`, and unauthorized requests return `404` before replay or live attach. |
+| Negative-space replay guard boundary value `1 foreign cursor` on durable replay | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_issue4812_session_sse_stream.py -v --timeout=60` | `tests/test_issue4812_session_sse_stream.py::test_session_journal_distinguishes_missing_and_foreign_cursor PASSED` | base: parser helpers alone did not stop a foreign run cursor from being reused; head: the explicit `1 foreign cursor` case returns `cursor_session_mismatch` and does not replay foreign data. |
+| Existing global `/api/sessions/events` handshake and keepalive behavior | `conhost --headless cmd /v:on /c D:\Repos\hermes-agent\venv\Scripts\python.exe -m pytest tests/test_session_events_http_integration.py -v --timeout=60` | `tests/test_session_events_http_integration.py::test_session_events_sse_endpoint_handshake_succeeds PASSED` | base: the global endpoint already returned `text/event-stream` with `sessions_changed` or keepalive; head: the same handshake and keepalive behavior still passes unchanged after the new per-session route landed. |

api/routes.py

@@ -8895,6 +8895,7 @@ from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_ena
 from api.run_journal import (
     find_run_summary,
     read_run_events,
+    read_session_run_events,
     stale_interrupted_event,
 )
 from api.todo_state import attach_todo_state
@@ -12432,6 +12433,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)
 
@@ -15851,6 +15856,44 @@ def _sse_with_id(handler, event, data, event_id=None):
     _sse(handler, event, data)
 
 
+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 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:
+        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]:
     raw = str(raw or "").strip()
     if not raw:
@@ -16112,6 +16155,123 @@ def _handle_sse_stream(handler, parsed):
     return True
 
 
+def _handle_session_sse_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)
+
+    resume_event_id = _session_events_resume_event_id(handler, parsed)
+    active_stream_id = _active_run_stream_for_session(session_id)
+    stream = STREAMS.get(active_stream_id) if active_stream_id else None
+    subscriber = None
+    stream_snapshot = {}
+    replay_cutoff_seq = None
+    if stream is not None:
+        if hasattr(stream, "subscribe_with_snapshot"):
+            subscriber, stream_snapshot = stream.subscribe_with_snapshot()
+        else:
+            subscriber = stream.subscribe() if hasattr(stream, "subscribe") else stream
+            stream_snapshot = {}
+        replay_cutoff_seq = _run_journal_same_run_seq(
+            str(stream_snapshot.get("last_event_id") or ""),
+            active_stream_id,
+        )
+    try:
+        if resume_event_id:
+            replay = read_session_run_events(session_id, after_event_id=resume_event_id)
+            if replay.get("status") != "ok":
+                _sse(handler, "session_snapshot", _session_snapshot_payload(session, active_stream_id=active_stream_id))
+            else:
+                for entry in replay.get("events") or []:
+                    event_run_id = str(entry.get("run_id") or "")
+                    event_seq = entry.get("seq")
+                    if (
+                        subscriber is not None
+                        and replay_cutoff_seq is not None
+                        and event_run_id == active_stream_id
+                        and event_seq is not None
+                        and int(event_seq) > int(replay_cutoff_seq)
+                    ):
+                        continue
+                    _sse_with_id(
+                        handler,
+                        entry.get("event") or entry.get("type") or "message",
+                        entry.get("payload"),
+                        entry.get("event_id"),
+                    )
+        if subscriber is None:
+            while True:
+                active_stream_id = _active_run_stream_for_session(session_id)
+                if not active_stream_id:
+                    handler.wfile.write(b": keepalive\n\n")
+                    handler.wfile.flush()
+                    time.sleep(_SSE_HEARTBEAT_INTERVAL_SECONDS)
+                    continue
+                stream = STREAMS.get(active_stream_id)
+                if stream is None:
+                    handler.wfile.write(b": keepalive\n\n")
+                    handler.wfile.flush()
+                    time.sleep(_SSE_HEARTBEAT_INTERVAL_SECONDS)
+                    continue
+                if hasattr(stream, "subscribe_with_snapshot"):
+                    subscriber, stream_snapshot = stream.subscribe_with_snapshot()
+                else:
+                    subscriber = stream.subscribe() if hasattr(stream, "subscribe") else stream
+                    stream_snapshot = {}
+                replay_cutoff_seq = _run_journal_same_run_seq(
+                    str(stream_snapshot.get("last_event_id") or ""),
+                    active_stream_id,
+                )
+                break
+        if subscriber is None:
+            return True
+        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:
+                    _sse_with_id(handler, event, data, event_id)
+                else:
+                    _sse(handler, event, data)
+                if event in ("stream_end", "error", "cancel"):
+                    break
+        except _CLIENT_DISCONNECT_ERRORS:
+            pass
+        finally:
+            if subscriber is not stream and hasattr(stream, "unsubscribe"):
+                try:
+                    stream.unsubscribe(subscriber)
+                except Exception:
+                    pass
+    except _CLIENT_DISCONNECT_ERRORS:
+        pass
+    return True
+
+
 def _terminal_session_lookup(body_or_query):
     sid = str(body_or_query.get("session_id", "")).strip()
     if not sid:

api/run_journal.py

@@ -75,6 +75,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)]
@@ -128,6 +143,56 @@ 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 _session_run_journal_entries(session_id: str, *, session_dir: Path | None = None) -> list[dict]:
+    sid = _validate_id(session_id, "session_id")
+    root = Path(session_dir) if session_dir is not None else _default_session_dir()
+    session_root = root / RUN_JOURNAL_DIR_NAME / sid
+    if not session_root.exists():
+        return []
+    entries: list[dict] = []
+    for path in sorted(session_root.glob("*.jsonl")):
+        run_id = path.stem
+        try:
+            run_id = _validate_id(run_id, "run_id")
+        except ValueError:
+            continue
+        events, malformed = _read_jsonl(path)
+        ordered_events = sorted(
+            [event for event in events if isinstance(event, dict)],
+            key=lambda event: (
+                int(event.get("seq") or 0),
+                _event_created_at(event),
+                str(event.get("event_id") or ""),
+            ),
+        )
+        summary = _summary_from_events(sid, run_id, ordered_events)
+        created_at = min((
+            _event_created_at(event)
+            for event in ordered_events
+        ), default=path.stat().st_mtime if path.exists() else 0.0)
+        entries.append(
+            {
+                "session_id": sid,
+                "run_id": run_id,
+                "path": str(path),
+                "created_at": created_at,
+                "event_count": len(ordered_events),
+                "events": ordered_events,
+                "summary": summary,
+                "malformed": malformed,
+            }
+        )
+    entries.sort(key=lambda entry: (entry["created_at"], entry["run_id"]))
+    return entries
+
+
 def append_run_event(
     session_id: str,
     run_id: str,
@@ -262,6 +327,69 @@ 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,
+) -> 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 None or cursor_seq is None):
+        return {
+            "session_id": sid,
+            "cursor_run_id": cursor_run_id,
+            "cursor_seq": cursor_seq,
+            "status": "cursor_invalid",
+            "events": [],
+        }
+    runs = _session_run_journal_entries(sid, session_dir=session_dir)
+    if not raw_cursor:
+        return {
+            "session_id": sid,
+            "cursor_run_id": None,
+            "cursor_seq": None,
+            "status": "ok",
+            "events": [],
+            "runs": runs,
+        }
+    run_lookup = {entry["run_id"]: entry for entry in runs}
+    cursor_entry = run_lookup.get(cursor_run_id or "")
+    if cursor_entry is None:
+        summary = find_run_summary(cursor_run_id or "", session_dir=session_dir) if cursor_run_id else None
+        status = "cursor_run_missing"
+        if summary and str(summary.get("session_id") or "") != sid:
+            status = "cursor_session_mismatch"
+        return {
+            "session_id": sid,
+            "cursor_run_id": cursor_run_id,
+            "cursor_seq": cursor_seq,
+            "status": status,
+            "events": [],
+        }
+    replay_events: list[dict] = []
+    started = False
+    for entry in runs:
+        if not started:
+            if entry["run_id"] != cursor_entry["run_id"]:
+                continue
+            started = True
+        events = entry["events"]
+        if entry["run_id"] == cursor_entry["run_id"] and cursor_seq is not None:
+            events = [event for event in events if int(event.get("seq") or 0) > int(cursor_seq)]
+        replay_events.extend(events)
+    return {
+        "session_id": sid,
+        "cursor_run_id": cursor_run_id,
+        "cursor_seq": cursor_seq,
+        "status": "ok",
+        "events": replay_events,
+        "runs": runs,
+    }
+
+
 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,332 @@
+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_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_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