Merge pull request #6537 from nesquena/gate-6527-sse-relay-close

Release exp-v0.52.152: SSE relay apperror-hang fix (#6527)

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)

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_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) ──────────────────────────────────────

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