fix(sse): close relays on apperror terminal events (gate-rebase of #6527)

Co-authored-by: Hermes Agent <hermes@nousresearch.com>

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)

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"

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:

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,

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
 

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