"""Drain thread for terminal(notify_on_complete=true) agent wakeup. The hermes-agent ``tools.process_registry.ProcessRegistry`` exposes a thread-safe``completion_queue`` (a ``queue.Queue``) that any background process pushes ontowhen it exits or matches a ``watch_patterns`` rule. In the CLI and in thegateway adapter this queue is drained by the host's main loop; in WebUI thequeue was never read, so the agent never woke up from a ``notify_on_complete``finish. This module restores that behavior. The drain thread: 1. blocks on ``completion_queue.get()`` in a worker thread, 2. looks up the WebUI session_id from ``PROCESS_SESSION_INDEX`` (keyed on the per-process ``session_key`` env var captured at spawn time), 3. formats a synthetic ``[IMPORTANT: ...]`` wakeup prompt, identical in intent to ``cli._format_process_notification`` and ``gateway.run._format_gateway_process_notification`` so the agent sees the same payload regardless of host, 4. emits a canonical ``bg_task_complete`` SSE event (plus a temporary ``process_complete`` alias for the migration window) on the active stream(s) for that session (DEMOTED to pure live-view — an open tab streams the turn live), records a server-side marker in ``PENDING_BG_TASK_COMPLETIONS``, and — Option Z PIVOT — starts the agent wakeup turn **directly server-side** when the session is idle (``_start_server_side_wakeup_turn`` → ``routes.start_session_turn``). This needs NO browser round-trip, so the closed-tab case works exactly like CLI / Telegram / gateway self-wake. When a turn is already active the wakeup is NOT started here; the ``PENDING_BG_TASK_COMPLETIONS`` marker is left for PR #2279's next-turn drain (``api/streaming._drain_webui_process_notifications``). The marker is *not* required for delivery — it's a telemetry-style flag theturn handler can read to know "this stream is a process_complete wakeup, not ahuman-typed prompt". It also lets the PR #2279 next-turn drain deliver thewakeup when a turn was active at completion time; the marker drains harmlesslyon the next turn for the session. Watch-pattern events share the same queue but produce a different SSE payload;this module routes them to the same listener so the frontend's single``process_complete`` handler can re-POST either flavor verbatim.""" from __future__ import annotations import loggingimport queueimport threadingimport timeimport uuidfrom typing import Any, Optional logger = logging.getLogger(__name__) _DRAIN_THREAD: Optional[threading.Thread] = None_DRAIN_STOP = threading.Event() _REAPER_THREAD: Optional[threading.Thread] = None_REAPER_STOP = threading.Event()_REAPER_INTERVAL_SECS = 60.0 # Serializes the check-then-start of the module's daemon threads# (``start_drain_thread`` / ``start_session_channel_reaper``). Without it two# concurrent callers can both observe ``is_alive() == False`` and each spawn a# thread; the loser's thread is never referenced by the module global and runs# forever, un-joinable. A dedicated lock (not the purpose-bound# ``SESSION_CHANNELS_LOCK`` / ``_EMIT_COALESCE_LOCK``) keeps this narrow._THREAD_LIFECYCLE_LOCK = threading.Lock() # T3: per-session coalesce gate for the public bg_task_complete SSE emit.# The server-side wakeup path remains immediate; only the browser-observation# frame is throttled so a burst of background task completions does not flood an# open tab. The first emit for a session fires immediately, then any further# emits inside the 1s window are payload-replaced and flushed after 1s of quiet._EMIT_COALESCE_WINDOW_SECS = 1.0_EMIT_COALESCE_LOCK = threading.Lock()_LAST_EMIT_TS: dict[str, float] = {}_PENDING_EMIT_PAYLOADS: dict[str, dict] = {}_PENDING_EMIT_TIMERS: dict[str, threading.Timer] = {} # ── Persistent per-session SSE channel (Option X) ──────────────────────────# SESSION_CHANNELS maps WebUI session_id -> SessionChannel. Each channel owns# zero or more queue.Queue subscribers (one per active EventSource tab) and# is collected by ``_reaper_loop`` after the last subscriber drops + a grace# period, or after the session has been idle for SESSION_CHANNEL_IDLE_TTL_SECS.## Why a sibling registry to STREAMS:# STREAMS is keyed on stream_id (one per agent turn) and is torn down by# /api/chat/stream's `finally` when the turn ends. process_complete events# from background processes that exit BETWEEN turns therefore have no live# STREAMS channel to ride. SESSION_CHANNELS is keyed on session_id, lives# across turns, and gives the frontend a stable subscription that survives# stream_end / cancel / reconnect.SESSION_CHANNELS: dict[str, "SessionChannel"] = {}SESSION_CHANNELS_LOCK = threading.Lock() class SessionChannel: """A long-lived multi-subscriber SSE channel for one WebUI session. Subscribers are ``queue.Queue`` instances owned by the SSE route handler — one per active EventSource (tab). ``emit`` broadcasts to every live subscriber; subscribers whose buffer is full silently drop the event (the tab will reconnect on disconnect and the SSE-level disconnect detection will tear it down). Lifecycle: - Created on demand by ``get_or_create_session_channel`` when the first tab subscribes. - ``subscribe`` / ``unsubscribe`` are refcount-style: zero subscribers does NOT immediately collect the channel; the reaper waits a 60s grace so a quick navigation away/back doesn't churn the registry. - The reaper collects the channel when subscribers stay empty past the grace period, OR when subscribers are empty AND ``created_at`` is older than SESSION_CHANNEL_IDLE_TTL_SECS (zombie cap — applies only when nobody is subscribed; a live subscriber keeps the channel even past the idle TTL). """ def __init__(self, session_id: str): self.session_id = session_id self._lock = threading.Lock() self._subscribers: list[queue.Queue] = [] now = time.time() self.created_at = now self.last_event_at = now self.last_subscriber_drop_at: float | None = None def subscribe(self, maxsize: int = 16) -> queue.Queue: q: queue.Queue = queue.Queue(maxsize=maxsize) with self._lock: self._subscribers.append(q) # Cancel any pending subscribers-empty grace timer. self.last_subscriber_drop_at = None return q def unsubscribe(self, q: queue.Queue) -> None: with self._lock: try: self._subscribers.remove(q) except ValueError: pass if not self._subscribers: self.last_subscriber_drop_at = time.time() def subscriber_count(self) -> int: with self._lock: return len(self._subscribers) def emit(self, event: str, data: Any) -> int: """Broadcast (event, data) to all live subscribers. Returns delivered count.""" delivered = 0 with self._lock: subs = list(self._subscribers) self.last_event_at = time.time() for q in subs: try: q.put_nowait((event, data)) delivered += 1 except queue.Full: # Slow tab: drop this event for that tab. SSE-level disconnect # detection will eventually tear the connection down and the # browser will reconnect, replaying the live stream from # whatever fires next. process_complete is intrinsically # idempotent (frontend dedupes by ``(session_id, event_id)`` # using a small ring-buffer in static/messages.js — see the # bg_task_complete consumer-side dedupe introduced in PR #2971). logger.debug("SessionChannel emit: subscriber buffer full, dropping") except Exception: logger.debug("SessionChannel emit failed", exc_info=True) return delivered def reaper_should_collect(self, now: float) -> bool: """True when the reaper should remove this channel. Two collection conditions (per Option X spec): 1. Subscribers empty AND last_subscriber_drop_at is older than SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS (normal teardown). 2. created_at older than SESSION_CHANNEL_IDLE_TTL_SECS AND subscribers empty (zombie cap — survived too long). """ from api import config as _cfg with self._lock: sub_count = len(self._subscribers) drop_at = self.last_subscriber_drop_at created_at = self.created_at if sub_count > 0: # Live subscriber — never collect, even past idle TTL (a tab is # genuinely listening). The browser will close on its own. return False # No subscribers — check grace period. grace = float(getattr(_cfg, "SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS", 60)) if drop_at is not None and (now - drop_at) >= grace: return True # Hard cap on lifetime (even if subscribers oscillated): if created # long ago AND nobody's subscribed right now, sweep. ttl = float(getattr(_cfg, "SESSION_CHANNEL_IDLE_TTL_SECS", 14400)) if (now - created_at) >= ttl: return True return False def get_or_create_session_channel(session_id: str) -> SessionChannel: """Return the channel for ``session_id``, creating it on first access.""" with SESSION_CHANNELS_LOCK: ch = SESSION_CHANNELS.get(session_id) if ch is None: ch = SessionChannel(session_id) SESSION_CHANNELS[session_id] = ch return ch def get_session_channel(session_id: str) -> Optional[SessionChannel]: """Return an existing channel or None — does NOT auto-create.""" with SESSION_CHANNELS_LOCK: return SESSION_CHANNELS.get(session_id) def subscribe_to_session_channel( session_id: str, maxsize: int = 16) -> tuple["SessionChannel", "queue.Queue"]: """Atomically get-or-create the channel for ``session_id`` AND register a subscriber on it, both under ``SESSION_CHANNELS_LOCK``. Closes a TOCTOU race flagged on PR #2971: callers that did ``ch = get_or_create_session_channel(sid); q = ch.subscribe()`` released ``SESSION_CHANNELS_LOCK`` between the two steps. The reaper acquires that same lock and evaluates+collects channels entirely within it (``_reaper_loop``), so a previously-idle channel with 0 subscribers past the grace/TTL window could be collected in the gap — leaving the SSE handler subscribed to a channel no longer in ``SESSION_CHANNELS``. Future ``bg_task_complete`` emits resolve the session via ``get_session_channel`` and reach a different (or absent) channel, so the orphaned subscriber's queue never fills: keepalives keep flowing, ``onerror`` never fires, no auto-reconnect, and only a manual refresh recovers. Holding ``SESSION_CHANNELS_LOCK`` across both the get-or-create and the ``subscribe`` makes the two steps indivisible w.r.t. the reaper: the reaper cannot run its critical section concurrently, and the next time it does the channel already has ``sub_count >= 1`` so ``reaper_should_collect`` refuses to collect it. Lock order is ``SESSION_CHANNELS_LOCK`` → ``SessionChannel._lock`` (taken inside ``subscribe``), identical to the order the reaper uses (``SESSION_CHANNELS_LOCK`` → ``reaper_should_collect`` → ``_lock``), so no new lock-ordering hazard is introduced. ``emit`` only takes ``_lock``. Returns ``(channel, queue)``. The caller still owns the subscriber slot and MUST ``channel.unsubscribe(queue)`` on every exit path. """ with SESSION_CHANNELS_LOCK: ch = SESSION_CHANNELS.get(session_id) if ch is None: ch = SessionChannel(session_id) SESSION_CHANNELS[session_id] = ch q = ch.subscribe(maxsize=maxsize) return ch, q def active_stream_id_for_session(session_id: str) -> Optional[str]: """Return the stream_id of the live run for *session_id*, or None. Used by the per-session SSE handler's on-subscribe recovery: the ``server_turn_started`` fan-out in ``routes.start_session_turn`` is a fire-and-forget broadcast with NO replay buffer (SessionChannel.emit drops to whoever is subscribed *at that instant*). A tab whose ``/api/session/stream`` EventSource is momentarily absent at the emit instant — a transient SSE drop, a reverse-proxy idle-timeout, or browser connection-pool starvation — misses the frame permanently and the server-initiated wakeup turn never renders live (the user must hard- refresh). The server-side wakeup itself still ran and persisted; only the live-view was lost. The handler replays a synthetic ``server_turn_started`` to a freshly-subscribed tab using this lookup so the open tab self-heals. Keys on ACTIVE_RUNS (worker-lifecycle registry) — the same source ``_session_has_active_turn`` / ``_emit_to_session_streams`` already trust to map a stream back to its owning session. Returns the first matching stream_id (a session has at most one live run; cancel/reconnect can briefly hold two — either is a valid attach target, the frontend dedupes by stream_id). """ from api import config as _cfg try: with _cfg.ACTIVE_RUNS_LOCK: for _stream_id, meta in (_cfg.ACTIVE_RUNS or {}).items(): if isinstance(meta, dict) and meta.get("session_id") == session_id: return str(_stream_id) except Exception: logger.debug( "active_stream_id_for_session lookup failed for %s", session_id, exc_info=True, ) return None def persisted_message_count_for_session(session_id: str) -> Optional[int]: """Cheap, metadata-only persisted ``message_count`` for *session_id*, or None. Companion to ``active_stream_id_for_session`` for the per-session SSE on-subscribe self-heal. ``active_stream_id_for_session`` recovers a turn that is live RIGHT NOW (replay ``server_turn_started``). But a SERVER-initiated turn (self-wake / cron / restart hook) can start AND finish entirely inside an SSE gap: the fire-and-forget ``server_turn_started`` reached no subscriber AND the run already cleared from ``ACTIVE_RUNS`` by the time the tab reconnects, so the replay finds nothing (returns None) and the tab's transcript stays stale until a hard refresh — the reported visible-tab defect. To detect that case the handler compares the freshly-(re)subscribed tab's last-known count against this persisted count; a server that is AHEAD means a turn landed during the gap. Reads via ``metadata_only=True`` so it never parses the full transcript (this runs on every per-session SSE (re)connect). The persisted count is written by ``Session.save`` as ``meta['message_count'] = len(messages)`` — the SAME basis the frontend's ``S.session.message_count`` is built from — so the comparison is apples-to-apples. Returns None when the count is unknown (legacy sidecars without a persisted count); the caller treats None as "cannot tell, do nothing", never as a trigger. """ try: from api.models import get_session s = get_session(session_id, metadata_only=True) count = getattr(s, "_metadata_message_count", None) if count is None: msgs = getattr(s, "messages", None) count = len(msgs) if isinstance(msgs, list) and msgs else None return int(count) if count is not None else None except Exception: logger.debug( "persisted_message_count_for_session lookup failed for %s", session_id, exc_info=True, ) return None def should_emit_session_updated( subscriber_known_count: Optional[int], persisted_count: Optional[int],) -> bool: """Gate for the per-session SSE "finished during the gap" self-heal emit. Single source of truth shared by the SSE handler and its tests so the two cannot drift (the handler MUST call this, not inline the comparison). Emit a ``session-updated`` frame ONLY when: * the (re)subscribing tab reported a last-known count (``?known_count``), AND * the persisted server-side count is known, AND * the server is STRICTLY ahead (a turn landed during the gap). A missing known count (tab didn't report), an unknown persisted count (legacy sidecar), or an equal/behind count all return False — never a spurious reload. """ if subscriber_known_count is None: return False if persisted_count is None: return False return persisted_count > subscriber_known_count def _reaper_loop() -> None: logger.info("SessionChannel reaper thread started") while not _REAPER_STOP.is_set(): try: now = time.time() collected: list[str] = [] with SESSION_CHANNELS_LOCK: for sid, ch in list(SESSION_CHANNELS.items()): if ch.reaper_should_collect(now): SESSION_CHANNELS.pop(sid, None) collected.append(sid) if collected: # Prune the per-session coalesce timestamp map for any collected # session so _LAST_EMIT_TS does not grow one permanent entry per # session that ever fired a bg task (greptile flag). The pending # payload/timer maps self-clean when their timers fire, but # _LAST_EMIT_TS is only ever written, never deleted — sweep it # here alongside the channel it belongs to. A session that fires # again after collection simply re-seeds its entry (first emit in # the new window fires immediately, which is correct). with _EMIT_COALESCE_LOCK: for sid in collected: _LAST_EMIT_TS.pop(sid, None) logger.debug("SessionChannel reaper collected: %s", collected) # Sweep the per-session completion-dedup map by DELIVERY lifecycle, # not channel collection. ``BG_TASK_COMPLETE_EVENTS_SEEN`` gains a # ``session_id -> set[process_id]`` entry the first time a bg task # completes for a session — in ``_process_one``, whether or not any # tab/SSE channel ever existed — and is otherwise never deleted, so it # grows unbounded. Coupling the prune to channel collection (an # earlier version of this fix) missed the dominant case: a headless # completion (task fires, tab closed or never opened) has no channel # to collect. Instead, once a completion has been drained (its # ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``), the # short ``_move_to_finished`` dedup window is closed and the entry is # pure leak — so sweep every delivered (not-pending) session here, # every tick. The registry's own per-``process_id`` # ``_completion_consumed`` gate remains the primary idempotency # backstop, so sweeping a delivered session's set can never resurrect # an already-delivered completion (even in the tiny window between # this module's ``SEEN.add`` and ``PENDING.add`` in ``_process_one``). from api import config as _cfg with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK: for sid in [ s for s in _cfg.BG_TASK_COMPLETE_EVENTS_SEEN if s not in _cfg.PENDING_BG_TASK_COMPLETIONS ]: _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) except Exception: logger.warning("SessionChannel reaper iteration failed", exc_info=True) # Wait but wake up promptly on stop. if _REAPER_STOP.wait(_REAPER_INTERVAL_SECS): break def start_session_channel_reaper() -> bool: """Start the SessionChannel reaper thread. Idempotent; returns True on first start.""" global _REAPER_THREAD with _THREAD_LIFECYCLE_LOCK: if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive(): return False _REAPER_STOP.clear() _REAPER_THREAD = threading.Thread( target=_reaper_loop, name="hermes-webui-session-channel-reaper", daemon=True, ) _REAPER_THREAD.start() return True def stop_session_channel_reaper(timeout: float = 2.0) -> None: _REAPER_STOP.set() th = _REAPER_THREAD if th is not None and th.is_alive(): th.join(timeout=timeout) def _truncate(text: str, limit: int) -> str: if text is None: return "" s = str(text) if len(s) <= limit: return s return s[:limit] + "\n…(truncated)" def format_wakeup_prompt(evt: object) -> str | None: """Build the synthetic [IMPORTANT: …] message the agent will see. Mirrors ``cli._format_process_notification`` so wakeup payloads look the same in CLI and WebUI sessions. """ if not isinstance(evt, dict) or not evt: return None evt_type = evt.get("type", "completion") sid = str(evt.get("session_id") or "").strip() cmd = str(evt.get("command") or "").strip() # The current server-side wakeup drain drops global watch-overflow events # before this formatter because they intentionally carry no session_key. # Keep this branch defensive so any future routable overflow summary is not # mis-rendered as a fake process completion. if evt_type in {"watch_overflow_tripped", "watch_overflow_released"}: msg = str(evt.get("message") or "").strip() return f"[IMPORTANT: {msg}]" if msg else None if evt_type == "watch_disabled": msg = str(evt.get("message") or "").strip() return f"[IMPORTANT: {msg}]" if msg else None if evt_type == "watch_match": pat = evt.get("pattern", "?") out = _truncate(evt.get("output", ""), 4000) sup = evt.get("suppressed", 0) body = ( f"[IMPORTANT: Background process {sid} matched watch pattern \"{pat}\".\n" f"Command: {cmd}\n" f"Matched output:\n{out}" ) if sup: body += f"\n({sup} earlier matches were suppressed by rate limit)" return body + "]" if evt_type == "async_delegation": # A background ``delegate_task`` completion. The agent-side formatter # renders these; delegate to it so the subagent result re-enters the # parent conversation instead of being silently dropped (#4912). try: from tools.process_registry import ( format_process_notification as _agent_fmt, ) result = _agent_fmt(evt) if result: return result except Exception: logger.debug( "agent-side format_process_notification fallback failed for "Lines 1–500 of 1424. Load more
"""Drain thread for terminal(notify_on_complete=true) agent wakeup. The hermes-agent ``tools.process_registry.ProcessRegistry`` exposes a thread-safe ``completion_queue`` (a ``queue.Queue``) that any background process pushes onto when it exits or matches a ``watch_patterns`` rule. In the CLI and in the gateway adapter this queue is drained by the host's main loop; in WebUI the queue was never read, so the agent never woke up from a ``notify_on_complete`` finish. This module restores that behavior. The drain thread: 1. blocks on ``completion_queue.get()`` in a worker thread, 2. looks up the WebUI session_id from ``PROCESS_SESSION_INDEX`` (keyed on the per-process ``session_key`` env var captured at spawn time), 3. formats a synthetic ``[IMPORTANT: ...]`` wakeup prompt, identical in intent to ``cli._format_process_notification`` and ``gateway.run._format_gateway_process_notification`` so the agent sees the same payload regardless of host, 4. emits a canonical ``bg_task_complete`` SSE event (plus a temporary ``process_complete`` alias for the migration window) on the active stream(s) for that session (DEMOTED to pure live-view — an open tab streams the turn live), records a server-side marker in ``PENDING_BG_TASK_COMPLETIONS``, and — Option Z PIVOT — starts the agent wakeup turn **directly server-side** when the session is idle (``_start_server_side_wakeup_turn`` → ``routes.start_session_turn``). This needs NO browser round-trip, so the closed-tab case works exactly like CLI / Telegram / gateway self-wake. When a turn is already active the wakeup is NOT started here; the ``PENDING_BG_TASK_COMPLETIONS`` marker is left for PR #2279's next-turn drain (``api/streaming._drain_webui_process_notifications``). The marker is *not* required for delivery — it's a telemetry-style flag the turn handler can read to know "this stream is a process_complete wakeup, not a human-typed prompt". It also lets the PR #2279 next-turn drain deliver the wakeup when a turn was active at completion time; the marker drains harmlessly on the next turn for the session. Watch-pattern events share the same queue but produce a different SSE payload; this module routes them to the same listener so the frontend's single ``process_complete`` handler can re-POST either flavor verbatim. """ from __future__ import annotations import logging import queue import threading import time import uuid from typing import Any, Optional logger = logging.getLogger(__name__) _DRAIN_THREAD: Optional[threading.Thread] = None _DRAIN_STOP = threading.Event() _REAPER_THREAD: Optional[threading.Thread] = None _REAPER_STOP = threading.Event() _REAPER_INTERVAL_SECS = 60.0 # Serializes the check-then-start of the module's daemon threads # (``start_drain_thread`` / ``start_session_channel_reaper``). Without it two # concurrent callers can both observe ``is_alive() == False`` and each spawn a # thread; the loser's thread is never referenced by the module global and runs # forever, un-joinable. A dedicated lock (not the purpose-bound # ``SESSION_CHANNELS_LOCK`` / ``_EMIT_COALESCE_LOCK``) keeps this narrow. _THREAD_LIFECYCLE_LOCK = threading.Lock() # T3: per-session coalesce gate for the public bg_task_complete SSE emit. # The server-side wakeup path remains immediate; only the browser-observation # frame is throttled so a burst of background task completions does not flood an # open tab. The first emit for a session fires immediately, then any further # emits inside the 1s window are payload-replaced and flushed after 1s of quiet. _EMIT_COALESCE_WINDOW_SECS = 1.0 _EMIT_COALESCE_LOCK = threading.Lock() _LAST_EMIT_TS: dict[str, float] = {} _PENDING_EMIT_PAYLOADS: dict[str, dict] = {} _PENDING_EMIT_TIMERS: dict[str, threading.Timer] = {} # ── Persistent per-session SSE channel (Option X) ────────────────────────── # SESSION_CHANNELS maps WebUI session_id -> SessionChannel. Each channel owns # zero or more queue.Queue subscribers (one per active EventSource tab) and # is collected by ``_reaper_loop`` after the last subscriber drops + a grace # period, or after the session has been idle for SESSION_CHANNEL_IDLE_TTL_SECS. # # Why a sibling registry to STREAMS: # STREAMS is keyed on stream_id (one per agent turn) and is torn down by # /api/chat/stream's `finally` when the turn ends. process_complete events # from background processes that exit BETWEEN turns therefore have no live # STREAMS channel to ride. SESSION_CHANNELS is keyed on session_id, lives # across turns, and gives the frontend a stable subscription that survives # stream_end / cancel / reconnect. SESSION_CHANNELS: dict[str, "SessionChannel"] = {} SESSION_CHANNELS_LOCK = threading.Lock() class SessionChannel: """A long-lived multi-subscriber SSE channel for one WebUI session. Subscribers are ``queue.Queue`` instances owned by the SSE route handler — one per active EventSource (tab). ``emit`` broadcasts to every live subscriber; subscribers whose buffer is full silently drop the event (the tab will reconnect on disconnect and the SSE-level disconnect detection will tear it down). Lifecycle: - Created on demand by ``get_or_create_session_channel`` when the first tab subscribes. - ``subscribe`` / ``unsubscribe`` are refcount-style: zero subscribers does NOT immediately collect the channel; the reaper waits a 60s grace so a quick navigation away/back doesn't churn the registry. - The reaper collects the channel when subscribers stay empty past the grace period, OR when subscribers are empty AND ``created_at`` is older than SESSION_CHANNEL_IDLE_TTL_SECS (zombie cap — applies only when nobody is subscribed; a live subscriber keeps the channel even past the idle TTL). """ def __init__(self, session_id: str): self.session_id = session_id self._lock = threading.Lock() self._subscribers: list[queue.Queue] = [] now = time.time() self.created_at = now self.last_event_at = now self.last_subscriber_drop_at: float | None = None def subscribe(self, maxsize: int = 16) -> queue.Queue: q: queue.Queue = queue.Queue(maxsize=maxsize) with self._lock: self._subscribers.append(q) # Cancel any pending subscribers-empty grace timer. self.last_subscriber_drop_at = None return q def unsubscribe(self, q: queue.Queue) -> None: with self._lock: try: self._subscribers.remove(q) except ValueError: pass if not self._subscribers: self.last_subscriber_drop_at = time.time() def subscriber_count(self) -> int: with self._lock: return len(self._subscribers) def emit(self, event: str, data: Any) -> int: """Broadcast (event, data) to all live subscribers. Returns delivered count.""" delivered = 0 with self._lock: subs = list(self._subscribers) self.last_event_at = time.time() for q in subs: try: q.put_nowait((event, data)) delivered += 1 except queue.Full: # Slow tab: drop this event for that tab. SSE-level disconnect # detection will eventually tear the connection down and the # browser will reconnect, replaying the live stream from # whatever fires next. process_complete is intrinsically # idempotent (frontend dedupes by ``(session_id, event_id)`` # using a small ring-buffer in static/messages.js — see the # bg_task_complete consumer-side dedupe introduced in PR #2971). logger.debug("SessionChannel emit: subscriber buffer full, dropping") except Exception: logger.debug("SessionChannel emit failed", exc_info=True) return delivered def reaper_should_collect(self, now: float) -> bool: """True when the reaper should remove this channel. Two collection conditions (per Option X spec): 1. Subscribers empty AND last_subscriber_drop_at is older than SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS (normal teardown). 2. created_at older than SESSION_CHANNEL_IDLE_TTL_SECS AND subscribers empty (zombie cap — survived too long). """ from api import config as _cfg with self._lock: sub_count = len(self._subscribers) drop_at = self.last_subscriber_drop_at created_at = self.created_at if sub_count > 0: # Live subscriber — never collect, even past idle TTL (a tab is # genuinely listening). The browser will close on its own. return False # No subscribers — check grace period. grace = float(getattr(_cfg, "SESSION_CHANNEL_SUBSCRIBER_GRACE_SECS", 60)) if drop_at is not None and (now - drop_at) >= grace: return True # Hard cap on lifetime (even if subscribers oscillated): if created # long ago AND nobody's subscribed right now, sweep. ttl = float(getattr(_cfg, "SESSION_CHANNEL_IDLE_TTL_SECS", 14400)) if (now - created_at) >= ttl: return True return False def get_or_create_session_channel(session_id: str) -> SessionChannel: """Return the channel for ``session_id``, creating it on first access.""" with SESSION_CHANNELS_LOCK: ch = SESSION_CHANNELS.get(session_id) if ch is None: ch = SessionChannel(session_id) SESSION_CHANNELS[session_id] = ch return ch def get_session_channel(session_id: str) -> Optional[SessionChannel]: """Return an existing channel or None — does NOT auto-create.""" with SESSION_CHANNELS_LOCK: return SESSION_CHANNELS.get(session_id) def subscribe_to_session_channel( session_id: str, maxsize: int = 16 ) -> tuple["SessionChannel", "queue.Queue"]: """Atomically get-or-create the channel for ``session_id`` AND register a subscriber on it, both under ``SESSION_CHANNELS_LOCK``. Closes a TOCTOU race flagged on PR #2971: callers that did ``ch = get_or_create_session_channel(sid); q = ch.subscribe()`` released ``SESSION_CHANNELS_LOCK`` between the two steps. The reaper acquires that same lock and evaluates+collects channels entirely within it (``_reaper_loop``), so a previously-idle channel with 0 subscribers past the grace/TTL window could be collected in the gap — leaving the SSE handler subscribed to a channel no longer in ``SESSION_CHANNELS``. Future ``bg_task_complete`` emits resolve the session via ``get_session_channel`` and reach a different (or absent) channel, so the orphaned subscriber's queue never fills: keepalives keep flowing, ``onerror`` never fires, no auto-reconnect, and only a manual refresh recovers. Holding ``SESSION_CHANNELS_LOCK`` across both the get-or-create and the ``subscribe`` makes the two steps indivisible w.r.t. the reaper: the reaper cannot run its critical section concurrently, and the next time it does the channel already has ``sub_count >= 1`` so ``reaper_should_collect`` refuses to collect it. Lock order is ``SESSION_CHANNELS_LOCK`` → ``SessionChannel._lock`` (taken inside ``subscribe``), identical to the order the reaper uses (``SESSION_CHANNELS_LOCK`` → ``reaper_should_collect`` → ``_lock``), so no new lock-ordering hazard is introduced. ``emit`` only takes ``_lock``. Returns ``(channel, queue)``. The caller still owns the subscriber slot and MUST ``channel.unsubscribe(queue)`` on every exit path. """ with SESSION_CHANNELS_LOCK: ch = SESSION_CHANNELS.get(session_id) if ch is None: ch = SessionChannel(session_id) SESSION_CHANNELS[session_id] = ch q = ch.subscribe(maxsize=maxsize) return ch, q def active_stream_id_for_session(session_id: str) -> Optional[str]: """Return the stream_id of the live run for *session_id*, or None. Used by the per-session SSE handler's on-subscribe recovery: the ``server_turn_started`` fan-out in ``routes.start_session_turn`` is a fire-and-forget broadcast with NO replay buffer (SessionChannel.emit drops to whoever is subscribed *at that instant*). A tab whose ``/api/session/stream`` EventSource is momentarily absent at the emit instant — a transient SSE drop, a reverse-proxy idle-timeout, or browser connection-pool starvation — misses the frame permanently and the server-initiated wakeup turn never renders live (the user must hard- refresh). The server-side wakeup itself still ran and persisted; only the live-view was lost. The handler replays a synthetic ``server_turn_started`` to a freshly-subscribed tab using this lookup so the open tab self-heals. Keys on ACTIVE_RUNS (worker-lifecycle registry) — the same source ``_session_has_active_turn`` / ``_emit_to_session_streams`` already trust to map a stream back to its owning session. Returns the first matching stream_id (a session has at most one live run; cancel/reconnect can briefly hold two — either is a valid attach target, the frontend dedupes by stream_id). """ from api import config as _cfg try: with _cfg.ACTIVE_RUNS_LOCK: for _stream_id, meta in (_cfg.ACTIVE_RUNS or {}).items(): if isinstance(meta, dict) and meta.get("session_id") == session_id: return str(_stream_id) except Exception: logger.debug( "active_stream_id_for_session lookup failed for %s", session_id, exc_info=True, ) return None def persisted_message_count_for_session(session_id: str) -> Optional[int]: """Cheap, metadata-only persisted ``message_count`` for *session_id*, or None. Companion to ``active_stream_id_for_session`` for the per-session SSE on-subscribe self-heal. ``active_stream_id_for_session`` recovers a turn that is live RIGHT NOW (replay ``server_turn_started``). But a SERVER-initiated turn (self-wake / cron / restart hook) can start AND finish entirely inside an SSE gap: the fire-and-forget ``server_turn_started`` reached no subscriber AND the run already cleared from ``ACTIVE_RUNS`` by the time the tab reconnects, so the replay finds nothing (returns None) and the tab's transcript stays stale until a hard refresh — the reported visible-tab defect. To detect that case the handler compares the freshly-(re)subscribed tab's last-known count against this persisted count; a server that is AHEAD means a turn landed during the gap. Reads via ``metadata_only=True`` so it never parses the full transcript (this runs on every per-session SSE (re)connect). The persisted count is written by ``Session.save`` as ``meta['message_count'] = len(messages)`` — the SAME basis the frontend's ``S.session.message_count`` is built from — so the comparison is apples-to-apples. Returns None when the count is unknown (legacy sidecars without a persisted count); the caller treats None as "cannot tell, do nothing", never as a trigger. """ try: from api.models import get_session s = get_session(session_id, metadata_only=True) count = getattr(s, "_metadata_message_count", None) if count is None: msgs = getattr(s, "messages", None) count = len(msgs) if isinstance(msgs, list) and msgs else None return int(count) if count is not None else None except Exception: logger.debug( "persisted_message_count_for_session lookup failed for %s", session_id, exc_info=True, ) return None def should_emit_session_updated( subscriber_known_count: Optional[int], persisted_count: Optional[int], ) -> bool: """Gate for the per-session SSE "finished during the gap" self-heal emit. Single source of truth shared by the SSE handler and its tests so the two cannot drift (the handler MUST call this, not inline the comparison). Emit a ``session-updated`` frame ONLY when: * the (re)subscribing tab reported a last-known count (``?known_count``), AND * the persisted server-side count is known, AND * the server is STRICTLY ahead (a turn landed during the gap). A missing known count (tab didn't report), an unknown persisted count (legacy sidecar), or an equal/behind count all return False — never a spurious reload. """ if subscriber_known_count is None: return False if persisted_count is None: return False return persisted_count > subscriber_known_count def _reaper_loop() -> None: logger.info("SessionChannel reaper thread started") while not _REAPER_STOP.is_set(): try: now = time.time() collected: list[str] = [] with SESSION_CHANNELS_LOCK: for sid, ch in list(SESSION_CHANNELS.items()): if ch.reaper_should_collect(now): SESSION_CHANNELS.pop(sid, None) collected.append(sid) if collected: # Prune the per-session coalesce timestamp map for any collected # session so _LAST_EMIT_TS does not grow one permanent entry per # session that ever fired a bg task (greptile flag). The pending # payload/timer maps self-clean when their timers fire, but # _LAST_EMIT_TS is only ever written, never deleted — sweep it # here alongside the channel it belongs to. A session that fires # again after collection simply re-seeds its entry (first emit in # the new window fires immediately, which is correct). with _EMIT_COALESCE_LOCK: for sid in collected: _LAST_EMIT_TS.pop(sid, None) logger.debug("SessionChannel reaper collected: %s", collected) # Sweep the per-session completion-dedup map by DELIVERY lifecycle, # not channel collection. ``BG_TASK_COMPLETE_EVENTS_SEEN`` gains a # ``session_id -> set[process_id]`` entry the first time a bg task # completes for a session — in ``_process_one``, whether or not any # tab/SSE channel ever existed — and is otherwise never deleted, so it # grows unbounded. Coupling the prune to channel collection (an # earlier version of this fix) missed the dominant case: a headless # completion (task fires, tab closed or never opened) has no channel # to collect. Instead, once a completion has been drained (its # ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``), the # short ``_move_to_finished`` dedup window is closed and the entry is # pure leak — so sweep every delivered (not-pending) session here, # every tick. The registry's own per-``process_id`` # ``_completion_consumed`` gate remains the primary idempotency # backstop, so sweeping a delivered session's set can never resurrect # an already-delivered completion (even in the tiny window between # this module's ``SEEN.add`` and ``PENDING.add`` in ``_process_one``). from api import config as _cfg with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK: for sid in [ s for s in _cfg.BG_TASK_COMPLETE_EVENTS_SEEN if s not in _cfg.PENDING_BG_TASK_COMPLETIONS ]: _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None) except Exception: logger.warning("SessionChannel reaper iteration failed", exc_info=True) # Wait but wake up promptly on stop. if _REAPER_STOP.wait(_REAPER_INTERVAL_SECS): break def start_session_channel_reaper() -> bool: """Start the SessionChannel reaper thread. Idempotent; returns True on first start.""" global _REAPER_THREAD with _THREAD_LIFECYCLE_LOCK: if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive(): return False _REAPER_STOP.clear() _REAPER_THREAD = threading.Thread( target=_reaper_loop, name="hermes-webui-session-channel-reaper", daemon=True, ) _REAPER_THREAD.start() return True def stop_session_channel_reaper(timeout: float = 2.0) -> None: _REAPER_STOP.set() th = _REAPER_THREAD if th is not None and th.is_alive(): th.join(timeout=timeout) def _truncate(text: str, limit: int) -> str: if text is None: return "" s = str(text) if len(s) <= limit: return s return s[:limit] + "\n…(truncated)" def format_wakeup_prompt(evt: object) -> str | None: """Build the synthetic [IMPORTANT: …] message the agent will see. Mirrors ``cli._format_process_notification`` so wakeup payloads look the same in CLI and WebUI sessions. """ if not isinstance(evt, dict) or not evt: return None evt_type = evt.get("type", "completion") sid = str(evt.get("session_id") or "").strip() cmd = str(evt.get("command") or "").strip() # The current server-side wakeup drain drops global watch-overflow events # before this formatter because they intentionally carry no session_key. # Keep this branch defensive so any future routable overflow summary is not # mis-rendered as a fake process completion. if evt_type in {"watch_overflow_tripped", "watch_overflow_released"}: msg = str(evt.get("message") or "").strip() return f"[IMPORTANT: {msg}]" if msg else None if evt_type == "watch_disabled": msg = str(evt.get("message") or "").strip() return f"[IMPORTANT: {msg}]" if msg else None if evt_type == "watch_match": pat = evt.get("pattern", "?") out = _truncate(evt.get("output", ""), 4000) sup = evt.get("suppressed", 0) body = ( f"[IMPORTANT: Background process {sid} matched watch pattern \"{pat}\".\n" f"Command: {cmd}\n" f"Matched output:\n{out}" ) if sup: body += f"\n({sup} earlier matches were suppressed by rate limit)" return body + "]" if evt_type == "async_delegation": # A background ``delegate_task`` completion. The agent-side formatter # renders these; delegate to it so the subagent result re-enters the # parent conversation instead of being silently dropped (#4912). try: from tools.process_registry import ( format_process_notification as _agent_fmt, ) result = _agent_fmt(evt) if result: return result except Exception: logger.debug( "agent-side format_process_notification fallback failed for "