diff --git a/api/routes.py b/api/routes.py index ea3ede54..9577f7e2 100644 --- a/api/routes.py +++ b/api/routes.py @@ -9671,6 +9671,7 @@ from api.run_journal import ( read_session_run_events, session_journal_fingerprint, stale_interrupted_event, + SSE_RELAY_CLOSE_EVENTS, ) from api.todo_state import attach_todo_state from api.providers import ( @@ -17452,7 +17453,7 @@ def _stream_runner_run_events(handler, run_id: str, cursor: str | None = None) - event = _runner_event_name(entry) _sse_with_id(handler, event, _runner_event_payload(entry), _runner_event_id(run_id, entry)) emitted = True - if event in ("stream_end", "error", "cancel"): + if event in SSE_RELAY_CLOSE_EVENTS: terminal = True next_cursor = getattr(event_stream, "cursor", None) if next_cursor not in (None, ""): @@ -17546,7 +17547,7 @@ def _handle_sse_stream(handler, parsed): _sse_with_id(handler, event, data, event_id) else: _sse(handler, event, data) - if event in ("stream_end", "error", "cancel"): + if event in SSE_RELAY_CLOSE_EVENTS: break except _CLIENT_DISCONNECT_ERRORS: pass @@ -17684,7 +17685,7 @@ def _handle_session_run_journal_stream_for_session(handler, parsed, session_id): 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) - _is_terminal = event in ("stream_end", "error", "cancel") + _is_terminal = event in SSE_RELAY_CLOSE_EVENTS _already_sent = ( (replay_cutoff_seq is not None and event_seq is not None and event_seq <= replay_cutoff_seq) or (event_id and event_id in sent_event_ids) diff --git a/api/run_journal.py b/api/run_journal.py index 81abf2fc..cabdacfa 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -36,7 +36,15 @@ _SEQ_CACHE_LOCK = threading.Lock() _SUMMARY_CACHE_MAX_ENTRIES = 128 _SUMMARY_CACHE: OrderedDict[str, tuple[tuple[int, int, int, int, int], dict]] = OrderedDict() _SUMMARY_CACHE_LOCK = threading.Lock() -_TERMINAL_SSE_EVENTS = {"done", "cancel", "apperror", "error", "stream_end"} +# Events that mark a run terminal in the journal / summary sense. +TERMINAL_SSE_EVENTS = frozenset({"done", "cancel", "apperror", "error", "stream_end"}) +# Events that should close an SSE relay drain loop. `done` is intentionally +# excluded: background title generation and `stream_end` are emitted after +# `done`, and breaking early would drop them. `apperror` is included because +# it terminates with no trailing `stream_end`. +SSE_RELAY_CLOSE_EVENTS = frozenset({"stream_end", "cancel", "apperror", "error"}) +# Back-compat alias used by older call sites / tests. +_TERMINAL_SSE_EVENTS = TERMINAL_SSE_EVENTS _FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC" _FSYNC_MODE_EAGER = "eager" _FSYNC_MODE_TERMINAL_ONLY = "terminal-only" diff --git a/api/streaming.py b/api/streaming.py index 27432500..e0bb13a0 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -7472,7 +7472,7 @@ def _run_agent_streaming( def put(event, data): # If cancelled, drop all further events except the cancel event itself - if cancel_event.is_set() and not _success_writeback_committed and event not in ('cancel', 'error'): + if cancel_event.is_set() and not _success_writeback_committed and event not in ('cancel', 'apperror'): return event_id = None if run_journal is not None: diff --git a/docs/CONTRACTS.md b/docs/CONTRACTS.md index e7bcc6fc..fcfc5d54 100644 --- a/docs/CONTRACTS.md +++ b/docs/CONTRACTS.md @@ -72,9 +72,12 @@ 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. 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. + lifecycle event delivery. The Phase 1 **server route and journal relay** for + `GET /api/sessions/{session_id}/events` are implemented; broader client, + platform, and semantic-taxonomy claims in the RFC remain behind the recorded + proof gates. Prefer the RFC's **Authoritative emitted events** table (live + `/api/chat/stream` wire names) over the aspirational semantic taxonomy when + writing clients against current source. When a change touches streaming, recovery, replay, compression, context reconstruction, cancellation, approval/clarify, session metadata, or run state, diff --git a/docs/rfcs/session-sse-contract-v1.md b/docs/rfcs/session-sse-contract-v1.md index 693adf5d..65b13366 100644 --- a/docs/rfcs/session-sse-contract-v1.md +++ b/docs/rfcs/session-sse-contract-v1.md @@ -142,6 +142,11 @@ gate rather than inventing values without source support. ## Event taxonomy (Phase 1 draft) +> **Semantic names below are aspirational for the per-session endpoint.** Live +> `/api/chat/stream` wire names are listed in **Authoritative emitted events** +> immediately after this table — use those when writing clients against current +> source. + | event_type | Source | Description | |---|---|---| | `chat_delta` | run journal / live stream | Token or chunk from an assistant reply. | @@ -154,8 +159,45 @@ gate rather than inventing values without source support. | `session_snapshot` | server fallback | Current session projection; emitted when replay is unavailable. | | `heartbeat` | server | Keepalive emitted on the `_SSE_HEARTBEAT_INTERVAL_SECONDS` cadence. | -The event-type table is a draft. The final table must be confirmed during -maintainer review before implementation. +## Authoritative emitted events (`/api/chat/stream`) + +These are the **real wire `event:` names** emitted by `api/streaming.py` today +(23 names). Clients and docs must use this table — not the semantic draft above — +when integrating with the live chat SSE relay. + +| Wire name | Role | +|---|---| +| `token` | Assistant text delta | +| `reasoning` | Model reasoning / thinking delta | +| `tool` | Tool call started | +| `tool_complete` | Tool call finished (result or error) | +| `interim_assistant` | Mid-turn assistant prose (pre-final) | +| `approval` | Destructive-command approval prompt | +| `clarify` | Structured clarification prompt | +| `compressing` | Context compression started | +| `compressed` | Context compression finished | +| `title` | Session title update (often after `done`) | +| `title_status` | Title generation status / skip reason | +| `warning` | Non-fatal provider/fallback warning | +| `apperror` | Terminal application error (no trailing `stream_end`) | +| `cancel` | Run cancelled | +| `done` | Turn finalized (session payload); title/`stream_end` may follow | +| `stream_end` | SSE fence — close the client EventSource | +| `metering` | Token/cost metering snapshot | +| `context_status` | Context window / usage status | +| `goal` | Goal / plan card update | +| `goal_continue` | Goal continuation signal | +| `pending_steer_leftover` | Leftover steer text after interrupt | +| `state_saved` | Durable state write acknowledgment | +| `todo_state` | Todo / checklist panel update | + +Relay close set (stop draining the live queue): `stream_end`, `cancel`, +`apperror`, and legacy `error` — see `api.run_journal.SSE_RELAY_CLOSE_EVENTS`. +`done` is **not** a relay-close event because `title` and `stream_end` follow it. + +The semantic taxonomy table remains a draft for the proposed per-session +endpoint vocabulary and must be confirmed during maintainer review before that +endpoint claims parity. ## Cursor and resume semantics diff --git a/tests/test_sse_relay_apperror_closes.py b/tests/test_sse_relay_apperror_closes.py new file mode 100644 index 00000000..2fa14172 --- /dev/null +++ b/tests/test_sse_relay_apperror_closes.py @@ -0,0 +1,116 @@ +"""Regression: apperror must close SSE relay drain loops. + +Backend emits `apperror` as a terminal event with no trailing `stream_end` +(streaming.py returns immediately after put('apperror', ...)). Relay loops that +only treat ("stream_end", "error", "cancel") as close signals hang forever — +the phantom name `error` is never emitted. The authoritative close set is +api.run_journal.SSE_RELAY_CLOSE_EVENTS. +""" +from __future__ import annotations + +from pathlib import Path + +from api.run_journal import SSE_RELAY_CLOSE_EVENTS, TERMINAL_SSE_EVENTS + + +def test_relay_close_events_include_apperror_not_only_phantom_error(): + assert "apperror" in SSE_RELAY_CLOSE_EVENTS + assert "stream_end" in SSE_RELAY_CLOSE_EVENTS + assert "cancel" in SSE_RELAY_CLOSE_EVENTS + # `error` may remain for legacy/adapter frames, but must not be the sole + # stand-in for apperror. + assert "apperror" in SSE_RELAY_CLOSE_EVENTS + # `done` must NOT close the live chat relay — title + stream_end follow it. + assert "done" not in SSE_RELAY_CLOSE_EVENTS + assert "done" in TERMINAL_SSE_EVENTS + + +def test_routes_relay_loops_import_sse_relay_close_events(): + src = Path("api/routes.py").read_text(encoding="utf-8") + assert "SSE_RELAY_CLOSE_EVENTS" in src + # The three historic hardcoded tuples must be gone. + assert 'event in ("stream_end", "error", "cancel")' not in src + assert 'event in ("stream_end", "error", "cancel")' not in src + assert '_is_terminal = event in ("stream_end", "error", "cancel")' not in src + + +def test_cancel_drop_list_allows_apperror_not_phantom_error(): + src = Path("api/streaming.py").read_text(encoding="utf-8") + assert "event not in ('cancel', 'apperror')" in src + assert "event not in ('cancel', 'error')" not in src + + +def test_apperror_terminates_relay_close_predicate(): + """Observable behavior of the shared predicate used by relay loops.""" + assert "apperror" in SSE_RELAY_CLOSE_EVENTS + # Simulate the drain decision: apperror must stop the loop. + events = ["token", "tool", "apperror", "token"] + closed_at = None + for i, event in enumerate(events): + if event in SSE_RELAY_CLOSE_EVENTS: + closed_at = i + break + assert closed_at == 2 + # done alone must not stop the relay (title / stream_end still pending). + assert "done" not in SSE_RELAY_CLOSE_EVENTS + + +def test_chat_sse_stream_unsubscribes_on_apperror(monkeypatch): + """Live /api/chat/stream must exit the drain loop when apperror arrives.""" + import io + import queue + from urllib.parse import urlparse + + import api.routes as routes + + class _Handler: + 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 + + class _FakeStream: + def __init__(self): + self.q = queue.Queue() + self.q.put_nowait(("token", {"text": "hi"}, "run1:1")) + self.q.put_nowait(("apperror", {"message": "provider failed"}, "run1:2")) + self.unsubscribed = False + + def subscribe_with_snapshot(self): + return self.q, {"last_event_id": None, "offline_buffered_events": 0} + + def unsubscribe(self, q): + self.unsubscribed = q is self.q + + stream = _FakeStream() + monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_a, **_k: True) + monkeypatch.setattr(routes, "STREAMS", {"run1": stream}) + monkeypatch.setattr( + routes, + "_sse_replay_run_journal_gap_checked", + lambda *_a, **_k: (False, None), + ) + + def _sleep(_seconds): + raise BrokenPipeError("hung waiting after apperror") + + monkeypatch.setattr(routes.time, "sleep", _sleep) + + handler = _Handler() + routes._handle_sse_stream(handler, urlparse("/api/chat/stream?stream_id=run1")) + + assert stream.unsubscribed is True, "handler must unsubscribe after apperror" + body = handler.wfile.getvalue().decode("utf-8", errors="replace") + assert "event: apperror" in body