diff --git a/api/routes.py b/api/routes.py index 2b70c39e..0846dc5c 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 @@ -1042,6 +1042,7 @@ def _skill_view_from_active_dir(name: str) -> dict: # Trivial; many production SSE deployments run 5-15s heartbeats specifically # to handle proxies and mobile NAT. _SSE_HEARTBEAT_INTERVAL_SECONDS = 5 +_SESSION_SSE_SENT_EVENT_ID_LIMIT = 4096 def _normalize_messaging_source(raw_source) -> str: @@ -9254,8 +9255,10 @@ from api.streaming import ( ) from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_enabled from api.run_journal import ( + _parse_run_journal_event_id as _shared_parse_run_journal_event_id, find_run_summary, read_run_events, + read_session_run_events, stale_interrupted_event, ) from api.todo_state import attach_todo_state @@ -12900,6 +12903,10 @@ def handle_get(handler, parsed) -> bool: if parsed.path == '/api/sessions/events': return _handle_session_events_stream(handler) + session_events_session_id = _session_events_path_session_id(parsed.path) + if session_events_session_id is not None: + return _handle_session_sse_stream_for_session(handler, parsed, session_events_session_id) + if parsed.path == "/api/media": return _handle_media(handler, parsed) @@ -16399,19 +16406,46 @@ def _sse_with_id(handler, event, data, event_id=None): _sse(handler, event, data) -def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]: +def _session_events_path_session_id(path: str | None) -> str | None: + path = str(path or "") + parts = path.strip("/").split("/") + if len(parts) != 4: + return None + if parts[0] != "api" or parts[1] != "sessions" or parts[3] != "events": + return None + sid = str(parts[2] or "").strip() + return sid or None + + +def _session_events_resume_event_id(handler, parsed) -> str | None: + headers = getattr(handler, "headers", None) + raw = None + if headers is not None: + try: + raw = headers.get("Last-Event-ID") + except Exception: + raw = None raw = str(raw or "").strip() - if not raw: - return None, None - if ":" in raw: - run_id, tail = raw.rsplit(":", 1) - else: - run_id, tail = None, raw + if raw: + return raw + qs = parse_qs(getattr(parsed, "query", "") or "") + raw = str(qs.get("after_event_id", [None])[0] or "").strip() + return raw or None + + +def _session_snapshot_payload(session, *, active_stream_id: str | None = None) -> dict: try: - seq = max(0, int(tail)) - except (TypeError, ValueError): - return run_id or None, None - return run_id or None, seq + payload = session.compact( + include_runtime=bool(active_stream_id), + active_stream_ids={active_stream_id} if active_stream_id else None, + ) + except Exception: + payload = {"session_id": str(getattr(session, "session_id", "") or "")} + return {"session": payload} + + +def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]: + return _shared_parse_run_journal_event_id(raw) def _parse_run_journal_after_seq(qs: dict, stream_id: str | None = None) -> int | None: @@ -16825,6 +16859,142 @@ def _handle_sse_stream(handler, parsed): return True +def _handle_session_run_journal_stream_for_session(handler, parsed, session_id): + if not _session_id_visible_to_request_profile(handler, session_id): + return True + try: + session = get_session(session_id, metadata_only=True) + except KeyError: + return j(handler, {"error": "Session not found"}, status=404) + + handler.send_response(200) + handler.send_header("Content-Type", "text/event-stream; charset=utf-8") + handler.send_header("Cache-Control", "no-cache") + handler.send_header("X-Accel-Buffering", "no") + # #3103: see _handle_gateway_sse_stream — `Connection: close` causes + # EventSource reconnect storms in browsers on long-lived SSE. + end_sse_headers(handler) + _sse_set_write_deadline(handler) + + resume_event_id = _session_events_resume_event_id(handler, parsed) + active_stream_id = _active_run_stream_for_session(session_id) + subscriber = None + subscriber_stream = None + replay_cutoff_seq = None + sent_event_ids: set[str] = set() + sent_event_order = deque() + + def note_sent_event_id(event_id): + if not event_id: + return + sent_event_ids.add(event_id) + sent_event_order.append(event_id) + while len(sent_event_order) > _SESSION_SSE_SENT_EVENT_ID_LIMIT: + sent_event_ids.discard(sent_event_order.popleft()) + + def attach_active_stream(): + stream_id = _active_run_stream_for_session(session_id) + stream = STREAMS.get(stream_id) if stream_id else None + if stream is None: + return None, None, None, stream_id + if hasattr(stream, "subscribe_with_snapshot"): + queue_, snapshot = stream.subscribe_with_snapshot() + else: + queue_ = stream.subscribe() if hasattr(stream, "subscribe") else stream + snapshot = {} + return queue_, stream, snapshot, stream_id + + def emit_replay(events, stream_id, cutoff_seq): + for entry in events: + event_id = str(entry.get("event_id") or "") + event_seq = _run_journal_same_run_seq(event_id, stream_id) + if cutoff_seq is not None and event_seq is not None and event_seq > cutoff_seq: + continue + if event_id and event_id in sent_event_ids: + continue + _sse_with_id(handler, entry.get("event") or entry.get("type") or "message", entry.get("payload"), event_id) + if event_id: + note_sent_event_id(event_id) + + def emit_session_snapshot(active_stream_id): + try: + fresh_session = get_session(session_id, metadata_only=True) + except KeyError: + fresh_session = session + _sse(handler, "session_snapshot", _session_snapshot_payload(fresh_session, active_stream_id=active_stream_id)) + + try: + replay_events = [] + replay_ok = False + if resume_event_id: + replay = read_session_run_events(session_id, after_event_id=resume_event_id) + if replay.get("status") != "ok": + emit_session_snapshot(active_stream_id) + else: + replay_ok = True + replay_events = replay.get("events") or [] + subscriber, subscriber_stream, stream_snapshot, active_stream_id = attach_active_stream() + if subscriber is None: + if replay_ok: + emit_replay(replay_events, active_stream_id, None) + while True: + subscriber, subscriber_stream, stream_snapshot, active_stream_id = attach_active_stream() + if subscriber is not None: + break + handler.wfile.write(b": keepalive\n\n") + handler.wfile.flush() + time.sleep(_SSE_HEARTBEAT_INTERVAL_SECONDS) + if subscriber is None: + return True + if replay_ok: + replay_cutoff_seq = _run_journal_same_run_seq(str(stream_snapshot.get("last_event_id") or ""), active_stream_id) + reconciled = read_session_run_events(session_id, after_event_id=resume_event_id) + if reconciled.get("status") == "ok": + emit_replay(reconciled.get("events") or [], active_stream_id, replay_cutoff_seq) + else: + emit_session_snapshot(active_stream_id) + try: + while True: + try: + item = subscriber.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS) + except queue.Empty: + handler.wfile.write(b": keepalive\n\n") + handler.wfile.flush() + continue + if len(item) >= 3: + event, data, queued_event_id = item[0], item[1], item[2] + else: + event, data = item + queued_event_id = STREAM_LAST_EVENT_ID.get(active_stream_id) + event_id = queued_event_id or STREAM_LAST_EVENT_ID.get(active_stream_id) + event_seq = _run_journal_same_run_seq(event_id, active_stream_id) + if replay_cutoff_seq is not None and event_seq is not None and event_seq <= replay_cutoff_seq: + continue + if event_id and event_id in sent_event_ids: + continue + if event_id: + _sse_with_id(handler, event, data, event_id) + note_sent_event_id(event_id) + else: + _sse(handler, event, data) + if event in ("stream_end", "error", "cancel"): + break + except _CLIENT_DISCONNECT_ERRORS: + pass + except _CLIENT_DISCONNECT_ERRORS: + pass + finally: + if subscriber is not None and subscriber is not subscriber_stream and hasattr(subscriber_stream, "unsubscribe"): + try: + subscriber_stream.unsubscribe(subscriber) + except Exception: + pass + return True + + +_handle_session_sse_stream_for_session = _handle_session_run_journal_stream_for_session + + def _terminal_session_lookup(body_or_query): sid = str(body_or_query.get("session_id", "")).strip() if not sid: diff --git a/api/run_journal.py b/api/run_journal.py index 81e92e75..5e277d96 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -31,6 +31,9 @@ _TERMINAL_SSE_EVENTS = {"done", "cancel", "apperror", "error", "stream_end"} _FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC" _FSYNC_MODE_EAGER = "eager" _FSYNC_MODE_TERMINAL_ONLY = "terminal-only" +_SESSION_REPLAY_MAX_BYTES = 4 * 1024 * 1024 +_SESSION_REPLAY_MAX_ROWS = 4096 +_SESSION_REPLAY_READ_CHUNK_BYTES = 64 * 1024 def _default_session_dir() -> Path: @@ -85,6 +88,21 @@ def _read_jsonl(path: Path) -> tuple[list[dict], list[dict]]: return events, malformed +def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]: + raw = str(raw or "").strip() + if not raw: + return None, None + if ":" in raw: + run_id, tail = raw.rsplit(":", 1) + else: + run_id, tail = None, raw + try: + seq = max(0, int(tail)) + except (TypeError, ValueError): + return run_id or None, None + return run_id or None, seq + + def _next_seq(path: Path) -> int: events, _malformed = _read_jsonl(path) seqs = [int(event.get("seq") or 0) for event in events if isinstance(event.get("seq"), int)] @@ -181,6 +199,49 @@ def _fsync_parent_dir(path: Path) -> None: pass +def _event_created_at(event: dict, *, fallback: float = 0.0) -> float: + try: + return float(event.get("created_at") or fallback) + except (TypeError, ValueError): + return fallback + + +def _iter_bounded_raw_jsonl_lines(path: Path, *, max_bytes: int, retained_bytes: int = 0): + line_no = 0 + buffered = bytearray() + total_bytes = int(retained_bytes) + try: + with path.open("rb") as fh: + while True: + chunk = fh.read(_SESSION_REPLAY_READ_CHUNK_BYTES) + if not chunk: + if buffered: + if total_bytes + len(buffered) > max_bytes: + raise ValueError("replay_limit_bytes") + line_no += 1 + total_bytes += len(buffered) + yield line_no, bytes(buffered), total_bytes + return + start = 0 + while start < len(chunk): + newline = chunk.find(b"\n", start) + if newline == -1: + buffered.extend(chunk[start:]) + if total_bytes + len(buffered) > max_bytes: + raise ValueError("replay_limit_bytes") + break + buffered.extend(chunk[start : newline + 1]) + if total_bytes + len(buffered) > max_bytes: + raise ValueError("replay_limit_bytes") + line_no += 1 + total_bytes += len(buffered) + yield line_no, bytes(buffered), total_bytes + buffered.clear() + start = newline + 1 + except FileNotFoundError: + return + + def append_run_event( session_id: str, run_id: str, @@ -319,6 +380,122 @@ def find_run_summary(run_id: str, *, session_dir: Path | None = None) -> dict | return None +def read_session_run_events( + session_id: str, + *, + after_event_id: str | None = None, + session_dir: Path | None = None, + max_bytes: int = _SESSION_REPLAY_MAX_BYTES, + max_rows: int = _SESSION_REPLAY_MAX_ROWS, +) -> dict: + """Replay durable run-journal rows for one session after an opaque cursor.""" + sid = _validate_id(session_id, "session_id") + cursor_run_id, cursor_seq = _parse_run_journal_event_id(after_event_id) + raw_cursor = str(after_event_id or "").strip() + if raw_cursor and cursor_run_id is not None: + try: + cursor_run_id = _validate_id(cursor_run_id, "run_id") + except ValueError: + cursor_seq = None + if raw_cursor: + try: + if int(raw_cursor.rsplit(":", 1)[-1]) < 0: + cursor_seq = None + except (TypeError, ValueError): + pass + if raw_cursor and (cursor_run_id is None or cursor_seq is None or cursor_seq <= 0): + return { + "session_id": sid, + "cursor_run_id": cursor_run_id, + "cursor_seq": cursor_seq, + "status": "cursor_invalid", + "events": [], + } + if not raw_cursor: + return { + "session_id": sid, + "cursor_run_id": None, + "cursor_seq": None, + "status": "ok", + "events": [], + } + root = Path(session_dir) if session_dir is not None else _default_session_dir() + session_root = root / RUN_JOURNAL_DIR_NAME / sid + runs: list[tuple[float, str, list[dict]]] = [] + retained_rows = 0 + retained_bytes = 0 + for path in sorted(session_root.glob("*.jsonl")) if session_root.exists() else []: + run_id = path.stem + try: + run_id = _validate_id(run_id, "run_id") + except ValueError: + continue + events: list[dict] = [] + expected_seq = 1 + try: + for _line_no, raw, total_bytes in _iter_bounded_raw_jsonl_lines( + path, + max_bytes=max_bytes, + retained_bytes=retained_bytes, + ): + retained_bytes = total_bytes + if not raw.strip(): + continue + try: + event = json.loads(raw.decode("utf-8")) + seq = int(event.get("seq")) if isinstance(event, dict) else 0 + except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError): + return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_malformed", "events": []} + if ( + seq != expected_seq + or event.get("event_id") != f"{run_id}:{seq}" + or event.get("run_id") != run_id + or event.get("session_id") != sid + ): + return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_noncontiguous", "events": []} + expected_seq += 1 + retained_rows += 1 + if retained_rows > max_rows: + return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_rows", "events": []} + events.append(event) + except FileNotFoundError: + continue + except ValueError as exc: + if str(exc) == "replay_limit_bytes": + return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "replay_limit_bytes", "events": []} + raise + created_at = min((_event_created_at(event) for event in events), default=path.stat().st_mtime) + runs.append((created_at, run_id, events)) + runs.sort(key=lambda run: (run[0], run[1])) + cursor_index = next((index for index, (_created_at, run_id, _events) in enumerate(runs) if run_id == cursor_run_id), None) + if cursor_index is None: + foreign_paths = root.joinpath(RUN_JOURNAL_DIR_NAME).glob(f"*/{cursor_run_id}.jsonl") if cursor_run_id else [] + foreign_session_id = next((path.parent.name for path in foreign_paths if path.parent.name != sid), "") + status = "cursor_run_missing" + if foreign_session_id: + status = "cursor_session_mismatch" + return { + "session_id": sid, + "cursor_run_id": cursor_run_id, + "cursor_seq": cursor_seq, + "status": status, + "events": [], + } + cursor_events = runs[cursor_index][2] + if cursor_seq is None or cursor_seq > len(cursor_events): + return {"session_id": sid, "cursor_run_id": cursor_run_id, "cursor_seq": cursor_seq, "status": "cursor_event_missing", "events": []} + replay_events = [event for event in cursor_events if event["seq"] > cursor_seq] + for _created_at, _run_id, events in runs[cursor_index + 1:]: + replay_events.extend(events) + return { + "session_id": sid, + "cursor_run_id": cursor_run_id, + "cursor_seq": cursor_seq, + "status": "ok", + "events": replay_events, + } + + def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> bool: """Remove the entire per-session run-journal directory (``_run_journal/{sid}/``). 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..60dbec33 --- /dev/null +++ b/tests/test_issue4812_session_sse_stream.py @@ -0,0 +1,675 @@ +from __future__ import annotations + +import io +import queue +from types import SimpleNamespace +from urllib.parse import urlparse + +from api.run_journal import append_run_event, read_session_run_events + + +class _FakeHandler: + def __init__(self): + self.status = None + self.headers = {} + self.wfile = io.BytesIO() + self.command = "GET" + self.path = "/" + self.client_address = ("127.0.0.1", 12345) + + def send_response(self, status): + self.status = status + + def send_header(self, key, value): + self.headers[key] = value + + def end_headers(self): + pass + + +def _capture(monkeypatch): + cap = {} + + def _j(_h, obj, *_, **kwargs): + cap["ok"] = obj + cap["status"] = kwargs.get("status", 200) + return True + + def _bad(_h, msg, code=400): + cap["bad"] = (msg, code) + return True + + monkeypatch.setattr("api.routes.j", _j) + monkeypatch.setattr("api.routes.bad", _bad) + return cap + + +def _stop_after_first_heartbeat(monkeypatch): + calls = {"count": 0} + + def _sleep(_seconds): + calls["count"] += 1 + raise BrokenPipeError("stop after the first heartbeat") + + monkeypatch.setattr("api.routes.time.sleep", _sleep) + return calls + + +def test_session_route_and_global_route_stay_separate(monkeypatch): + import api.routes as routes + + calls = {"global": 0, "session": 0} + + monkeypatch.setattr(routes, "_handle_extension_sidecar_proxy", lambda *_args, **_kwargs: False) + monkeypatch.setattr( + routes, + "_handle_session_events_stream", + lambda *_args, **_kwargs: calls.__setitem__("global", calls["global"] + 1) or True, + ) + monkeypatch.setattr( + routes, + "_handle_session_sse_stream_for_session", + lambda *_args, **_kwargs: calls.__setitem__("session", calls["session"] + 1) or True, + ) + + assert routes._session_events_path_session_id("/api/sessions/events") is None + assert routes._session_events_path_session_id("/api/sessions/session_1/events") == "session_1" + assert routes._session_events_path_session_id("/api/sessions/session_1/events/extra") is None + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("/api/sessions/events")) + routes.handle_get(handler, urlparse("/api/sessions/session_1/events")) + + assert calls == {"global": 1, "session": 1} + + +def test_session_journal_replays_later_runs_after_cursor(tmp_path): + append_run_event( + "session_1", + "run_a", + "token", + {"text": "a1"}, + session_dir=tmp_path, + seq=1, + created_at=100.0, + ) + append_run_event( + "session_1", + "run_a", + "token", + {"text": "a2"}, + session_dir=tmp_path, + seq=2, + created_at=101.0, + ) + append_run_event( + "session_1", + "run_b", + "token", + {"text": "b1"}, + session_dir=tmp_path, + seq=1, + created_at=200.0, + ) + append_run_event( + "session_1", + "run_b", + "done", + {"session": {"session_id": "session_1"}}, + session_dir=tmp_path, + seq=2, + created_at=201.0, + ) + append_run_event( + "session_2", + "run_other", + "token", + {"text": "other"}, + session_dir=tmp_path, + seq=1, + created_at=150.0, + ) + + replay = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path) + + assert replay["status"] == "ok" + assert [event["event_id"] for event in replay["events"]] == ["run_a:2", "run_b:1", "run_b:2"] + + +def test_session_journal_distinguishes_missing_and_foreign_cursor(tmp_path): + append_run_event( + "session_1", + "run_a", + "token", + {"text": "a1"}, + session_dir=tmp_path, + seq=1, + created_at=100.0, + ) + append_run_event( + "session_2", + "run_b", + "token", + {"text": "b1"}, + session_dir=tmp_path, + seq=1, + created_at=200.0, + ) + + missing = read_session_run_events("session_1", after_event_id="run_missing:1", session_dir=tmp_path) + foreign = read_session_run_events("session_1", after_event_id="run_b:1", session_dir=tmp_path) + + assert missing["status"] == "cursor_run_missing" + assert foreign["status"] == "cursor_session_mismatch" + + +def test_session_journal_rejects_hostile_cursor_without_replay(tmp_path, monkeypatch): + append_run_event( + "session_1", + "run_a", + "token", + {"text": "a1"}, + session_dir=tmp_path, + seq=1, + created_at=100.0, + ) + monkeypatch.setattr( + "api.run_journal.find_run_summary", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("invalid cursor should not scan summaries")), + ) + + malformed = read_session_run_events("session_1", after_event_id="bad/run:1", session_dir=tmp_path) + negative = read_session_run_events("session_1", after_event_id="run_a:-5", session_dir=tmp_path) + zero = read_session_run_events("session_1", after_event_id="run_a:0", session_dir=tmp_path) + + assert malformed["status"] == "cursor_invalid" + assert malformed["events"] == [] + assert negative["status"] == "cursor_invalid" + assert negative["events"] == [] + assert zero["status"] == "cursor_invalid" + assert zero["events"] == [] + + +def test_session_journal_blank_summary_session_stays_missing(tmp_path, monkeypatch): + monkeypatch.setattr("api.run_journal.find_run_summary", lambda *_args, **_kwargs: {"session_id": ""}) + + replay = read_session_run_events("session_1", after_event_id="run_missing:1", session_dir=tmp_path) + + assert replay["status"] == "cursor_run_missing" + + +def test_session_journal_rejects_bounds_and_invalid_contiguous_rows(tmp_path): + append_run_event("session_1", "run_a", "token", {"text": "one"}, session_dir=tmp_path, seq=1) + append_run_event("session_1", "run_a", "token", {"text": "two"}, session_dir=tmp_path, seq=2) + + rows_limited = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_rows=1) + bytes_limited = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=1) + assert rows_limited["status"] != "ok" and rows_limited["events"] == [] + assert bytes_limited["status"] != "ok" and bytes_limited["events"] == [] + + path = tmp_path / "_run_journal" / "session_1" / "run_a.jsonl" + path.write_text(path.read_text(encoding="utf-8").replace('"event_id":"run_a:2"', '"event_id":"wrong:2"'), encoding="utf-8") + invalid = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path) + assert invalid["status"] == "replay_noncontiguous" + assert invalid["events"] == [] + + +def test_session_journal_rejects_oversized_line_before_decode(tmp_path, monkeypatch): + path = tmp_path / "_run_journal" / "session_1" / "run_a.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + payload = '{"event_id":"run_a:1","seq":1,"run_id":"run_a","session_id":"session_1","event":"done","type":"done","payload":{"text":"' + ("x" * 5000) + '"}}\n' + path.write_bytes(payload.encode("utf-8")) + monkeypatch.setattr("api.run_journal.json.loads", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("oversized line should not decode"))) + + limited = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=1024) + + assert limited["status"] == "replay_limit_bytes" + assert limited["events"] == [] + + +def test_session_journal_accepts_line_at_cap_and_rejects_next_byte(tmp_path): + path = tmp_path / "_run_journal" / "session_1" / "run_a.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + payload = '{"event_id":"run_a:1","seq":1,"run_id":"run_a","session_id":"session_1","event":"done","type":"done","payload":{"text":"ok"}}\n' + path.write_bytes(payload.encode("utf-8")) + cap = len(payload.encode("utf-8")) + + at_cap = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=cap) + over = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path, max_bytes=cap - 1) + + assert at_cap["status"] == "ok" + assert at_cap["events"] == [] + assert over["status"] == "replay_limit_bytes" + assert over["events"] == [] + + +def test_session_journal_requires_exact_cursor_and_contiguous_suffix(tmp_path): + append_run_event("session_1", "run_a", "token", {}, session_dir=tmp_path, seq=1) + exact_missing = read_session_run_events("session_1", after_event_id="run_a:2", session_dir=tmp_path) + append_run_event("session_1", "run_a", "token", {}, session_dir=tmp_path, seq=3) + + gap = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path) + + assert exact_missing["status"] != "ok" + assert exact_missing["events"] == [] + assert gap["events"] == [] + + +def test_session_journal_accepts_final_cursor_and_replays_later_runs(tmp_path): + append_run_event("session_1", "run_a", "token", {}, session_dir=tmp_path, seq=1, created_at=100.0) + append_run_event("session_1", "run_b", "token", {}, session_dir=tmp_path, seq=1, created_at=200.0) + + replay = read_session_run_events("session_1", after_event_id="run_a:1", session_dir=tmp_path) + + assert replay["status"] == "ok" + assert [event["event_id"] for event in replay["events"]] == ["run_b:1"] + + +def test_session_route_prefers_last_event_id_then_query_fallback(monkeypatch): + import api.routes as routes + + captured = [] + stop = _stop_after_first_heartbeat(monkeypatch) + + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: SimpleNamespace( + session_id=sid, + compact=lambda **_kwargs: {"session_id": sid, "title": "Session"}, + ), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda _session_id, *, after_event_id=None, session_dir=None: captured.append(after_event_id) or {"status": "ok", "events": []}, + ) + + handler = _FakeHandler() + handler.headers["Last-Event-ID"] = "run_a:2" + routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events?after_event_id=run_b:1"), + "session_1", + ) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events?after_event_id=run_b:1"), + "session_1", + ) + + assert stop["count"] >= 2 + assert captured == ["run_a:2", "run_b:1"] + + +def test_session_route_arms_deadline_immediately_after_headers(monkeypatch): + import api.routes as routes + + calls = [] + stop = _stop_after_first_heartbeat(monkeypatch) + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {})) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: None) + monkeypatch.setattr(routes, "end_sse_headers", lambda _handler: calls.append("headers")) + monkeypatch.setattr(routes, "_sse_set_write_deadline", lambda _handler: calls.append("deadline")) + + routes._handle_session_sse_stream_for_session(_FakeHandler(), urlparse("/api/sessions/session_1/events"), "session_1") + + assert calls[:2] == ["headers", "deadline"] + assert stop["count"] == 1 + + +def test_session_route_emits_snapshot_without_id_for_missing_cursor_and_keepalive(monkeypatch): + import api.routes as routes + + stop = _stop_after_first_heartbeat(monkeypatch) + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: SimpleNamespace( + session_id=sid, + compact=lambda **_kwargs: {"session_id": sid, "title": "Session"}, + ), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: {"status": "cursor_run_missing", "events": []}, + ) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events?after_event_id=run_missing:1"), + "session_1", + ) + + body = handler.wfile.getvalue().decode("utf-8") + assert "event: session_snapshot\n" in body + assert ": keepalive\n\n" in body + assert "id: " not in body + assert stop["count"] == 1 + + +def test_session_route_blocks_hidden_sessions_before_replay_or_live_attach(monkeypatch): + import api.routes as routes + + cap = _capture(monkeypatch) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, profile="other"), + ) + monkeypatch.setattr(routes, "_get_active_profile_name", lambda: "default") + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("replay should not run")), + ) + monkeypatch.setattr( + routes, + "_active_run_stream_for_session", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("live attach should not run")), + ) + + handler = _FakeHandler() + routes.handle_get(handler, urlparse("/api/sessions/hidden_session/events")) + + assert cap["bad"] == ("Session not found", 404) + + +def test_session_route_live_delivery_skips_replayed_active_run_items(monkeypatch): + import api.routes as routes + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "live replayed"}, "run_active:1")) + self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": "run_active:1", "offline_buffered_events": 1} + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + stream = _FakeStream() + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: SimpleNamespace( + session_id=sid, + compact=lambda **_kwargs: {"session_id": sid, "title": "Session"}, + ), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active") + monkeypatch.setattr(routes, "STREAMS", {"run_active": stream}) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: { + "status": "ok", + "events": [ + { + "run_id": "run_prev", + "seq": 1, + "event": "token", + "payload": {"text": "replayed"}, + "event_id": "run_prev:1", + }, + { + "run_id": "run_active", + "seq": 1, + "event": "token", + "payload": {"text": "replayed current"}, + "event_id": "run_active:1", + }, + ], + }, + ) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"), + "session_1", + ) + + body = handler.wfile.getvalue().decode("utf-8") + assert "id: run_prev:1\n" in body + assert body.count("id: run_active:1\n") == 1 + assert "id: run_active:2\n" in body + assert "event: stream_end\n" in body + assert stream.unsubscribed is True + + +def test_session_route_unsubscribes_when_replay_disconnects(monkeypatch): + import api.routes as routes + + class _DisconnectingWfile: + def write(self, _data): + raise BrokenPipeError("client closed") + + def flush(self): + pass + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": "run_active:1"} + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + stream = _FakeStream() + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: SimpleNamespace( + session_id=sid, + compact=lambda **_kwargs: {"session_id": sid, "title": "Session"}, + ), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active") + monkeypatch.setattr(routes, "STREAMS", {"run_active": stream}) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: { + "status": "ok", + "events": [ + { + "run_id": "run_prev", + "seq": 1, + "event": "token", + "payload": {"text": "replayed"}, + "event_id": "run_prev:1", + }, + ], + }, + ) + + handler = _FakeHandler() + handler.wfile = _DisconnectingWfile() + assert routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"), + "session_1", + ) + assert stream.unsubscribed is True + + +def test_session_route_live_delivery_without_cursor_keeps_buffered_active_items(monkeypatch): + import api.routes as routes + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "buffered"}, "run_active:1")) + self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2")) + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": "run_active:1", "offline_buffered_events": 1} + + stream = _FakeStream() + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: SimpleNamespace( + session_id=sid, + compact=lambda **_kwargs: {"session_id": sid, "title": "Session"}, + ), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active") + monkeypatch.setattr(routes, "STREAMS", {"run_active": stream}) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("replay should not run without a cursor")), + ) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events"), + "session_1", + ) + + body = handler.wfile.getvalue().decode("utf-8") + assert "id: run_active:1\n" in body + assert "id: run_active:2\n" in body + assert "event: stream_end\n" in body + + +def test_session_route_reconciles_late_attach_cutoff_once(monkeypatch): + import api.routes as routes + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "six"}, "run_active:6")) + self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:7")) + self.unsubscribed = 0 + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": "run_active:6"} + + def unsubscribe(self, q): + self.unsubscribed += q is self.q + + stream = _FakeStream() + available = {"value": False} + reads = [] + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {})) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active" if available["value"] else None) + monkeypatch.setattr(routes, "STREAMS", {"run_active": stream}) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: reads.append(1) or {"status": "ok", "events": [] if len(reads) == 1 else [{"event": "token", "payload": {"text": "six"}, "event_id": "run_active:6"}]}, + ) + monkeypatch.setattr(routes.time, "sleep", lambda _seconds: available.__setitem__("value", True)) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session(handler, urlparse("/api/sessions/session_1/events?after_event_id=run_active:5"), "session_1") + + body = handler.wfile.getvalue().decode("utf-8") + assert body.count("id: run_active:6\n") == 1 + assert "id: run_active:7\n" in body + assert stream.unsubscribed == 1 + + +def test_session_route_emits_snapshot_when_reconciliation_fails(monkeypatch): + import api.routes as routes + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "buffered"}, "run_active:1")) + self.q.put_nowait(("stream_end", {"status": "done"}, "run_active:2")) + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": "run_active:1"} + + stream = _FakeStream() + reads = [] + sessions = iter( + [ + SimpleNamespace(session_id="session_1", compact=lambda **_kwargs: {"session_id": "session_1", "title": "stale"}), + SimpleNamespace(session_id="session_1", compact=lambda **_kwargs: {"session_id": "session_1", "title": "fresh"}), + ] + ) + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: next(sessions), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active") + monkeypatch.setattr(routes, "STREAMS", {"run_active": stream}) + monkeypatch.setattr( + routes, + "read_session_run_events", + lambda *_args, **_kwargs: reads.append(1) or ( + {"status": "ok", "events": [{"event": "token", "payload": {"text": "replayed"}, "event_id": "run_prev:1"}]} + if len(reads) == 1 + else {"status": "replay_noncontiguous", "events": []} + ), + ) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session( + handler, + urlparse("/api/sessions/session_1/events?after_event_id=run_prev:0"), + "session_1", + ) + + body = handler.wfile.getvalue().decode("utf-8") + assert "event: session_snapshot\n" in body + assert '"title": "fresh"' in body + assert '"title": "stale"' not in body + assert "id: run_prev:1\n" not in body + assert body.count("id: run_active:1\n") == 0 + assert "id: run_active:2\n" in body + + +def test_session_route_bounds_sent_event_id_deduplication(monkeypatch): + import api.routes as routes + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + for seq in (1, 2, 3, 1): + self.q.put_nowait(("token", {"seq": seq}, f"run_active:{seq}")) + self.q.put_nowait(("stream_end", {}, "run_active:4")) + + def subscribe_with_snapshot(self): + return self.q, {} + + stream = _FakeStream() + monkeypatch.setattr(routes, "_SESSION_SSE_SENT_EVENT_ID_LIMIT", 2) + monkeypatch.setattr(routes, "_session_id_visible_to_request_profile", lambda *_args, **_kwargs: True) + monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: SimpleNamespace(session_id=sid, compact=lambda **_kwargs: {})) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args, **_kwargs: "run_active") + monkeypatch.setattr(routes, "STREAMS", {"run_active": stream}) + + handler = _FakeHandler() + routes._handle_session_sse_stream_for_session(handler, urlparse("/api/sessions/session_1/events"), "session_1") + + body = handler.wfile.getvalue().decode("utf-8") + assert body.count("id: run_active:1\n") == 2 + assert "id: run_active:4\n" in body