diff --git a/CHANGELOG.md b/CHANGELOG.md index 1661f7f6..8e12ffbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,8 @@ ### Fixed +- **Live SSE relays no longer hang after an application error.** The relay drain loops for `/api/chat/stream` and the run-event stream only closed on the hardcoded set `("stream_end", "error", "cancel")`, but the live chat path emits `apperror` as a terminal event with **no** trailing `stream_end` (it returns immediately after `put('apperror', …)`), and a bare `error` event is never emitted on that path — so a turn that ended in an application error left the client's EventSource loop draining forever. The three relay loops (`_stream_runner_run_events`, `_handle_sse_stream`, `emit_session_snapshot`) now share a single `api.run_journal.SSE_RELAY_CLOSE_EVENTS` close set (`stream_end`, `cancel`, `apperror`, and the still-real runner-adapter `error`); `done` is intentionally excluded because `title` and `stream_end` legitimately follow it. The post-cancel event-drop guard is corrected to keep delivering `apperror` (not the phantom `error`). Thanks @cabluvsmkm2009-source. (#6527) + - **The admin model-reload endpoint (`POST /api/admin/reload`) no longer crashes.** Reloading `api.models` rebound three names onto the routes module, but one of them — `_routes.compact = _models.compact` — referenced `compact` as if it were a module-level attribute when it is actually a method of the `Session` class, so the handler raised `AttributeError: module 'api.models' has no attribute 'compact'` and the reload failed with a 500. The invalid rebinding is removed (nothing reads `api.routes.compact`); the two genuinely module-level rebindings (`get_session`, `Session`) are unchanged, and the endpoint now returns `{"status": "ok"}`. Thanks @mspgle-dotcom. (#6477) - **Abandoned embedded terminals are now reaped instead of leaking a running shell forever (#4633 backend half).** A WebUI terminal whose viewer dropped its output stream without an explicit close (tab close, crash, network drop) previously kept its shell process and pty fd alive indefinitely — there is no `PDEATHSIG`, so nothing tore it down until the terminal cap happened to evict it. A background reaper now closes any terminal that has been unwatched (zero attached output streams) for longer than a 15-minute grace, and sweeps dead-process terminals immediately; the grace deliberately spans a tab refresh or brief network drop so a genuine reconnect keeps the session. The output-stream attach is now atomic against the reaper (a single registry-locked lookup+subscribe), so a viewer can never end up subscribed to a terminal the reaper is tearing down, and a race resolves to exactly one winner — attach keeps the terminal, or the reaper wins and the stream cleanly reports "terminal not running." A related leak is closed too: if writing the SSE response headers fails after attaching (client already gone), the subscription is now always unsubscribed instead of pinning the terminal un-reapable. Thanks @ai-ag2026. (#5844, #4633) 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_regressions.py b/tests/test_regressions.py index 8d38e449..df8790af 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -195,10 +195,20 @@ def test_server_py_sse_loop_breaks_on_cancel(cleanup_test_sessions): src = (REPO_ROOT / "server.py").read_text() routes_src = (REPO_ROOT / "api" / "routes.py").read_text() if (REPO_ROOT / "api" / "routes.py").exists() else "" combined = src + routes_src - m = re.search(r"if event in \([^)]+\):\s*break", combined) - assert m, "SSE break condition not found in server.py or api/routes.py" - assert "cancel" in m.group(), \ - f"'cancel' missing from SSE break condition: {m.group()}" + # #6527: the break condition may be an inline tuple ("stream_end", ...) OR + # the named SSE_RELAY_CLOSE_EVENTS frozenset. Accept either shape and pin + # the BEHAVIOR (cancel closes the relay) against the resolved close set, + # not the source syntax. + m = re.search( + r"if event in (?:SSE_RELAY_CLOSE_EVENTS|\([^)]*cancel[^)]*\)):\s*break", + combined, + ) + assert m, "SSE break/close condition not found in server.py or api/routes.py" + from api.run_journal import SSE_RELAY_CLOSE_EVENTS + assert "cancel" in SSE_RELAY_CLOSE_EVENTS, \ + f"'cancel' missing from SSE relay close set: {SSE_RELAY_CLOSE_EVENTS}" + assert "apperror" in SSE_RELAY_CLOSE_EVENTS, \ + f"'apperror' missing from SSE relay close set: {SSE_RELAY_CLOSE_EVENTS}" # ── R6: Test cron isolation (Sprint 10) ────────────────────────────────────── 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