diff --git a/CHANGELOG.md b/CHANGELOG.md index 03133da1..fb5c3b2b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ### Fixed +- **Reconnecting to a live chat stream now resumes from the client's last received event instead of losing the tail or replaying the whole turn.** The `/api/chat/stream` endpoint honors the SSE `Last-Event-ID` header (and `after_event_id` / `after_seq` / `replay=1`) with a resume cursor that carries provenance, so a dropped connection re-subscribes at the right place: an in-range cursor replays only the gap, an unparseable/foreign/ahead cursor fails closed to replay-from-start (never silently skipping frames), and a genuinely truncated buffer still emits `recovery_control`. A prior edge where an ahead cursor with no known snapshot cutoff installed itself as the live dedup bound — discarding every queued frame including the terminal `stream_end` and stalling the reconnect on heartbeats — is fixed by treating an unknown cutoff as fence 0 before normalization. Thanks @Paladin173. (#6886) - **A parked no-run approval can no longer mask the real pending approval when a gateway producer is live.** No-run "legacy" approval mirrors are now stamped with a token derived from their own live producer, and whenever a session's gateway queue has live producers the reconciler tokenizes from the authoritative head and discards any unmatched or tokenless mirror instead of letting it linger — so a stale card can't sit in front of the genuine pending approval (previously gateway mode returned 409 and permanently retained the stale mirror, and local mode reported success while resolving nothing). Tokenless-orphan retention is preserved for the no-producer case, and responding to a stale id fails without waking a different producer. Thanks @VHSgunzo. (#7086) - **Toggling YOLO from an approval card no longer hangs the gateway run or leaks across a session switch.** The approval card's YOLO branch now takes the same single `gateway_yolo_handoff(sid)` as the Runs dispatcher (revalidating the exact mirror after acquisition and holding it through settlement, with no nested-lock path), and the client captures one immutable owner tuple (card-owner session, approval id, load generation) at click time — failing closed without a POST when the visible card is not the active session, and fencing every post-response mutation (card hide, pending clears, stale-clear requery, controls, toast, status) on that exact tuple so a stale card can't project its result into the successor session. Thanks @dss539. (#6945) - **Git operations no longer flash a console window on Windows.** Every git subprocess boundary (agent runtime, rollback, updates, workspace, worktrees) now routes through a shared `windows_hide_flags()` helper that passes `CREATE_NO_WINDOW` on Win32 and stays a no-op on POSIX, so short-lived git commands no longer pop a visible console window on Windows while keeping captured stdout/stderr intact. Thanks @ya1yi2fa3. (#6534) diff --git a/README.md b/README.md index ad139126..7e73b9f1 100644 --- a/README.md +++ b/README.md @@ -692,6 +692,7 @@ The WebUI is still coupled to Hermes Agent internals for runtime execution, prov - [`TESTING.md`](TESTING.md) — manual browser test plan and automated coverage reference - [`DESIGN.md`](DESIGN.md) — design tokens and the calm-console direction - [`docs/UIUX-GUIDE.md`](docs/UIUX-GUIDE.md) — UI/UX principles sourced from the design docs and visual inventories +- [`docs/sse-streams.md`](docs/sse-streams.md) — cross-client SSE endpoint reference: session streaming, gateway SSE probe scope, heartbeats, and proxy behavior - [`docs/CONTRACTS.md`](docs/CONTRACTS.md) — project contract/RFC/design index for contributors and agents - [`docs/rfcs/README.md`](docs/rfcs/README.md) — RFC index for larger architecture and durability proposals diff --git a/api/routes.py b/api/routes.py index 14a7cffa..b793a89c 100644 --- a/api/routes.py +++ b/api/routes.py @@ -17475,6 +17475,126 @@ def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None return _shared_parse_run_journal_event_id(raw) +def _chat_stream_resume_cursor(handler, qs: dict, stream_id: str | None = None) -> tuple[int | None, bool, str | None, str | None]: + """Resolve the client's resume cursor for ``/api/chat/stream``. + + Returns ``(after_seq, resume_requested, raw_cursor, runner_cursor)``: + + - ``after_seq``: the parsed same-run cursor seq, or ``None`` when there is + no usable same-run (journal-shaped) cursor. + - ``resume_requested``: True when the client SUPPLIED any cursor — via the + ``after_event_id`` / ``after_seq`` query params, ``replay=1``, or the + ``Last-Event-ID`` header — regardless of whether it parsed. + - ``raw_cursor``: the opaque cursor string exactly as the client supplied it + (the ``Last-Event-ID`` value, or the explicit ``after_event_id``). + - ``runner_cursor``: the cursor to hand to the runner observe path, resolved + with PROVENANCE so the runner adapter (whose cursors are opaque, not + journal-shaped) gets a cursor it can actually use: + + * a valid ``after_seq`` pairs with whatever ``after_event_id`` was + supplied — even an opaque runner id like ``event:2`` that the + journal parser reads as a foreign run — so the paired runner cursor + resumes at the seq (never ``None`` / full replay, which would + duplicate events); + * a header-only opaque runner id (``Last-Event-ID: event:2``) is + preserved as-is so the runner resumes from it; + * a malformed or foreign explicit cursor WITHOUT a valid paired + ``after_seq`` yields ``None`` — it must block the header and replay + from start (this preserves the r2 malformed-blocks-header rule and + never forwards an unusable cursor to the runner). + + The presence flag must stay separate from validity: a malformed, foreign-run, + or ahead-of-stream cursor resolves to ``after_seq=None`` but still means the + client *asked* to resume. That request must be honored with a + replay-from-start so no journal events are silently skipped — whereas a + genuinely cursor-less request is a fresh subscribe (no replay). + + Precedence is decided by query-parameter PRESENCE, not successful parsing: + when an explicit ``after_seq`` / ``after_event_id`` is supplied (even an + unparseable one), the ``Last-Event-ID`` header is never consulted, so a + header can never override an explicit cursor and silently skip events. + + ``Last-Event-ID`` is the cursor every spec-compliant SSE client (browser + ``EventSource`` auto-reconnect, Android/CLI clients) sends automatically on + reconnect, carrying the ``id:`` of the last event it received. Every + journaled event on this stream already emits ``id: stream_id:seq`` via + ``_sse_with_id()``. Same resolution-chain precedent as + ``api/kanban_bridge.py`` (``?since=`` → ``Last-Event-ID``). + """ + after_seq_raw = qs.get("after_seq", [None])[0] + has_explicit_query = ( + after_seq_raw not in (None, "") + or bool(qs.get("after_event_id", [None])[0]) + or bool(qs.get("replay", [""])[0]) + ) + if has_explicit_query: + explicit_raw = str(qs.get("after_event_id", [None])[0] or "").strip() or None + after_seq = _parse_run_journal_after_seq(qs, stream_id) + # Runner cursor provenance. ``after_seq`` is authoritative when it + # parses — it pairs with whatever ``after_event_id`` shape was + # supplied, including opaque runner ids (``event:2``) that the journal + # parser reads as a foreign run. Read it directly here (not via + # ``_parse_run_journal_after_seq``, which checks ``after_event_id`` + # FIRST and would swallow a paired opaque runner id as "foreign"). + paired_seq = _parse_run_journal_after_seq_value(after_seq_raw) + if paired_seq is not None: + runner_cursor = str(paired_seq) + else: + event_run_id, event_seq = _parse_run_journal_event_id(explicit_raw) + runner_cursor = ( + explicit_raw + if explicit_raw and event_seq is not None and (not stream_id or event_run_id == stream_id) + else None + ) + return after_seq, True, explicit_raw, runner_cursor + headers = getattr(handler, "headers", None) + if headers is None: + return None, False, None, None + try: + raw = headers.get("Last-Event-ID") + except Exception: + return None, False, None, None + raw = str(raw or "").strip() + if not raw: + return None, False, None, None + event_run_id, event_seq = _parse_run_journal_event_id(raw) + if event_run_id and event_seq is not None: + if stream_id and event_run_id != stream_id: + # Foreign-run journal cursor: asked to resume THIS run but the cursor + # names a different journal run — can't honor it for the journal + # path (after_seq stays None → replay-from-start). The runner path + # keys cursors by run_id independently (the cursor is forwarded as + # an opaque per-run query param), so a ``run:seq`` header still + # reaches it as-is; this mirrors how a foreign after_event_id on + # the journal path is rejected while the same client's explicit + # opaque cursor= would still reach the runner. + return None, True, raw, raw + return event_seq, True, raw, raw + # Malformed as a JOURNAL cursor. A colon-less opaque value is a plausible + # runner cursor (runner ids need not be journal-shaped), so preserve it for + # the runner; a value that merely fails int() parsing is unusable anywhere. + runner_cursor = raw if ":" not in raw else None + return None, True, raw, runner_cursor + + +def _parse_run_journal_after_seq_value(raw) -> int | None: + """Parse a bare ``after_seq`` value, independent of any ``after_event_id``. + + Used by the runner-cursor provenance path, where ``after_seq`` is + authoritative on its own and must NOT be gated behind the + ``after_event_id``-first ordering of ``_parse_run_journal_after_seq`` (a + paired opaque runner id like ``event:2`` would otherwise be read as a + foreign run and swallow the seq). Mirrors the ``after_seq`` tail of that + parser: absent/blank → None, non-numeric → 0. + """ + if raw in (None, ""): + return None + try: + return max(0, int(raw)) + except (TypeError, ValueError): + return 0 + + 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: @@ -17586,7 +17706,8 @@ def _run_journal_covers_offline_gap( def _sse_replay_run_journal_gap_checked( - handler, qs: dict, stream_id: str, stream_snapshot: dict + handler, qs: dict, stream_id: str, stream_snapshot: dict, + *, resume_cursor: tuple[int | None, bool] | None = None, ) -> tuple[bool, int | None]: """Journal-replay for a reconnecting client, enforcing offline-gap coverage. @@ -17596,12 +17717,30 @@ def _sse_replay_run_journal_gap_checked( (see ``_run_journal_covers_offline_gap``), a recovery_control apperror has been emitted and the caller must return instead of draining the retained tail (``gap_recovered=True``). + + ``resume_cursor`` is the caller-resolved ``(after_seq, resume_requested)`` + pair from ``_chat_stream_resume_cursor``. Presence and validity are kept + separate: an invalid / foreign-run / unparseable cursor means the client + *asked* to resume but we couldn't honor it, so it is normalized to + replay-from-start (``after_seq=None`` inside the replay) rather than + treated as "no cursor" — which would skip replay and silently drain a + truncated buffer. A genuinely cursor-less request (``resume_requested`` + False) is a fresh subscribe and returns ``(False, None)`` with no replay. + + Direct callers that only supply ``qs`` keep the historical behavior: the + cursor is derived from the query params (presence of ``replay`` / + ``after_seq`` / ``after_event_id`` counts as resume-requested). """ - if not ( - qs.get("replay", [""])[0] - or qs.get("after_seq", [None])[0] not in (None, "") - or qs.get("after_event_id", [None])[0] - ): + if resume_cursor is None: + after_seq = _parse_run_journal_after_seq(qs, stream_id) + resume_requested = ( + bool(qs.get("replay", [""])[0]) + or qs.get("after_seq", [None])[0] not in (None, "") + or bool(qs.get("after_event_id", [None])[0]) + ) + else: + after_seq, resume_requested = resume_cursor + if not resume_requested: return False, None try: offline_dropped = int(stream_snapshot.get("offline_dropped_events") or 0) @@ -17611,7 +17750,33 @@ def _sse_replay_run_journal_gap_checked( str(stream_snapshot.get("last_event_id") or ""), stream_id, ) - after_seq = _parse_run_journal_after_seq(qs, stream_id) + # Normalize an unparseable / foreign cursor (asked to resume, but no usable + # same-run seq) to replay-from-start so the gap check and dedup operate on + # a real cursor instead of silently skipping the whole journal. + if after_seq is None: + after_seq = 0 + # Normalize a numeric cursor strictly AHEAD of the snapshot's last known + # frame to replay-from-start too: the client believes it already holds + # everything, so on a truncated buffer the coverage check's + # ``floor >= replay_max_seq`` would falsely declare the gap covered, replay + # nothing, and drain only the retained tail — silently losing every event + # before it (Codex r2 #2). ``>`` (not ``>=``) — a cursor EQUAL to the + # cutoff is a valid in-range cursor (see the dedup bound below). + # + # An UNKNOWN snapshot cutoff (no parseable ``last_event_id`` — e.g. the + # channel has not seen an id-bearing frame yet) is treated as fence 0: + # with no cutoff to bound it, any positive client cursor would otherwise + # be installed verbatim as the live dedup bound and filter EVERY queued + # frame — including the terminal ``stream_end`` fence — leaving the + # reconnect stalled on heartbeats with an empty body (Codex r4). Failing + # closed to replay-from-start delivers the buffered events (at worst + # duplicating what the client already holds) instead of silently losing + # them. Frames dropped while the cutoff is unknown still cannot prove + # coverage below (``cutoff_seq is None`` → not covered), so the + # recovery_control fail-closed path is preserved. + effective_cutoff = snapshot_cutoff_seq if snapshot_cutoff_seq is not None else 0 + if after_seq > effective_cutoff: + after_seq = 0 # The subscribe snapshot already queued the retained offline tail, which # covers [first buffered frame → snapshot cutoff] by itself. The journal # only has to bridge (client cursor → first buffered frame) — and the @@ -17659,18 +17824,21 @@ def _sse_replay_run_journal_gap_checked( # tail (after_seq >= first buffered frame) would otherwise get the queued # copy of frames it already rendered — a double-render, since this filter # is the only dedup for replayed streams. - if after_seq is not None: - cursor_bound = after_seq - if snapshot_cutoff_seq is not None: - # A legitimate cursor can never exceed the channel's last known - # frame; clamping keeps a bogus/corrupt cursor from filtering the - # queued terminal frame and pinning the loop on heartbeats. - cursor_bound = min(cursor_bound, snapshot_cutoff_seq) - replay_cutoff_seq = ( - cursor_bound - if replay_cutoff_seq is None - else max(replay_cutoff_seq, cursor_bound) - ) + # + # Dedup bound semantics: the drain filter skips ``seq <= replay_cutoff_seq``. + # The event AT the cursor (seq == after_seq) was already delivered to this + # client, so the cursor must itself enter the bound — equality included — + # otherwise the buffered copy of that event double-sends. A cursor strictly + # ahead of the snapshot was already normalized to 0 above; here only + # in-range cursors (after_seq <= snapshot_cutoff_seq) contribute a bound, + # and the terminal frame must always survive. + if after_seq is not None and after_seq > 0: + if snapshot_cutoff_seq is None or after_seq <= snapshot_cutoff_seq: + replay_cutoff_seq = ( + after_seq + if replay_cutoff_seq is None + else max(replay_cutoff_seq, after_seq) + ) return False, replay_cutoff_seq @@ -17832,16 +18000,47 @@ def _handle_sse_stream(handler, parsed): stream_id = qs.get("stream_id", [""])[0] if not _stream_id_visible_to_request_profile(handler, stream_id): return True + # Resume cursor: explicit query params (after_event_id/after_seq/replay) + # win; the Last-Event-ID header that spec-compliant SSE clients auto-send + # on reconnect is the fallback. Presence is tracked separately from the + # parsed seq — a client that supplied ANY cursor asked to resume, and an + # unusable (invalid/foreign/ahead-of-stream) cursor must replay from start + # rather than silently skip journal events. + resume_cursor = _chat_stream_resume_cursor(handler, qs, stream_id) + resume_after_seq, resume_requested, resume_raw_cursor, runner_resume_cursor = resume_cursor stream = STREAMS.get(stream_id) if stream is None: - if _stream_runner_run_events(handler, stream_id, _runner_stream_cursor_from_query(qs)): + # Runner-observe path: consume the ALREADY-RESOLVED cursor — do not + # re-parse query params or re-read the header (Codex r2 #3 / r3). The + # explicit opaque ``cursor`` query param still wins for runner clients + # that speak that contract. Otherwise use the resolver's + # provenance-resolved runner cursor: a valid ``after_seq`` pairs with + # opaque runner ids (event:2), a header-only opaque runner id resumes + # as-is, and a malformed/foreign cursor without a valid paired seq + # yields None (replay from start, never forwarding an unusable cursor). + runner_cursor = str(qs.get("cursor", [""])[0] or "").strip() or None + if runner_cursor is None and resume_requested: + runner_cursor = runner_resume_cursor + if _stream_runner_run_events(handler, stream_id, runner_cursor): return True try: - journal_available = bool(find_run_summary(stream_id)) if stream_id else False + journal_summary = find_run_summary(stream_id) if stream_id else None except Exception: - journal_available = False - if not journal_available: + journal_summary = None + if not journal_summary: return j(handler, {"error": "stream not found"}, status=404) + # Normalize a cursor strictly AHEAD of the dead stream's authoritative + # last_seq to replay-from-start: passing it straight through would make + # the journal reader emit an empty SSE body for a journal that actually + # holds events (Codex r2 #2). Equality is in-range — the event at the + # cursor was already delivered, so the replay correctly resumes after it. + dead_after_seq = resume_after_seq + try: + last_seq = int(journal_summary.get("last_seq") or 0) + except (TypeError, ValueError): + last_seq = 0 + if dead_after_seq is not None and dead_after_seq > last_seq: + dead_after_seq = 0 handler.send_response(200) handler.send_header("Content-Type", "text/event-stream; charset=utf-8") handler.send_header("Cache-Control", "no-cache") @@ -17849,7 +18048,7 @@ def _handle_sse_stream(handler, parsed): handler.send_header("Connection", "close") end_sse_headers(handler) try: - _replay_run_journal(handler, stream_id, _parse_run_journal_after_seq(qs, stream_id)) + _replay_run_journal(handler, stream_id, dead_after_seq) except _CLIENT_DISCONNECT_ERRORS: pass return True @@ -17868,7 +18067,8 @@ def _handle_sse_stream(handler, parsed): # Replay shares the drain loop's try/finally so every exit path unsubscribes. try: gap_recovered, replay_cutoff_seq = _sse_replay_run_journal_gap_checked( - handler, qs, stream_id, stream_snapshot + handler, qs, stream_id, stream_snapshot, + resume_cursor=(resume_after_seq, resume_requested), ) if gap_recovered: return True @@ -18267,6 +18467,14 @@ def _gateway_sse_probe_payload(settings, watcher): 'fallback_poll_ms': 30000, 'ok': enabled and watcher_alive, 'watcher_running': watcher_alive, + # Cross-client scope markers (hermes-webui/hermes-android#58 follow-up): + # this probe ONLY describes the optional gateway/agent-sessions stream. + # Persistent per-session streaming (GET /api/session/stream) is always + # available and is NOT gated by show_cli_sessions, so a negative gateway + # probe result must not be read as "session SSE unavailable". + 'scope': 'gateway_sessions', + 'session_stream_available': True, + 'session_stream_path': '/api/session/stream', } if not enabled: payload['error'] = 'agent sessions not enabled' @@ -18281,6 +18489,12 @@ def _handle_gateway_sse_stream(handler, parsed): """SSE endpoint for real-time gateway session updates. Streams change events from the gateway watcher background thread. Only active when show_cli_sessions (show_agent_sessions) setting is enabled. + + Probe mode (``?probe=1``) reports the status of THIS optional stream only. + Its result says nothing about the always-on persistent per-session stream + (``/api/session/stream``) — the probe payload carries explicit + ``scope`` / ``session_stream_available`` markers so cross-client consumers + do not misclassify usable session streaming as unavailable. """ settings = load_settings() diff --git a/docs/sse-streams.md b/docs/sse-streams.md new file mode 100644 index 00000000..07fca5b6 --- /dev/null +++ b/docs/sse-streams.md @@ -0,0 +1,69 @@ +# SSE streams and capability signaling + +Cross-client reference for the server-sent events (SSE) endpoints Hermes WebUI +exposes. Browser and non-browser clients (Android wrapper, CLI observers) +should integrate against this page so every client describes the same +behavior. + +All endpoints below are served by the WebUI origin and sit behind the same +authentication as every other `/api/*` route: when a WebUI password or OIDC +is configured, clients must authenticate before opening any stream. + +## Endpoint inventory + +| Endpoint | Availability | Purpose | +|---|---|---| +| `GET /api/chat/stream?stream_id=` | Always on | Live agent-turn relay (tokens, tool calls, approvals, `done`, `stream_end`). Falls back to run-journal replay when the in-memory stream is gone. Resume cursors: `after_event_id` / `after_seq` query params, with the standard `Last-Event-ID` header as fallback (events carry `id: :`). An invalid/foreign/ahead-of-stream cursor is honored as replay-from-start rather than silently skipping events. | +| `GET /api/session/stream?session_id=` | Always on | Persistent per-session channel that survives across agent turns (`initial`, `server_turn_started`, `session-updated`, `bg_task_complete`). This is the stream the WebUI frontend keeps open per session and the stream non-browser clients should prefer for background session updates. | +| `GET /api/sessions/events` | Always on | Global session-list invalidation (`sessions_changed` + keepalives). A signal to re-read `/api/sessions`, not a per-session lifecycle feed. | +| `GET /api/sessions/{session_id}/events` | Always on | Per-session run-journal relay with `Last-Event-ID` / `after_event_id` resume and snapshot fallback. See `docs/rfcs/session-sse-contract-v1.md` for the contract and its proof gates. | +| `GET /api/sessions/gateway/stream` | Optional | Real-time updates for CLI/TUI/messaging (agent) sessions merged into the sidebar. Only streams when the **Agent sessions** setting (`show_cli_sessions`) is enabled and the gateway watcher thread is running. | + +The authoritative `event:` names on `/api/chat/stream` are listed in the +**Authoritative emitted events** table of +[`docs/rfcs/session-sse-contract-v1.md`](rfcs/session-sse-contract-v1.md). + +## Gateway probe scope (important for non-browser clients) + +`GET /api/sessions/gateway/stream?probe=1` returns a JSON capability payload +for the **optional gateway stream only** instead of holding an SSE connection: + +```json +{ + "enabled": false, + "ok": false, + "watcher_running": false, + "fallback_poll_ms": 30000, + "error": "agent sessions not enabled", + "scope": "gateway_sessions", + "session_stream_available": true, + "session_stream_path": "/api/session/stream" +} +``` + +- `404` + `error: "agent sessions not enabled"` means only that the optional + gateway/agent-sessions stream is disabled on this server. +- `503` + `error: "watcher not started"` means the setting is on but the + gateway watcher thread is not running. +- `200` with `ok: true` means gateway SSE is usable. + +**A negative gateway probe result must not be treated as "SSE unavailable".** +The persistent per-session stream (`/api/session/stream`) and the chat-turn +relay (`/api/chat/stream`) are always on and are not gated by the Agent +sessions setting. The `scope`, `session_stream_available`, and +`session_stream_path` fields make this explicit so clients do not need to +infer it from the status code. Clients that only need session updates should +use `/api/session/stream` directly; the gateway probe is only relevant for +clients that display CLI/TUI/messaging sessions. + +## Heartbeats and proxy behavior + +- All long-lived streams emit SSE keepalive comment lines on the + `_SSE_HEARTBEAT_INTERVAL_SECONDS` cadence (currently 5 seconds), which is + short enough to survive typical reverse-proxy idle timeouts. +- Handlers send `X-Accel-Buffering: no` so nginx-style proxies pass events + through unbuffered. +- Deployments behind buffering proxies that read-until-close (notably + Tornado-based `jupyter-server-proxy`) can set `HERMES_WEBUI_SSE_CHUNKED=1` + to frame each event as an HTTP/1.1 chunk. The default wire format is + unchanged when the flag is unset. diff --git a/tests/test_gateway_sync.py b/tests/test_gateway_sync.py index 131e8b9a..17ac4d23 100644 --- a/tests/test_gateway_sync.py +++ b/tests/test_gateway_sync.py @@ -2540,10 +2540,40 @@ def test_gateway_sse_stream_probe_reports_status(): assert data['enabled'] is True assert 'watcher_running' in data assert data['fallback_poll_ms'] == 30000 + # Cross-client scope markers: the probe is scoped to the optional + # gateway stream and must not imply session SSE is unavailable. + assert data['scope'] == 'gateway_sessions' + assert data['session_stream_available'] is True + assert data['session_stream_path'] == '/api/session/stream' finally: post('/api/settings', {'show_cli_sessions': False}) +def test_gateway_sse_stream_probe_disabled_keeps_session_stream_markers(): + """Disabled probe (404) still signals that /api/session/stream is usable. + + Regression for hermes-webui/hermes-android#58: a client probing the gateway + stream while 'agent sessions' are off must be able to tell that persistent + per-session streaming remains available instead of classifying all SSE as + unsupported. + """ + post('/api/settings', {'show_cli_sessions': False}) + req = urllib.request.Request(BASE + '/api/sessions/gateway/stream?probe=1') + try: + urllib.request.urlopen(req, timeout=5) + raise AssertionError('Expected 404 when agent sessions are disabled') + except urllib.error.HTTPError as e: + assert e.code == 404, f"Expected 404, got {e.code}" + data = json.loads(e.read().decode('utf-8')) + assert data['ok'] is False + assert data['enabled'] is False + assert data['error'] == 'agent sessions not enabled' + # The negative gateway result is scoped: session streaming stays usable. + assert data['scope'] == 'gateway_sessions' + assert data['session_stream_available'] is True + assert data['session_stream_path'] == '/api/session/stream' + + def test_gateway_webui_sessions_not_duplicated(): """If a session_id exists both in WebUI store and state.db, it's not duplicated.""" # Create a WebUI session with a known ID @@ -2631,6 +2661,11 @@ def test_probe_payload_when_disabled(): assert body['watcher_running'] is False assert body['error'] == 'agent sessions not enabled' assert body['fallback_poll_ms'] == 30000 + # Scope markers stay present on the negative result so clients do not + # misread "gateway SSE off" as "session SSE unavailable". + assert body['scope'] == 'gateway_sessions' + assert body['session_stream_available'] is True + assert body['session_stream_path'] == '/api/session/stream' def test_probe_payload_when_watcher_missing(): @@ -2642,6 +2677,8 @@ def test_probe_payload_when_watcher_missing(): assert body['watcher_running'] is False assert body['error'] == 'watcher not started' assert body['fallback_poll_ms'] == 30000 + assert body['scope'] == 'gateway_sessions' + assert body['session_stream_available'] is True def test_probe_payload_when_watcher_instance_no_thread(): @@ -2674,6 +2711,8 @@ def test_probe_payload_when_watcher_thread_alive(): assert body['ok'] is True assert body['watcher_running'] is True assert body['fallback_poll_ms'] == 30000 + assert body['scope'] == 'gateway_sessions' + assert body['session_stream_available'] is True finally: done.set() live.join(timeout=1) diff --git a/tests/test_run_journal_routes.py b/tests/test_run_journal_routes.py index 36a70822..a0bd82d3 100644 --- a/tests/test_run_journal_routes.py +++ b/tests/test_run_journal_routes.py @@ -90,12 +90,12 @@ def test_stream_status_exposes_replay_summary(): def test_dead_stream_sse_replays_journal_before_404_fallback(): handler_pos = ROUTES_SRC.index("def _handle_sse_stream") - block = ROUTES_SRC[handler_pos : handler_pos + 1800] + block = ROUTES_SRC[handler_pos : handler_pos + 5400] assert "find_run_summary(stream_id)" in block assert "stream not found" in block assert "_replay_run_journal" in block - assert "_parse_run_journal_after_seq" in block + assert "_chat_stream_resume_cursor" in block assert 'Content-Type", "text/event-stream; charset=utf-8"' in block @@ -827,3 +827,635 @@ def test_active_stream_replay_keeps_items_for_new_run_with_same_seq_range(monkey assert "id: run_new:2\n" in body assert body.count("id: run_new:1\n") == 1 assert stream.unsubscribed is True + + +# ── Last-Event-ID fallback resume cursor on /api/chat/stream ──────────────── +# The stream emits `id: stream_id:seq` on every journaled event, so any +# spec-compliant SSE client (browser EventSource auto-reconnect, Android/CLI +# clients) sends `Last-Event-ID` automatically on reconnect. The handler must +# honor it when no explicit query cursor is present instead of replaying from +# seq 0 and double-rendering the transcript. + + +class _HeaderHandler: + """Minimal request handler double carrying headers + a writable wfile.""" + + def __init__(self, last_event_id=None): + self.wfile = io.BytesIO() + self.headers = ( + {"Last-Event-ID": last_event_id} if last_event_id is not None else {} + ) + + def send_response(self, _code): + pass + + def send_header(self, _name, _value): + pass + + def end_headers(self): + pass + + +def test_chat_stream_resume_cursor_prefers_query_params_over_header(): + """Explicit after_event_id/after_seq always win over Last-Event-ID.""" + import api.routes as routes + + handler = _HeaderHandler(last_event_id="run_1:9") + qs = {"after_event_id": ["run_1:4"]} + # (after_seq, requested, raw_cursor, runner_cursor) + assert routes._chat_stream_resume_cursor(handler, qs, "run_1") == (4, True, "run_1:4", "run_1:4") + + qs = {"after_seq": ["7"]} + assert routes._chat_stream_resume_cursor(handler, qs, "run_1") == (7, True, None, "7") + + +def test_chat_stream_resume_cursor_explicit_unparseable_query_blocks_header(): + """A supplied-but-unparseable explicit query cursor must NOT fall through to + the Last-Event-ID header — precedence is by query-param PRESENCE, not + successful parsing (Codex CORE #1). The header can never override an + explicit cursor and silently skip events.""" + import api.routes as routes + + # Foreign-run after_event_id parses to None for this stream, but its + # presence must still block the (parseable) header. + handler = _HeaderHandler(last_event_id="run_1:9") + qs = {"after_event_id": ["run_other:2"]} + after_seq, requested, raw, runner = routes._chat_stream_resume_cursor(handler, qs, "run_1") + assert after_seq is None + assert requested is True # asked to resume → replay-from-start downstream + assert raw == "run_other:2" # explicit query cursor surfaced, header ignored + # No valid paired seq and a foreign journal run → no runner cursor (full replay). + assert runner is None + + +def test_chat_stream_resume_cursor_reads_last_event_id_header(): + """With no query cursor, Last-Event-ID is the resume position.""" + import api.routes as routes + + handler = _HeaderHandler(last_event_id="run_1:3") + assert routes._chat_stream_resume_cursor(handler, {}, "run_1") == (3, True, "run_1:3", "run_1:3") + + +def test_chat_stream_resume_cursor_foreign_run_header_is_requested_but_unusable(): + """A Last-Event-ID for a different run id parses out as unusable, but the + resume was still REQUESTED — presence survives so the caller replays from + start instead of skipping the journal (Codex CORE #2).""" + import api.routes as routes + + handler = _HeaderHandler(last_event_id="run_other:5") + after_seq, requested, raw, runner = routes._chat_stream_resume_cursor(handler, {}, "run_1") + assert after_seq is None + assert requested is True + assert raw == "run_other:5" + # The runner keys cursors per run independently, so the opaque header value + # is preserved for the runner even though the journal path rejects it. + assert runner == "run_other:5" + + +def test_chat_stream_resume_cursor_absent_without_any_cursor(): + """No query cursor and no header → (None, False): fresh attach, no replay.""" + import api.routes as routes + + handler = _HeaderHandler() + assert routes._chat_stream_resume_cursor(handler, {}, "run_1") == (None, False, None, None) + # Malformed header values are unusable but still a resume request. A + # colon-less opaque value is preserved for the runner (runner ids need not + # be journal-shaped); a colon value that fails int() parsing is not. + handler = _HeaderHandler(last_event_id="not-a-cursor") + assert routes._chat_stream_resume_cursor(handler, {}, "run_1") == (None, True, "not-a-cursor", "not-a-cursor") + + +def test_dead_stream_replay_uses_last_event_id_header(monkeypatch): + """Dead-stream path: reconnect with only Last-Event-ID resumes mid-journal.""" + import api.routes as routes + + captured = {} + handler = _HeaderHandler(last_event_id="run_1:2") + monkeypatch.setattr( + routes, + "find_run_summary", + lambda stream_id: { + "session_id": "session_1", + "run_id": stream_id, + "terminal": True, + "last_seq": 4, + }, + ) + + journal = [ + {"event": "token", "payload": {"text": "j1"}, "event_id": "run_1:1", "seq": 1}, + {"event": "token", "payload": {"text": "j2"}, "event_id": "run_1:2", "seq": 2}, + {"event": "token", "payload": {"text": "j3"}, "event_id": "run_1:3", "seq": 3}, + {"event": "done", "payload": {"session": {"session_id": "session_1"}}, "event_id": "run_1:4", "seq": 4}, + ] + + def fake_read_run_events(session_id, run_id, after_seq=None, max_seq=None): + captured["after_seq"] = after_seq + return { + "events": [ + e for e in journal + if after_seq is None or int(e["seq"]) > int(after_seq) + ] + } + + monkeypatch.setattr(routes, "read_run_events", fake_read_run_events) + monkeypatch.setattr(routes, "stale_interrupted_event", lambda *_a, **_k: None) + previous_streams = dict(routes.STREAMS) + routes.STREAMS.clear() + try: + routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run_1") + ) + finally: + routes.STREAMS.clear() + routes.STREAMS.update(previous_streams) + + assert captured["after_seq"] == 2 + body = handler.wfile.getvalue().decode("utf-8") + # Resume after seq 2: only seq 3 and 4 are emitted, not 1-2. + assert "id: run_1:1\n" not in body + assert "id: run_1:2\n" not in body + assert "id: run_1:3\n" in body + assert "id: run_1:4\n" in body + assert "event: done\n" in body + + +def test_live_stream_replay_uses_last_event_id_header(monkeypatch): + """Live path: header cursor drives the replay gap-check and dedup cutoff.""" + import api.routes as routes + + class FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "tail"}, "run_1:4")) + self.q.put_nowait(("stream_end", {}, "run_1:5")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, { + "last_event_id": "run_1:4", + "offline_buffered_events": 2, + "offline_first_event_id": "run_1:4", + } + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + captured = {} + handler = _HeaderHandler(last_event_id="run_1:2") + stream = FakeStream() + monkeypatch.setattr( + routes, + "find_run_summary", + lambda stream_id: { + "session_id": "session_1", + "run_id": stream_id, + "terminal": False, + }, + ) + + def fake_read_run_events(session_id, run_id, after_seq=None, max_seq=None): + captured["after_seq"] = after_seq + return { + "events": [ + { + "event": "token", + "payload": {"text": "bridged"}, + "event_id": f"{run_id}:3", + "seq": 3, + } + ] + } + + monkeypatch.setattr(routes, "read_run_events", fake_read_run_events) + monkeypatch.setattr(routes, "stale_interrupted_event", lambda *_a, **_k: None) + previous_streams = dict(routes.STREAMS) + routes.STREAMS.clear() + routes.STREAMS["run_1"] = stream + try: + routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run_1") + ) + finally: + routes.STREAMS.clear() + routes.STREAMS.update(previous_streams) + + # Journal replay bridged (cursor → buffered tail) using the header cursor. + assert captured["after_seq"] == 2 + body = handler.wfile.getvalue().decode("utf-8") + assert "id: run_1:3\n" in body + assert "id: run_1:4\n" in body + assert "id: run_1:5\n" in body + assert stream.unsubscribed is True + + +def test_live_stream_invalid_cursor_replays_from_start_not_skips(monkeypatch): + """A foreign/malformed reconnect cursor (asked to resume, can't honor it) + must replay the journal from START and still drain the tail — never + silently skip the whole journal (Codex CORE #2).""" + import api.routes as routes + + class FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "tail"}, "run_1:3")) + self.q.put_nowait(("stream_end", {}, "run_1:4")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, { + "last_event_id": "run_1:3", + "offline_buffered_events": 2, + "offline_first_event_id": "run_1:3", + } + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + captured = {} + # Foreign-run header cursor: parses unusable for run_1, but requested=True. + handler = _HeaderHandler(last_event_id="run_other:9") + stream = FakeStream() + monkeypatch.setattr( + routes, + "find_run_summary", + lambda stream_id: { + "session_id": "session_1", + "run_id": stream_id, + "terminal": False, + }, + ) + + def fake_read_run_events(session_id, run_id, after_seq=None, max_seq=None): + captured["after_seq"] = after_seq + captured["max_seq"] = max_seq + return { + "events": [ + {"event": "token", "payload": {"text": "j1"}, "event_id": f"{run_id}:1", "seq": 1}, + {"event": "token", "payload": {"text": "j2"}, "event_id": f"{run_id}:2", "seq": 2}, + ] + } + + monkeypatch.setattr(routes, "read_run_events", fake_read_run_events) + monkeypatch.setattr(routes, "stale_interrupted_event", lambda *_a, **_k: None) + previous_streams = dict(routes.STREAMS) + routes.STREAMS.clear() + routes.STREAMS["run_1"] = stream + try: + routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run_1") + ) + finally: + routes.STREAMS.clear() + routes.STREAMS.update(previous_streams) + + # Normalized to replay-from-start: the journal bridge covers 1..first-1. + assert captured["after_seq"] == 0 + body = handler.wfile.getvalue().decode("utf-8") + assert "id: run_1:1\n" in body + assert "id: run_1:2\n" in body + # Buffered tail still drained, terminal survives. + assert "id: run_1:3\n" in body + assert "id: run_1:4\n" in body + assert stream.unsubscribed is True + + +def test_live_stream_cursor_equal_to_cutoff_does_not_double_send(monkeypatch): + """A valid cursor EQUAL to the snapshot cutoff must enter the drain dedup + bound (equality is in-range, NOT ahead). The buffered copy of the event at + the cursor was already delivered to this client, so it must be filtered — + only the not-yet-seen frames after it are drained (Codex r2 #1 off-by-one). + """ + import api.routes as routes + + class FakeStream: + def __init__(self): + self.q = queue.Queue() + # Retained tail: client already holds through seq 3; seq 4 is new. + self.q.put_nowait(("token", {"text": "f2"}, "run_1:2")) + self.q.put_nowait(("token", {"text": "f3"}, "run_1:3")) + self.q.put_nowait(("stream_end", {}, "run_1:4")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, { + "last_event_id": "run_1:3", + "offline_buffered_events": 3, + "offline_first_event_id": "run_1:2", + } + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + captured = {} + handler = _HeaderHandler(last_event_id="run_1:3") # cursor == snapshot cutoff + stream = FakeStream() + monkeypatch.setattr( + routes, + "find_run_summary", + lambda stream_id: { + "session_id": "session_1", + "run_id": stream_id, + "terminal": False, + }, + ) + + def fake_read_run_events(session_id, run_id, after_seq=None, max_seq=None): + captured["after_seq"] = after_seq + captured["max_seq"] = max_seq + return {"events": []} + + monkeypatch.setattr(routes, "read_run_events", fake_read_run_events) + monkeypatch.setattr(routes, "stale_interrupted_event", lambda *_a, **_k: None) + previous_streams = dict(routes.STREAMS) + routes.STREAMS.clear() + routes.STREAMS["run_1"] = stream + try: + routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run_1") + ) + finally: + routes.STREAMS.clear() + routes.STREAMS.update(previous_streams) + + body = handler.wfile.getvalue().decode("utf-8") + # The events at/below the cursor (seqs 2 and 3) were already delivered — + # they must be filtered out, NOT re-sent. Only the new terminal frame (4) + # is drained. + assert "id: run_1:2\n" not in body + assert "id: run_1:3\n" not in body + assert "id: run_1:4\n" in body + assert stream.unsubscribed is True + + +def test_live_stream_ahead_cursor_unknown_cutoff_still_delivers(monkeypatch): + """An ahead-of-stream Last-Event-ID with an UNKNOWN snapshot cutoff + (no parseable last_event_id) must not become the live dedup bound. + + Before the fix, header run_1:999 + snapshot_cutoff_seq=None installed 999 + as the drain filter bound, so every queued frame — including the terminal + stream_end fence — was discarded and the reconnect stalled on heartbeats + with an empty SSE body (Codex r4 data-loss). The unknown cutoff is treated + as fence 0, normalizing the cursor to replay-from-start so both queued + events are delivered exactly once and the subscriber is released. + """ + import api.routes as routes + + class FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "t1"}, "run_1:1")) + self.q.put_nowait(("stream_end", {}, "run_1:2")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + # No last_event_id / first_event_id: cutoff is UNKNOWN (None). + return self.q, {"offline_buffered_events": 2} + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + handler = _HeaderHandler(last_event_id="run_1:999") + stream = FakeStream() + monkeypatch.setattr(routes, "find_run_summary", lambda _sid: None) + monkeypatch.setattr( + routes, "read_run_events", lambda *a, **k: {"events": []} + ) + monkeypatch.setattr(routes, "stale_interrupted_event", lambda *_a, **_k: None) + previous_streams = dict(routes.STREAMS) + routes.STREAMS.clear() + routes.STREAMS["run_1"] = stream + try: + routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run_1") + ) + finally: + routes.STREAMS.clear() + routes.STREAMS.update(previous_streams) + + body = handler.wfile.getvalue().decode("utf-8") + # Both events emit exactly once — including the terminal fence. + assert body.count("id: run_1:1\n") == 1 + assert body.count("id: run_1:2\n") == 1 + assert body.count("event: stream_end\n") == 1 + assert stream.unsubscribed is True + + +def test_dead_stream_ahead_of_stream_cursor_returns_full_replay(monkeypatch): + """A cursor AHEAD of the dead stream's authoritative last_seq is normalized + to replay-from-start, so the journal's real events are emitted instead of + an empty SSE body (Codex r2 #2 dead-stream half).""" + import api.routes as routes + + captured = {} + handler = _HeaderHandler(last_event_id="run_1:999") + monkeypatch.setattr( + routes, + "find_run_summary", + lambda stream_id: { + "session_id": "session_1", + "run_id": stream_id, + "terminal": True, + "last_seq": 2, + }, + ) + + journal = [ + {"event": "token", "payload": {"text": "hello"}, "event_id": "run_1:1", "seq": 1}, + {"event": "done", "payload": {"session": {"session_id": "session_1"}}, "event_id": "run_1:2", "seq": 2}, + ] + + def fake_read_run_events(session_id, run_id, after_seq=None, max_seq=None): + captured["after_seq"] = after_seq + return { + "events": [ + e for e in journal + if after_seq is None or int(e["seq"]) > int(after_seq) + ] + } + + monkeypatch.setattr(routes, "read_run_events", fake_read_run_events) + monkeypatch.setattr(routes, "stale_interrupted_event", lambda *_a, **_k: None) + previous_streams = dict(routes.STREAMS) + routes.STREAMS.clear() + try: + routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run_1") + ) + finally: + routes.STREAMS.clear() + routes.STREAMS.update(previous_streams) + + # 999 > last_seq (2) → normalized to replay-from-start, and the replay body + # actually carries the journal's events (not an empty response). + assert captured["after_seq"] == 0 + body = handler.wfile.getvalue().decode("utf-8") + assert "id: run_1:1\n" in body + assert "id: run_1:2\n" in body + assert "event: done\n" in body + + +def test_runner_observe_reconnect_uses_last_event_id_header(monkeypatch): + """Runner-local path: header-only reconnect carries Last-Event-ID through + to observe_run so the runner resumes instead of duplicating from the start + (Codex CORE #3).""" + import api.routes as routes + + calls = [] + + class FakeRunnerClient: + def observe_run(self, run_id, *, cursor=None): + calls.append((run_id, cursor)) + return { + "run_id": run_id, + "cursor": "7", + "events": [ + {"event": "message", "payload": {"content": "hi"}, "event_id": "run-1:6"}, + {"event": "stream_end", "payload": {"ok": True}, "event_id": "run-1:7"}, + ], + } + + class Handler: + def __init__(self, last_event_id=None): + self.wfile = io.BytesIO() + self.headers = ( + {"Last-Event-ID": last_event_id} if last_event_id is not None else {} + ) + + def send_response(self, _code): + pass + + def send_header(self, _name, _value): + pass + + def end_headers(self): + pass + + monkeypatch.setenv("HERMES_WEBUI_RUNTIME_ADAPTER", "runner-local") + monkeypatch.setattr(routes, "_runtime_runner_client_factory", lambda: FakeRunnerClient()) + handler = Handler(last_event_id="run-1:5") + try: + assert routes._handle_sse_stream( + handler, urlparse("/api/chat/stream?stream_id=run-1") + ) is True + finally: + monkeypatch.delenv("HERMES_WEBUI_RUNTIME_ADAPTER", raising=False) + + # The header cursor is carried through to the runner, not dropped. + assert calls == [("run-1", "run-1:5")] + body = handler.wfile.getvalue().decode("utf-8") + assert "event: stream_end" in body + + +class _RunnerProbeHandler: + def __init__(self, last_event_id=None): + self.wfile = io.BytesIO() + self.headers = ( + {"Last-Event-ID": last_event_id} if last_event_id is not None else {} + ) + + def send_response(self, _code): + pass + + def send_header(self, _name, _value): + pass + + def end_headers(self): + pass + + +def _run_runner_probe(monkeypatch, url, last_event_id=None): + """Drive _handle_sse_stream down the runner-observe path, returning the + (run_id, cursor) the runner's observe_run was called with.""" + import api.routes as routes + + calls = [] + + class FakeRunnerClient: + def observe_run(self, run_id, *, cursor=None): + calls.append((run_id, cursor)) + return { + "run_id": run_id, + "cursor": "7", + "events": [ + {"event": "stream_end", "payload": {"ok": True}, "event_id": f"{run_id}:7"}, + ], + } + + monkeypatch.setenv("HERMES_WEBUI_RUNTIME_ADAPTER", "runner-local") + monkeypatch.setattr(routes, "_runtime_runner_client_factory", lambda: FakeRunnerClient()) + handler = _RunnerProbeHandler(last_event_id=last_event_id) + try: + routes._handle_sse_stream(handler, urlparse(url)) + finally: + monkeypatch.delenv("HERMES_WEBUI_RUNTIME_ADAPTER", raising=False) + return calls + + +def test_runner_malformed_explicit_cursor_blocks_header(monkeypatch): + """Runner path: an explicit but malformed after_event_id must NOT let the + Last-Event-ID header override it — the runner gets NO cursor (full replay), + not the header value (Codex r2 #3 probe).""" + calls = _run_runner_probe( + monkeypatch, + "/api/chat/stream?stream_id=run_1&after_event_id=malformed", + last_event_id="run_1:9", + ) + assert calls == [("run_1", None)] + + +def test_runner_foreign_cursor_passes_no_cursor_for_full_replay(monkeypatch): + """Runner path: a foreign after_event_id (foreign:2) is unusable for this + run — the runner must get NO cursor (full replay), never the leaked seq + '2' that would skip runner events (Codex r2 #3 probe).""" + calls = _run_runner_probe( + monkeypatch, + "/api/chat/stream?stream_id=run_1&after_event_id=foreign:2", + ) + assert calls == [("run_1", None)] + + +def test_runner_explicit_opaque_cursor_wins(monkeypatch): + """Runner path: an explicit opaque ``cursor=`` query param keeps precedence + over both after_* params and the Last-Event-ID header (the runner-local + contract).""" + calls = _run_runner_probe( + monkeypatch, + "/api/chat/stream?stream_id=run_1&cursor=opaque-xyz", + last_event_id="run_1:5", + ) + assert calls == [("run_1", "opaque-xyz")] + + +def test_runner_paired_opaque_event_id_resumes(monkeypatch): + """Runner path: an opaque runner event id (event:2) PAIRED with a valid + after_seq resumes at that seq — never None (full replay), which would + duplicate tokens/tool events (Codex r3 probe 1).""" + calls = _run_runner_probe( + monkeypatch, + "/api/chat/stream?stream_id=run_1&after_event_id=event:2&after_seq=2", + ) + assert calls == [("run_1", "2")] + + +def test_runner_header_only_opaque_event_id_resumes(monkeypatch): + """Runner path: a header-only opaque runner event id (Last-Event-ID: + event:2) resumes from it — never None (full replay / duplicates) + (Codex r3 probe 2).""" + calls = _run_runner_probe( + monkeypatch, + "/api/chat/stream?stream_id=run_1", + last_event_id="event:2", + ) + assert calls == [("run_1", "event:2")] + + +def test_runner_malformed_explicit_with_valid_seq_uses_seq(monkeypatch): + """Runner path: a malformed after_event_id PAIRED with a valid after_seq + resumes at the seq, not the malformed raw value (Codex r3 probe 2 + counterpart).""" + calls = _run_runner_probe( + monkeypatch, + "/api/chat/stream?stream_id=run_1&after_event_id=malformed&after_seq=5", + ) + assert calls == [("run_1", "5")] diff --git a/tests/test_stage364_opus_live_sse_event_id.py b/tests/test_stage364_opus_live_sse_event_id.py index eb36504c..bc141f5c 100644 --- a/tests/test_stage364_opus_live_sse_event_id.py +++ b/tests/test_stage364_opus_live_sse_event_id.py @@ -81,7 +81,7 @@ def test_sse_handler_reads_event_id_from_side_channel(): and pass it to _sse_with_id when present.""" handler_idx = ROUTES_PY.find("def _handle_sse_stream(handler, parsed):") assert handler_idx != -1, "_handle_sse_stream not found" - handler_body = ROUTES_PY[handler_idx:handler_idx + 4000] + handler_body = ROUTES_PY[handler_idx:handler_idx + 5400] assert "STREAM_LAST_EVENT_ID.get(stream_id)" in handler_body, ( "_handle_sse_stream must read STREAM_LAST_EVENT_ID[stream_id] to " "get the event_id for emit" diff --git a/tests/test_stream_offline_gap_recovery.py b/tests/test_stream_offline_gap_recovery.py index 06c62209..2914eb1b 100644 --- a/tests/test_stream_offline_gap_recovery.py +++ b/tests/test_stream_offline_gap_recovery.py @@ -377,14 +377,17 @@ def test_cursor_past_buffer_head_does_not_double_render_queued_frames(monkeypatc assert stream.unsubscribed is True -def test_bogus_cursor_is_clamped_so_terminal_frame_survives(monkeypatch): - """The client-cursor dedup bound is clamped at the snapshot cutoff. +def test_ahead_of_stream_cursor_does_not_filter_terminal_frame(monkeypatch): + """An ahead-of-stream cursor (after_seq past the snapshot cutoff) is NOT + clamped onto the snapshot's terminal fence. - A legitimate cursor can never exceed the channel's last known frame; an - out-of-range one (corrupt client state) must not raise the dedup cutoff - past the queued terminal frame — the drain loop's `seq <=` skip runs - BEFORE its terminal break, so a filtered stream_end would pin the loop on - heartbeats until the write deadline.""" + The old clamp (``min(cursor, snapshot_cutoff)``) set the dedup bound to the + terminal frame's own seq, so the drain loop's ``seq <=`` filter — which runs + BEFORE its terminal break — swallowed the queued terminal frame and pinned + the loop on heartbeats until the write deadline (Codex CORE #2 terminal + filter). Ahead-of-stream means the client already believes it holds + everything the buffer has; nothing must be filtered and the terminal frame + must survive.""" from urllib.parse import parse_qs monkeypatch.setattr(routes, "find_run_summary", lambda _sid: None) @@ -401,6 +404,8 @@ def test_bogus_cursor_is_clamped_so_terminal_frame_survives(monkeypatch): }, ) assert handled is False - # Clamped to the cutoff (105): the queued stream_end (seq 106) passes the - # drain loop's dedup filter and terminates the connection. - assert cutoff == 105 + # Not clamped to 105: the cutoff stays None (journal replay found nothing, + # and an ahead-of-stream cursor contributes no dedup bound), so the queued + # terminal frame (seq 106) passes the drain loop's filter and terminates + # the connection instead of being filtered onto the fence. + assert cutoff is None