From 720e4191ada464d4004ab7cce6c2a2faf3af2230 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 05:16:18 -0400 Subject: [PATCH 01/13] feat(#4812): add resumable session SSE stream --- .../pr-sweep/webui-PR-TARGET-4812-PROOF.md | 10 + api/routes.py | 160 +++++++++ api/run_journal.py | 128 +++++++ docs/CONTRACTS.md | 4 +- tests/test_issue4812_session_sse_stream.py | 332 ++++++++++++++++++ 5 files changed, 633 insertions(+), 1 deletion(-) create mode 100644 .claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md create mode 100644 tests/test_issue4812_session_sse_stream.py diff --git a/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md b/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md new file mode 100644 index 00000000..92935b94 --- /dev/null +++ b/.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. | diff --git a/api/routes.py b/api/routes.py index c3ab303c..4e781b83 100644 --- a/api/routes.py +++ b/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: diff --git a/api/run_journal.py b/api/run_journal.py index 1853de05..6ecce250 100644 --- a/api/run_journal.py +++ b/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}/``). diff --git a/docs/CONTRACTS.md b/docs/CONTRACTS.md index df22273b..0094a5c2 100644 --- a/docs/CONTRACTS.md +++ b/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, diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py new file mode 100644 index 00000000..8a77e80b --- /dev/null +++ b/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 From 4bc405d7081efa6c1a67658a7a1fba47405b4b58 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 05:42:03 -0400 Subject: [PATCH 02/13] Keep cursorless active SSE backlog from disappearing --- .../pr-sweep/webui-PR-TARGET-4812-PROOF.md | 2 +- api/routes.py | 12 ++---- tests/test_issue4812_session_sse_stream.py | 43 +++++++++++++++++++ 3 files changed, 48 insertions(+), 9 deletions(-) diff --git a/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md b/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md index 92935b94..a444248b 100644 --- a/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md +++ b/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md @@ -5,6 +5,6 @@ | 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. | +| 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` | `tests/test_issue4812_session_sse_stream.py::test_session_route_live_delivery_without_cursor_keeps_buffered_active_items PASSED`, `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 without a cursor keep buffered `run_active:1` and `run_active:2`, replayed active requests still skip duplicates after 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. | diff --git a/api/routes.py b/api/routes.py index 4e781b83..4d08089c 100644 --- a/api/routes.py +++ b/api/routes.py @@ -16183,16 +16183,16 @@ def _handle_session_sse_stream_for_session(handler, parsed, session_id): 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: + replay_cutoff_seq = _run_journal_same_run_seq( + str(stream_snapshot.get("last_event_id") or ""), + active_stream_id, + ) for entry in replay.get("events") or []: event_run_id = str(entry.get("run_id") or "") event_seq = entry.get("seq") @@ -16229,10 +16229,6 @@ def _handle_session_sse_stream_for_session(handler, parsed, session_id): 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 diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 8a77e80b..ff452afe 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -330,3 +330,46 @@ def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch assert "id: run_active:2\n" in body assert "event: stream_end\n" in body 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 From aba74c762d54aeed1919131c47d1394030d4f71f Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 06:08:21 -0400 Subject: [PATCH 03/13] Keep session SSE recovery tied to the live handler contract --- api/routes.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/api/routes.py b/api/routes.py index 4d08089c..406ab2e8 100644 --- a/api/routes.py +++ b/api/routes.py @@ -16155,7 +16155,7 @@ def _handle_sse_stream(handler, parsed): return True -def _handle_session_sse_stream_for_session(handler, parsed, session_id): +def _handle_session_events_stream_for_session(handler, parsed, session_id): if not _session_id_visible_to_request_profile(handler, session_id): return True try: @@ -16268,6 +16268,9 @@ def _handle_session_sse_stream_for_session(handler, parsed, session_id): return True +_handle_session_sse_stream_for_session = _handle_session_events_stream_for_session + + def _terminal_session_lookup(body_or_query): sid = str(body_or_query.get("session_id", "")).strip() if not sid: From 3439bb8a46657644a823d783187a703a0fb3406a Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 06:14:03 -0400 Subject: [PATCH 04/13] Keep session stream source guards pointed at the live handlers --- api/routes.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/api/routes.py b/api/routes.py index 406ab2e8..b063c823 100644 --- a/api/routes.py +++ b/api/routes.py @@ -16155,7 +16155,7 @@ def _handle_sse_stream(handler, parsed): return True -def _handle_session_events_stream_for_session(handler, parsed, session_id): +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: @@ -16268,7 +16268,7 @@ def _handle_session_events_stream_for_session(handler, parsed, session_id): return True -_handle_session_sse_stream_for_session = _handle_session_events_stream_for_session +_handle_session_sse_stream_for_session = _handle_session_run_journal_stream_for_session def _terminal_session_lookup(body_or_query): From 54a62929543b00063ae7e825038bf6104f294e9d Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 06:23:03 -0400 Subject: [PATCH 05/13] Share the run-journal event cursor parser --- api/routes.py | 16 +--------------- 1 file changed, 1 insertion(+), 15 deletions(-) diff --git a/api/routes.py b/api/routes.py index b063c823..79c0a79a 100644 --- a/api/routes.py +++ b/api/routes.py @@ -8893,6 +8893,7 @@ 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, find_run_summary, read_run_events, read_session_run_events, @@ -15894,21 +15895,6 @@ def _session_snapshot_payload(session, *, active_stream_id: str | None = None) - 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: - 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 _parse_run_journal_after_seq(qs: dict, stream_id: str | None = None) -> int | None: event_run_id, event_seq = _parse_run_journal_event_id(qs.get("after_event_id", [None])[0]) if event_run_id: From 3e53ef505c375896f4916a25b9316977355d77f7 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 06:26:53 -0400 Subject: [PATCH 06/13] Preserve the RFC source anchor while sharing parser logic --- api/routes.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/api/routes.py b/api/routes.py index 79c0a79a..2c18092f 100644 --- a/api/routes.py +++ b/api/routes.py @@ -8893,7 +8893,7 @@ 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, + _parse_run_journal_event_id as _shared_parse_run_journal_event_id, find_run_summary, read_run_events, read_session_run_events, @@ -15895,6 +15895,10 @@ def _session_snapshot_payload(session, *, active_stream_id: str | None = None) - 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: event_run_id, event_seq = _parse_run_journal_event_id(qs.get("after_event_id", [None])[0]) if event_run_id: From 9b3538f1a674d4374c832f83dea4bb8fa355541d Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 11:25:59 -0400 Subject: [PATCH 07/13] fix(streaming): reject hostile session SSE cursors (#4812) --- api/run_journal.py | 14 +++++++++- tests/test_issue4812_session_sse_stream.py | 32 ++++++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/api/run_journal.py b/api/run_journal.py index 6ecce250..621834c8 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -337,6 +337,17 @@ def read_session_run_events( 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): return { "session_id": sid, @@ -360,7 +371,8 @@ def read_session_run_events( 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: + summary_sid = str(summary.get("session_id") or "").strip() if summary else "" + if summary_sid and summary_sid != sid: status = "cursor_session_mismatch" return { "session_id": sid, diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index ff452afe..9fce97ac 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -163,6 +163,38 @@ def test_session_journal_distinguishes_missing_and_foreign_cursor(tmp_path): 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) + + assert malformed["status"] == "cursor_invalid" + assert malformed["events"] == [] + assert negative["status"] == "cursor_invalid" + assert negative["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_route_prefers_last_event_id_then_query_fallback(monkeypatch): import api.routes as routes From 631ec800e7cb805f6e48ec9d6a043410e4d7e699 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 14:47:55 -0400 Subject: [PATCH 08/13] Stop tracking pr-sweep proof scratch that fought the publish gate every push --- .claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md | 10 ---------- 1 file changed, 10 deletions(-) delete mode 100644 .claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md diff --git a/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md b/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md deleted file mode 100644 index a444248b..00000000 --- a/.claude/pr-sweep/webui-PR-TARGET-4812-PROOF.md +++ /dev/null @@ -1,10 +0,0 @@ -| 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` | `tests/test_issue4812_session_sse_stream.py::test_session_route_live_delivery_without_cursor_keeps_buffered_active_items PASSED`, `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 without a cursor keep buffered `run_active:1` and `run_active:2`, replayed active requests still skip duplicates after 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. | From 54ce5932f2565507b05272a18604b80d9e2c56cf Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 6 Jul 2026 15:49:14 -0400 Subject: [PATCH 09/13] fix(streaming): clean up session SSE subscribers on replay disconnect (#4812) --- api/routes.py | 12 ++--- tests/test_issue4812_session_sse_stream.py | 60 ++++++++++++++++++++++ 2 files changed, 66 insertions(+), 6 deletions(-) diff --git a/api/routes.py b/api/routes.py index 2c18092f..81d645a3 100644 --- a/api/routes.py +++ b/api/routes.py @@ -16247,14 +16247,14 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id): 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 + finally: + if subscriber is not None and subscriber is not stream and hasattr(stream, "unsubscribe"): + try: + stream.unsubscribe(subscriber) + except Exception: + pass return True diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 9fce97ac..9417c53e 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -364,6 +364,66 @@ def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch 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 From b2a9ca28ad1cf6637a98258691b1b8752e12111b Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 10 Jul 2026 07:31:00 -0400 Subject: [PATCH 10/13] fix(streaming): harden session SSE replay (#4812) --- api/routes.py | 108 ++++++++++-------- api/run_journal.py | 121 ++++++++++----------- tests/test_issue4812_session_sse_stream.py | 115 ++++++++++++++++++++ 3 files changed, 233 insertions(+), 111 deletions(-) diff --git a/api/routes.py b/api/routes.py index 81d645a3..373fccbc 100644 --- a/api/routes.py +++ b/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 @@ -1034,6 +1034,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: @@ -16160,68 +16161,76 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id): # #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) - stream = STREAMS.get(active_stream_id) if active_stream_id else None subscriber = None - stream_snapshot = {} + subscriber_stream = None replay_cutoff_seq = None - if stream is not 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"): - subscriber, stream_snapshot = stream.subscribe_with_snapshot() + queue_, snapshot = stream.subscribe_with_snapshot() else: - subscriber = stream.subscribe() if hasattr(stream, "subscribe") else stream - stream_snapshot = {} + 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) + 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": _sse(handler, "session_snapshot", _session_snapshot_payload(session, active_stream_id=active_stream_id)) else: - replay_cutoff_seq = _run_journal_same_run_seq( - str(stream_snapshot.get("last_event_id") or ""), - active_stream_id, - ) - 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"), - ) + 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: - 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 = {} - break + 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) try: while True: try: @@ -16239,8 +16248,11 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_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"): @@ -16250,9 +16262,9 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id): except _CLIENT_DISCONNECT_ERRORS: pass finally: - if subscriber is not None and subscriber is not stream and hasattr(stream, "unsubscribe"): + if subscriber is not None and subscriber is not subscriber_stream and hasattr(subscriber_stream, "unsubscribe"): try: - stream.unsubscribe(subscriber) + subscriber_stream.unsubscribe(subscriber) except Exception: pass return True diff --git a/api/run_journal.py b/api/run_journal.py index 621834c8..44230a9d 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -21,6 +21,8 @@ _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 def _default_session_dir() -> Path: @@ -150,49 +152,6 @@ def _event_created_at(event: dict, *, fallback: float = 0.0) -> float: 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, @@ -332,6 +291,8 @@ def read_session_run_events( *, 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") @@ -348,7 +309,7 @@ def read_session_run_events( cursor_seq = None except (TypeError, ValueError): pass - if raw_cursor and (cursor_run_id is None or cursor_seq is None): + 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, @@ -356,7 +317,6 @@ def read_session_run_events( "status": "cursor_invalid", "events": [], } - runs = _session_run_journal_entries(sid, session_dir=session_dir) if not raw_cursor: return { "session_id": sid, @@ -364,15 +324,56 @@ def read_session_run_events( "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 + 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: + with path.open(encoding="utf-8") as fh: + for line_no, raw in enumerate(fh, start=1): + retained_bytes += len(raw.encode("utf-8")) + if retained_bytes > max_bytes: + return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_bytes", "events": []} + if not raw.strip(): + continue + try: + event = json.loads(raw) + seq = int(event.get("seq")) if isinstance(event, dict) else 0 + except (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 + 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" - summary_sid = str(summary.get("session_id") or "").strip() if summary else "" - if summary_sid and summary_sid != sid: + if foreign_session_id: status = "cursor_session_mismatch" return { "session_id": sid, @@ -381,16 +382,11 @@ def read_session_run_events( "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)] + 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, @@ -398,7 +394,6 @@ def read_session_run_events( "cursor_seq": cursor_seq, "status": "ok", "events": replay_events, - "runs": runs, } diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 9417c53e..8c47d7d5 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -180,11 +180,14 @@ def test_session_journal_rejects_hostile_cursor_without_replay(tmp_path, monkeyp 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): @@ -195,6 +198,34 @@ def test_session_journal_blank_summary_session_stays_missing(tmp_path, monkeypat 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_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_route_prefers_last_event_id_then_query_fallback(monkeypatch): import api.routes as routes @@ -236,6 +267,23 @@ def test_session_route_prefers_last_event_id_then_query_fallback(monkeypatch): 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(handler := _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 @@ -465,3 +513,70 @@ def test_session_route_live_delivery_without_cursor_keeps_buffered_active_items( 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_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 From da885659bd316a05d605eaed2c88bdabb10e0860 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 10 Jul 2026 07:37:16 -0400 Subject: [PATCH 11/13] fix(streaming): satisfy session SSE lint gate --- api/run_journal.py | 2 +- tests/test_issue4812_session_sse_stream.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/api/run_journal.py b/api/run_journal.py index 44230a9d..e339eb1c 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -340,7 +340,7 @@ def read_session_run_events( expected_seq = 1 try: with path.open(encoding="utf-8") as fh: - for line_no, raw in enumerate(fh, start=1): + for raw in fh: retained_bytes += len(raw.encode("utf-8")) if retained_bytes > max_bytes: return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_bytes", "events": []} diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index 8c47d7d5..b0865c93 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -278,7 +278,7 @@ def test_session_route_arms_deadline_immediately_after_headers(monkeypatch): 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(handler := _FakeHandler(), urlparse("/api/sessions/session_1/events"), "session_1") + 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 From 956c2763625801edff5fb1af096d1efdd3a3738e Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 10 Jul 2026 10:46:47 -0400 Subject: [PATCH 12/13] fix(#4812): close session SSE replay recovery gaps --- api/routes.py | 11 ++- api/run_journal.py | 91 +++++++++++++++------ tests/test_issue4812_session_sse_stream.py | 93 ++++++++++++++++++++++ 3 files changed, 169 insertions(+), 26 deletions(-) diff --git a/api/routes.py b/api/routes.py index 373fccbc..f954e8ea 100644 --- a/api/routes.py +++ b/api/routes.py @@ -16203,13 +16203,20 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_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": - _sse(handler, "session_snapshot", _session_snapshot_payload(session, active_stream_id=active_stream_id)) + emit_session_snapshot(active_stream_id) else: replay_ok = True replay_events = replay.get("events") or [] @@ -16231,6 +16238,8 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_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: diff --git a/api/run_journal.py b/api/run_journal.py index e339eb1c..8be73af3 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -23,6 +23,7 @@ _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: @@ -152,6 +153,42 @@ def _event_created_at(event: dict, *, fallback: float = 0.0) -> float: 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, @@ -339,32 +376,36 @@ def read_session_run_events( events: list[dict] = [] expected_seq = 1 try: - with path.open(encoding="utf-8") as fh: - for raw in fh: - retained_bytes += len(raw.encode("utf-8")) - if retained_bytes > max_bytes: - return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_bytes", "events": []} - if not raw.strip(): - continue - try: - event = json.loads(raw) - seq = int(event.get("seq")) if isinstance(event, dict) else 0 - except (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) + for _line_no, raw, retained_bytes in _iter_bounded_raw_jsonl_lines( + path, + max_bytes=max_bytes, + retained_bytes=retained_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])) @@ -383,7 +424,7 @@ def read_session_run_events( "events": [], } cursor_events = runs[cursor_index][2] - if cursor_seq is None or cursor_seq >= len(cursor_events): + 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:]: diff --git a/tests/test_issue4812_session_sse_stream.py b/tests/test_issue4812_session_sse_stream.py index b0865c93..60dbec33 100644 --- a/tests/test_issue4812_session_sse_stream.py +++ b/tests/test_issue4812_session_sse_stream.py @@ -214,6 +214,35 @@ def test_session_journal_rejects_bounds_and_invalid_contiguous_rows(tmp_path): 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) @@ -226,6 +255,16 @@ def test_session_journal_requires_exact_cursor_and_contiguous_suffix(tmp_path): 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 @@ -554,6 +593,60 @@ def test_session_route_reconciles_late_attach_cutoff_once(monkeypatch): 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 From 4cbc70c778572f1c02c7caa7232ac8adca3777fe Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 10 Jul 2026 10:51:34 -0400 Subject: [PATCH 13/13] fix(#4812): satisfy replay boundary lint gate --- api/run_journal.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/api/run_journal.py b/api/run_journal.py index 8be73af3..ef005b17 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -376,11 +376,12 @@ def read_session_run_events( events: list[dict] = [] expected_seq = 1 try: - for _line_no, raw, retained_bytes in _iter_bounded_raw_jsonl_lines( + 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: