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