"""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 # 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 _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) 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 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 != "completion": return None if not (sid or cmd or "exit_code" in evt or evt.get("output")): return None # Default: completion event exit_code = evt.get("exit_code", "?") out = _truncate(evt.get("output", ""), 4000) return ( f"[IMPORTANT: Background process {sid} completed (exit_code={exit_code}).\n" f"Command: {cmd}\n" f"Output:\n{out}]" ) def _build_payload(evt: dict, session_id: str) -> dict: """Build the SSE data payload. Shape per maintainer decision on PR #2242 (R2 §Q1): ``{session_id, task_id, completed_at, summary?, event_id}`` Minimal by design — consumers re-fetch task detail by ``task_id`` if they need ``command`` / ``exit_code`` / ``stdout_preview`` etc. - ``task_id``: the background process id (registry uuid). Stable across the process's lifetime; was previously surfaced as ``process_id``. - ``completed_at``: float wall-clock seconds; was ``emitted_at``. - ``summary``: optional one-liner derived from the completion event when available (e.g. ``[IMPORTANT: …]`` synthetic body's first line); omitted otherwise to honour "keep the payload minimal". - ``event_id``: server-generated uuid hex — per-emit, so a re-emit for the same ``task_id`` (shouldn't happen today, but the spec is forward-looking) still produces a distinct id. The cross-A/B dedupe stays keyed on ``task_id`` (it prevents double-emit at the source, which is the right layer); the WebUI consumer-side ring buffer (PR (b)) keys on ``(session_id, event_id)`` to dedupe across reconnects. """ # ProcessRegistry completion events use the field name ``session_id`` for # the process id. Alias it locally before exposing it as payload ``task_id`` # to avoid confusing that wire-format name with the WebUI session id. process_id = str(evt.get("session_id") or "") payload: dict[str, Any] = { "session_id": str(session_id), "task_id": process_id, "completed_at": time.time(), "event_id": uuid.uuid4().hex, } # Best-effort optional summary: the first non-empty line of the synthetic # wakeup body, trimmed. Omitted entirely when nothing useful is available. try: wakeup_body = format_wakeup_prompt(evt) if wakeup_body: # Strip leading "[IMPORTANT: " marker noise — take the first # informative line, cap length. first_line = next( ( ln.strip().lstrip("[").rstrip("]").strip() for ln in wakeup_body.splitlines() if ln.strip() ), "", ) if first_line: payload["summary"] = _truncate(first_line, 200) except Exception: # Summary is optional; never let its derivation block the emit. logger.debug("summary derivation failed", exc_info=True) return payload def _emit_to_session_streams(session_id: str, event: str, data: dict) -> int: """Push (event, data) to every active SSE channel for *session_id*. Streams in WebUI are keyed by ``stream_id`` (not session_id) — a single session can have at most one active stream at a time, but cancel/reconnect flows can briefly hold two. We push to every channel whose tracked ``session_id`` matches; the channel implementation broadcasts to all live subscribers and buffers when offline. """ from api import config as _cfg emitted = 0 # Snapshot ACTIVE_RUNS under its lock before STREAMS_LOCK so owner_sid # lookups are consistent without nesting the two independent locks. if hasattr(_cfg, "ACTIVE_RUNS") and hasattr(_cfg, "ACTIVE_RUNS_LOCK"): with _cfg.ACTIVE_RUNS_LOCK: active_runs_snapshot: dict = dict(_cfg.ACTIVE_RUNS) elif hasattr(_cfg, "ACTIVE_RUNS"): active_runs_snapshot = dict(_cfg.ACTIVE_RUNS) else: active_runs_snapshot = {} with _cfg.STREAMS_LOCK: items = list(_cfg.STREAMS.items()) for stream_id, channel in items: meta = active_runs_snapshot.get(stream_id) owner_sid = (meta or {}).get("session_id") if isinstance(meta, dict) else None # Copilot review #3: skip non-matching and owner-unknown STREAMS # channels. Cross-turn delivery is handled by SESSION_CHANNELS below. if owner_sid != session_id: continue try: channel.put_nowait((event, data)) emitted += 1 except Exception: logger.debug("process_complete emit failed for stream %s", stream_id, exc_info=True) # Option X: also emit to the persistent per-session SSE channel. This is # the path that survives between turns (when STREAMS is torn down). We # keep the STREAMS emit above as defense-in-depth — if a turn IS active # the frontend dedupes by process_id, so a double-delivery is harmless. ch = get_session_channel(session_id) if ch is not None: try: delivered = ch.emit(event, data) emitted += delivered except Exception: logger.debug("SessionChannel emit failed for session %s", session_id, exc_info=True) return emitted def _emit_bg_task_complete_events_now(session_id: str, payload: dict) -> int: """Emit the canonical bg_task_complete event and temporary legacy alias.""" # T1 emit rename: the canonical event name is now ``bg_task_complete`` # (per maintainer decision on PR #2242). Until PR (b) lands the new # consumer-side listener, we ALSO emit the legacy ``process_complete`` # name so any in-flight WebUI build (subscribed to the old listener) # still receives the wakeup. PR (b) will: # 1. Add `bg_task_complete` listeners on the WebUI side. # 2. Remove this dual-emit shim (drop the legacy alias). # Both emits carry the SAME trimmed payload + the SAME event_id, so a # consumer that ever sees both can dedupe by ``event_id``. # # Greptile P2: hand each emit its OWN shallow copy of the payload. The # same dict object would otherwise be referenced by every subscriber queue # across STREAMS and SESSION_CHANNELS for BOTH event names; a downstream # consumer that mutates it in place would silently corrupt all other # concurrent consumers' views. Shallow copies are sufficient — the payload # is a flat trimmed dict of scalars. return ( _emit_to_session_streams(session_id, "bg_task_complete", dict(payload)) + _emit_to_session_streams(session_id, "process_complete", dict(payload)) ) def _flush_coalesced_bg_task_complete(session_id: str) -> None: """Timer callback: flush the latest pending payload for one session.""" payload: dict | None = None with _EMIT_COALESCE_LOCK: payload = _PENDING_EMIT_PAYLOADS.pop(session_id, None) _PENDING_EMIT_TIMERS.pop(session_id, None) if payload is not None: _LAST_EMIT_TS[session_id] = time.time() if payload is None: return try: _emit_bg_task_complete_events_now(session_id, payload) except Exception: logger.debug( "coalesced bg_task_complete flush failed for session %s", session_id, exc_info=True, ) def _emit_bg_task_complete_events_coalesced(session_id: str, payload: dict) -> int: """Per-session 1s coalesce gate around the bg_task_complete dual emit. The first payload for a session emits immediately. If another payload for that session arrives within ``_EMIT_COALESCE_WINDOW_SECS`` of the last emit, replace the pending payload and reset the quiet timer. The deferred flush therefore uses the latest payload from a burst. """ if not session_id: return 0 should_emit_now = False now = time.time() with _EMIT_COALESCE_LOCK: last = _LAST_EMIT_TS.get(session_id) has_pending = session_id in _PENDING_EMIT_TIMERS if last is None or (now - last) >= _EMIT_COALESCE_WINDOW_SECS: _LAST_EMIT_TS[session_id] = now should_emit_now = True if has_pending: _PENDING_EMIT_PAYLOADS.pop(session_id, None) old_timer = _PENDING_EMIT_TIMERS.pop(session_id, None) if old_timer is not None: try: old_timer.cancel() except Exception: logger.debug( "coalesced bg_task_complete timer cancel failed for session %s", session_id, exc_info=True, ) else: _PENDING_EMIT_PAYLOADS[session_id] = payload if not should_emit_now: old_timer = _PENDING_EMIT_TIMERS.get(session_id) if old_timer is not None: try: old_timer.cancel() except Exception: logger.debug( "coalesced bg_task_complete timer cancel failed for session %s", session_id, exc_info=True, ) timer = threading.Timer( _EMIT_COALESCE_WINDOW_SECS, _flush_coalesced_bg_task_complete, args=(session_id,), ) timer.daemon = True _PENDING_EMIT_TIMERS[session_id] = timer timer.start() if should_emit_now: return _emit_bg_task_complete_events_now(session_id, payload) return 0 # ── Coupling contract: agent ProcessRegistry cross-A/B dedupe key ────────── # This WebUI drain (B) and the merged upstream PR #2279 next-turn drain (A) # dedupe a process_id against a SINGLE shared key inside the agent's # ``tools.process_registry.ProcessRegistry``: # # * READ side: the PUBLIC ``is_completion_consumed(process_id)`` method # (used in ``_process_one`` above) — stable public API. # * WRITE side: there is NO public ``mark_completion_consumed`` upstream, so # B must reach into the registry's private ``_completion_consumed`` set # (guarded by its private ``_lock``) to set the shared marker A reads. # # That private WRITE coupling is what Copilot review #2242 comment #4 flagged. # The long-term fix is an upstream PUBLIC ``mark_completion_consumed`` — see # the test ``test_registry_completion_consumed_contract`` which fails CI LOUD # the moment ``_completion_consumed`` / ``_lock`` / ``is_completion_consumed`` # is renamed or retyped upstream, instead of a future rename silently # reintroducing the double-wakeup bug. ``_mark_registry_completion_consumed`` # narrows the exception handling so a rename is logged at ERROR (visible in # errors.log + monitoring) rather than swallowed by a broad ``except`` at # DEBUG. ImportError stays best-effort (the registry is legitimately absent in # non-agent unit-test contexts; this module's own locked # ``BG_TASK_COMPLETE_EVENTS_SEEN`` gate still dedupes B's own duplicates there). _REGISTRY_CONSUMED_CONTRACT = ("_lock", "_completion_consumed", "is_completion_consumed") def _mark_registry_completion_consumed(process_id: str) -> None: """Set the shared cross-A/B dedupe marker on the agent ProcessRegistry. Couples to ``ProcessRegistry`` privates (``_lock`` / ``_completion_consumed``) because no public ``mark_completion_consumed`` exists upstream (the read side uses the public ``is_completion_consumed``). A future upstream rename must FAIL LOUD, not silently reintroduce the double-wakeup: an ``AttributeError`` / ``TypeError`` from the private access is logged at ERROR with a contract-violation message (and ``test_registry_completion_consumed_contract`` breaks CI at test time). Only ``ImportError`` is treated as best-effort/expected (the registry is absent in pure unit-test contexts). """ try: from tools.process_registry import process_registry as _pr except ImportError: # Agent registry not importable (e.g. isolated unit test) — B's own # locked BG_TASK_COMPLETE_EVENTS_SEEN gate still prevents this module's # duplicates; cross-A/B dedupe is moot when A isn't running either. logger.debug( "tools.process_registry not importable; skipping shared " "completion-consumed marker (best-effort, expected off-agent)", exc_info=True, ) return try: lock = _pr._lock consumed = _pr._completion_consumed except AttributeError: logger.error( "ProcessRegistry coupling contract VIOLATED: expected private " "attrs %s for cross-A/B wakeup dedupe are missing — an upstream " "rename has broken the shared marker; process_complete wakeups may " "now double-fire. A public mark_completion_consumed() upstream is " "the durable fix (Copilot #2242 review #4).", _REGISTRY_CONSUMED_CONTRACT, exc_info=True, ) return try: with lock: consumed.add(process_id) except (AttributeError, TypeError): logger.error( "ProcessRegistry coupling contract VIOLATED: _lock/" "_completion_consumed changed shape (not a Lock / not a set) — " "cross-A/B wakeup dedupe is broken; wakeups may double-fire. " "Upstream public mark_completion_consumed() is the durable fix " "(Copilot #2242 review #4).", exc_info=True, ) # ── xsession wakeup misroute defense-in-depth (Option 3) ─────────────────── # Option 1 (api/streaming._set_turn_session_identity) is the ROOT fix: it binds # the per-turn session identity to a contextvar so a notify_on_complete spawn # can no longer capture a concurrent turn's process-global env. Option 3 is an # INDEPENDENT completion-time safety net at the wakeup-routing layer: even if # some future regression reintroduces a capture race, a positively-detected # mismatch must not wake the wrong session. # # The proc->owner link the WebUI drain trusts is ProcessSession.session_key, # which the terminal tool captured from the (historically racy) env at spawn. # An env-IMMUNE spawn-time owner would be the authoritative cross-check, but # adding such a field to the core ProcessSession is out of scope for this # WebUI-only change. So this resolver is forward-compatible by DUCK-TYPING: # if a future core grows an env-immune spawn-owner attribute (any of the # names below) AND it positively disagrees with the session_key-resolved # target, re-route to the env-immune owner and log ERROR. When the owner is # absent/empty/unknown (today's core, cron/CLI processes sharing the # registry, pre-Option-1 spawns) it is a PURE PASS-THROUGH — it never # suppresses a legitimate Option Z wakeup on uncertainty (Option Z must keep # working). _ENV_IMMUNE_OWNER_ATTRS = ("spawn_session_id", "owner_session_id", "turn_session_id") def _env_immune_spawn_owner(proc_session) -> str: """Return the env-immune spawn-time owner sid from the ProcessSession, or "" when none is available (the only contract today; forward-compatible).""" if proc_session is None: return "" for attr in _ENV_IMMUNE_OWNER_ATTRS: try: val = getattr(proc_session, attr, "") except Exception: val = "" if val: return str(val) return "" def _resolve_wakeup_target( *, process_id: str, session_key_resolved_sid: str, proc_session, ) -> str: """Cross-check the session_key-resolved wakeup target against the env-immune spawn owner. Returns the sid the server-side wakeup turn should actually target. - Owner unknown/empty -> pass-through (return session_key_resolved_sid). - Owner == resolved -> pass-through (the normal + post-Option-1 case). - Owner != resolved -> POSITIVE mismatch: log ERROR and RE-ROUTE to the env-immune owner (do NOT wake the wrong session — this is the exact agent.log:6632 cross-session misroute). """ resolved = str(session_key_resolved_sid or "") owner = _env_immune_spawn_owner(proc_session) if not owner or owner == resolved: return resolved logger.error( "xsession wakeup misroute BLOCKED (Option 3 safety net): process %r " "session_key resolved to session %r but the env-immune spawn owner " "is %r — re-routing the server-side wakeup to the true owner. This " "means a per-turn session-identity capture race occurred upstream " "(Option 1 should have prevented it); investigate streaming.py " "_set_turn_session_identity coverage.", process_id, resolved, owner, ) return owner def _process_one(evt: dict) -> None: """Route a single completion_queue event to the matching WebUI session.""" from api import config as _cfg # Hoist the process-registry import once per event: it was imported in # three separate blocks below (session_key recovery, env-immune owner # cross-check, upstream is_completion_consumed dedupe) on every completion # event, paying repeated import-system overhead. Single local rebind keeps # the ImportError fallback contract (process_registry may be missing in # cut-down vendoring) while collapsing to one lookup per call. try: from tools.process_registry import process_registry as _process_registry except Exception: _process_registry = None process_id = str(evt.get("session_id") or "") session_key = str(evt.get("session_key") or "") # Root-cause fix (t_0f447014): the notify_on_complete completion event # enqueued by ProcessRegistry._move_to_finished() carries NO "session_key" # field — only the watch_match enqueue includes one. Without it the old # `evt.get("session_key") or process_id` fell back to the process id # ("proc_xxxx"), which is never a PROCESS_SESSION_INDEX key (only # webui_session_id -> webui_session_id is registered at chat-start), so # every wakeup was silently dropped here and the frontend never POSTed an # ack. Recover the spawn-time session_key from the process registry's # ProcessSession: the terminal tool captured it synchronously at spawn # (while the turn's env was active), so it survives the turn-end env # restore and is the WebUI session_id for WebUI-spawned processes. if not session_key and process_id: try: if _process_registry is not None: _ps = _process_registry.get(process_id) if _ps is not None and getattr(_ps, "session_key", ""): session_key = str(_ps.session_key) except Exception: logger.debug( "session_key recovery from process registry failed for %r", process_id, exc_info=True, ) if not session_key: logger.debug( "process_complete drop: no recoverable session_key for process_id=%r", process_id, ) return with _cfg.PROCESS_SESSION_INDEX_LOCK: session_id = _cfg.PROCESS_SESSION_INDEX.get(session_key) if not session_id: # No mapping — could be a cron/gateway process that uses the same # registry but a non-WebUI session_key. Ignore. logger.debug("process_complete drop: no session mapping for key=%r", session_key) return # ── xsession wakeup misroute defense-in-depth (Option 3) ────────────── # session_id above came from PROCESS_SESSION_INDEX.get(session_key), and # session_key was captured by the terminal tool from the (historically # racy) process-global env at spawn. Option 1 binds the per-turn identity # to a contextvar so that capture is no longer racy — but as an INDEPENDENT # safety net, cross-check the resolved target against the env-immune # spawn owner (when the core ProcessSession exposes one). On a positive # mismatch this re-routes the wakeup (and the live-view emit + dedupe # markers below) to the TRUE owner instead of waking the wrong session. # Pure pass-through when no env-immune owner is available (today's core, # cron/CLI procs, pre-Option-1 spawns) — never suppresses a valid wakeup. try: _ps_xs = _process_registry.get(process_id) if (_process_registry is not None and process_id) else None except Exception: _ps_xs = None session_id = _resolve_wakeup_target( process_id=process_id, session_key_resolved_sid=session_id, proc_session=_ps_xs, ) # ── Idempotency vs the REAL merged upstream #2279 (shared dedupe key) ── # The real merged #2279 next-turn drain # (api/streaming._drain_webui_process_notifications) dedupes ONLY via # process_registry.is_completion_consumed() / _completion_consumed — it # does NOT populate BG_TASK_COMPLETE_EVENTS_SEEN (that set is ours-original # and private to this module). So the cross-A/B shared dedupe contract is # process_registry._completion_consumed, NOT BG_TASK_COMPLETE_EVENTS_SEEN. # If the upstream A-drain already delivered this process_id (A-first # order), it marked _completion_consumed; B must early-return here or it # would double-fire a wakeup. This guard aligns our B-drain to the real # upstream key (verified against origin/master streaming.py). if process_id: try: if _process_registry is not None and _process_registry.is_completion_consumed(process_id): return except Exception: logger.debug( "is_completion_consumed check failed on B drain; " "falling back to BG_TASK_COMPLETE_EVENTS_SEEN gate", exc_info=True, ) # Secondary (ours-original) idempotency: if we've already emitted for this # (session_id, process_id) pair via THIS module, skip the duplicate. Two # _move_to_finished() callers (kill_process racing the reader thread) can # occasionally enqueue twice despite the process_registry guard. with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK: seen = _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.setdefault(session_id, set()) if process_id and process_id in seen: return if process_id: seen.add(process_id) payload = _build_payload(evt, session_id) _emit_bg_task_complete_events_coalesced(session_id, payload) _cfg.PENDING_BG_TASK_COMPLETIONS.add(session_id) # Mark the event consumed in the agent's process registry so the REAL # merged PR #2279's next-turn drain # (api/streaming._drain_webui_process_notifications) treats this process_id # as already-delivered and does not re-fire a wakeup (B-first order). # This is the SHARED upstream dedupe key (see _mark_registry_completion_ # consumed for the coupling contract + why a future rename now fails loud). if process_id: _mark_registry_completion_consumed(process_id) # ── Option Z (PRIMARY): server-side wakeup, NO browser round-trip ────── # The SSE emit above is now demoted to a pure live-view layer (an open tab # streams the turn live via the per-session SSE channel). The ACTUAL agent # wakeup is started HERE, server-side, so a CLOSED tab still gets the turn # — parity with how CLI / Telegram / gateway self-wake from a # notify_on_complete completion. This is the fix for the structural flaw: # "fire a long background task, close the tab, come back later" is THE # primary background-task use case and browser-mediated wakeup could never # serve it. # # - turn ACTIVE → do NOT start a turn. Leave the PENDING_PROCESS_ # COMPLETIONS marker so PR #2279's next-turn drain # (api/streaming._drain_webui_process_notifications) injects the wakeup # when the active turn ends. (That path already works when a turn is # active — it was never the gap.) # - turn IDLE → start a new server-side turn directly with wakeup_prompt # as the user message (the real gap Option Z closes). # # Idempotency is already guaranteed above: BG_TASK_COMPLETE_EVENTS_SEEN + # the registry _completion_consumed marker mean this process_id reached # here at most once, so the wakeup turn starts at most once. try: # ``wakeup_prompt`` is server-internal state used only by the # Option Z server-side wakeup; it was previously surfaced on the # SSE payload but T1 trimmed the payload to the minimal shape # `{session_id, task_id, completed_at, summary?, event_id}`, so # we derive the prompt directly from the evt here (same source the # prior _build_payload used). wakeup_prompt_raw = format_wakeup_prompt(evt) wakeup_prompt = wakeup_prompt_raw.strip() if wakeup_prompt_raw else "" if wakeup_prompt: if _session_has_active_turn(session_id): # Defer-path fix: persist the prompt so a turn-teardown # idle-hook can redeliver it once the session goes idle. # The OLD behavior only logged + left a bare # PENDING_BG_TASK_COMPLETIONS session flag; the prompt was # discarded and the next-turn drain reads completion_queue # (already emptied by THIS drain thread), so for an # autonomous agent with no next user turn the wakeup was # lost forever. process_id is already in # BG_TASK_COMPLETE_EVENTS_SEEN + the registry # _completion_consumed marker (set above), so persisting it # here cannot cause a double-fire — the atomic claim in # ``claim_deferred_wakeups`` guarantees exactly one delivery. record_deferred_wakeup(session_id, process_id, wakeup_prompt) logger.debug( "server-side wakeup deferred: turn active for session %s " "(persisted for turn-teardown idle-hook redelivery)", session_id, ) else: # Idle-path sibling of the F1 (409/teardown) fix: pass # ``process_id`` so that if this idle wakeup's daemon thread # loses the per-session lock race and 409s, the re-defer in # ``_start_server_side_wakeup_turn`` records the entry WITH its # process_id — keeping the ``record_deferred_wakeup`` dedup # guard (``if process_id and any(...)``) live on that re-defer # path so a second 409 race cannot accumulate a duplicate # deferred entry (which would deliver the same wakeup twice). _start_server_side_wakeup_turn( session_id, wakeup_prompt, process_id=process_id ) except Exception: logger.warning( "server-side wakeup dispatch failed for session %s", session_id, exc_info=True ) def record_deferred_wakeup(session_id: str, process_id: str, wakeup_prompt: str) -> None: """Persist a deferred process-completion wakeup for later redelivery. Called from ``_process_one`` when a completion arrives while a turn is active (the Option Z drain branch cannot start a turn — it would 409). The turn-teardown idle-hook (``drain_deferred_wakeups_for_session``) redelivers it once the session goes idle, OR the PR #2279 next-turn drain claims it if a user turn comes first. Whoever claims first wins (atomic pop in ``claim_deferred_wakeups``); the other finds nothing. Idempotent per process_id: if the same process_id is already queued for this session (kill_process racing the reader thread), it is not appended twice. Best-effort — never raises into the drain loop. """ if not session_id or not wakeup_prompt: return from api import config as _cfg try: with _cfg.DEFERRED_PROCESS_WAKEUPS_LOCK: entries = _cfg.DEFERRED_PROCESS_WAKEUPS.setdefault(session_id, []) if process_id and any( e.get("process_id") == process_id for e in entries ): return entries.append( {"process_id": process_id, "wakeup_prompt": wakeup_prompt} ) except Exception: logger.debug( "record_deferred_wakeup failed for session %s", session_id, exc_info=True ) def claim_deferred_wakeups(session_id: str) -> list[dict]: """Atomically remove and return all deferred wakeups for *session_id*. The single-delivery guarantee for the defer path: the dict ``pop`` under ``DEFERRED_PROCESS_WAKEUPS_LOCK`` means whichever caller runs first (turn-teardown idle-hook OR PR #2279 next-turn drain) gets the entries and delivers them; every subsequent caller gets ``[]``. This is what makes the teardown hook idempotent with the next-turn drain (no double-fire) AND prevents a wakeup loop (the wakeup turn's own teardown re-runs the hook, finds nothing already-claimed → no re-fire). """ if not session_id: return [] from api import config as _cfg try: with _cfg.DEFERRED_PROCESS_WAKEUPS_LOCK: return _cfg.DEFERRED_PROCESS_WAKEUPS.pop(session_id, []) or [] except Exception: logger.debug( "claim_deferred_wakeups failed for session %s", session_id, exc_info=True ) return [] def drain_deferred_wakeups_for_session(session_id: str) -> int: """Turn-teardown idle-hook: redeliver deferred wakeups once idle. Called from ``api/streaming`` right AFTER ``unregister_active_run`` so ``_session_has_active_turn`` no longer counts the just-ended stream. This makes the active-at-completion case symmetric with the idle-at-completion case: idle now → fire now (Option Z idle branch); busy now → fire here when the turn ends and the session goes idle. Multi-stream / cancel-reconnect guard: if ANY other ACTIVE_RUNS row still exists for this session (a second stream from cancel/reconnect), the session is NOT yet idle — leave the deferred entries untouched so a later teardown (or the next-turn drain) delivers them. Only the teardown of the LAST active stream for the session claims + fires. Returns the number of wakeup turns started (0 when nothing pending or the session is still busy). Best-effort — never raises into the streaming teardown thread; the actual turn is started on the same throwaway daemon thread the idle branch uses, so this never blocks teardown. """ if not session_id: return 0 from api import config as _cfg try: # Multi-stream guard: only fire when the session is TRULY idle. if _session_has_active_turn(session_id): return 0 # Peek without claiming: avoid taking the entries then discovering # there is nothing to do under contention. with _cfg.DEFERRED_PROCESS_WAKEUPS_LOCK: if not _cfg.DEFERRED_PROCESS_WAKEUPS.get(session_id): return 0 # Atomic claim — exactly one caller gets the entries. entries = claim_deferred_wakeups(session_id) if not entries: return 0 # The session-level PENDING marker is server-internal telemetry; the # real delivery is the prompt(s) we just claimed. Discard it now that # the deferred wakeups are owned by this teardown. try: _cfg.PENDING_BG_TASK_COMPLETIONS.discard(session_id) except Exception: logger.debug( "PENDING discard failed for session %s", session_id, exc_info=True ) started = 0 # Greptile P1 fix: do NOT fire one daemon-threaded wakeup per entry in # a tight loop. Each ``_start_server_side_wakeup_turn`` spawns a daemon # thread that races for the per-session agent lock; only ONE can win, # the rest 409. Since we already claimed + popped every entry (line # ~938) and discarded the PENDING marker, the losers' prompts would be # permanently lost. Instead: start exactly the FIRST prompt, and # re-defer the remaining entries so each subsequent turn-teardown # (or next-turn drain) delivers the next one — one wakeup per turn, # which matches the single-prompt-per-turn design and the # BG_TASK_COMPLETE_EVENTS_SEEN dedup (no double-fire). leftover = [e for e in entries if str((e or {}).get("wakeup_prompt") or "").strip()] if leftover: first = leftover[0] # Re-defer entries 2..N BEFORE starting the first turn, so they are # already persisted if the first wakeup's own teardown re-runs this # hook and tries to claim them. for entry in leftover[1:]: record_deferred_wakeup( session_id, str((entry or {}).get("process_id") or ""), str((entry or {}).get("wakeup_prompt") or "").strip(), ) _start_server_side_wakeup_turn( session_id, str((first or {}).get("wakeup_prompt") or "").strip(), process_id=str((first or {}).get("process_id") or ""), ) started = 1 if started: logger.info( "turn-teardown idle-hook redelivered %d deferred wakeup(s) " "for session %s", started, session_id, ) return started except Exception: logger.warning( "drain_deferred_wakeups_for_session failed for session %s", session_id, exc_info=True, ) return 0 def _session_has_active_turn(session_id: str) -> bool: """True if a foreground/streaming agent turn is currently active for *session_id*. The drain thread has no Session object, so we key on ACTIVE_RUNS — the worker-lifecycle registry that this module already uses (see ``_emit_to_session_streams``) to map a stream back to its owning session. ACTIVE_RUNS is registered at agent-worker start and removed in the worker's outer ``finally``, so it survives cancel/reconnect races better than STREAMS. There is a brief window where ``_start_chat_stream_for_session`` has populated STREAMS but the worker thread has not yet called ``register_active_run``; in that window this returns False and the subsequent ``start_session_turn`` is rejected with a 409 by ``_start_chat_stream_for_session``'s own active-stream guard — i.e. the same lock /api/chat/start uses is the authoritative race backstop. """ 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 True except Exception: logger.debug("ACTIVE_RUNS active-turn check failed", exc_info=True) return False def _start_server_side_wakeup_turn( session_id: str, wakeup_prompt: str, *, process_id: str = "" ) -> None: """Start an agent turn server-side for a process_complete wakeup (Option Z). Runs on a short-lived daemon thread so the drain loop NEVER blocks: ``start_session_turn`` itself spawns the agent worker thread, but does synchronous session-load / workspace / model resolution first, which must not stall the single drain thread shared by every WebUI session. Concurrency + idempotency are enforced by the layers below, not here: - ``start_session_turn`` → ``_start_chat_stream_for_session`` serializes on the per-session agent lock and returns ``_status=409`` if a turn is already active. A human ``/api/chat/start`` racing this wakeup wins (one starts, the other 409s). On 409 we re-queue the prompt via ``record_deferred_wakeup`` (see below) so the racing turn's own teardown idle-hook — or PR #2279's next-turn drain — redelivers it. - ``BG_TASK_COMPLETE_EVENTS_SEEN`` already deduped this process_id in ``_process_one`` before we were called, so a process wakes at most once. Why re-queue on 409 instead of trusting the PENDING marker: this helper is called from two sites. The idle branch of ``_process_one`` leaves the PENDING_BG_TASK_COMPLETIONS marker intact, but the teardown-hook caller ``drain_deferred_wakeups_for_session`` has ALREADY atomically claimed the deferred entry and discarded the marker before spawning this thread. So in the teardown path a 409 would otherwise lose the wakeup permanently — nothing is left to drain. Re-queuing here is uniformly correct: it is idempotent per ``process_id`` (``record_deferred_wakeup`` dedupes) and the claim in ``claim_deferred_wakeups`` is atomic, so re-queue can never cause a double delivery. """ def _runner() -> None: try: from api.routes import start_session_turn resp = start_session_turn( session_id, wakeup_prompt, source="process_wakeup" ) status = int((resp or {}).get("_status", 200) or 200) if status == 409: # Raced an active turn (e.g. a human /api/chat/start, or a # sibling deferred-wakeup thread). Re-defer this prompt so it # is delivered by the winning turn's teardown / next-turn drain # instead of being lost. The atomic claim in # ``claim_deferred_wakeups`` still guarantees exactly-once # delivery, and BG_TASK_COMPLETE_EVENTS_SEEN already deduped # this process_id, so re-recording cannot double-fire. if wakeup_prompt: record_deferred_wakeup(session_id, process_id, wakeup_prompt) logger.debug( "server-side wakeup raced an active turn for session %s; " "re-deferred for redelivery on next teardown/turn", session_id, ) elif status >= 400: logger.warning( "server-side wakeup failed for session %s: status=%s err=%r", session_id, status, (resp or {}).get("error"), ) else: logger.info( "server-side wakeup turn started for session %s (stream_id=%s)", session_id, (resp or {}).get("stream_id"), ) except Exception: logger.warning( "server-side wakeup turn raised for session %s", session_id, exc_info=True, ) threading.Thread( target=_runner, name=f"hermes-webui-process-wakeup-{str(session_id)[:8]}", daemon=True, ).start() def _drain_loop() -> None: try: from tools import process_registry as _pr_mod # noqa: F401 from tools.process_registry import process_registry except Exception as exc: logger.warning("bg_task_complete drain unavailable: %s", exc) return logger.info("bg_task_complete drain thread started") while not _DRAIN_STOP.is_set(): try: evt = process_registry.completion_queue.get(timeout=1.0) except Exception: # queue.Empty or transient — re-check stop flag and continue. continue if not isinstance(evt, dict): continue try: _process_one(evt) except Exception: logger.warning("bg_task_complete event handling failed", exc_info=True) def register_process_session(session_key: str, session_id: str) -> None: """Bind a process-registry session_key to a WebUI session_id. Called at chat-start time, before the agent thread spawns any background processes. The same ``session_key`` is exported to the child via ``HERMES_SESSION_KEY`` (already done by streaming.py), so when the child pushes onto ``completion_queue`` it carries the key we registered. """ if not session_key or not session_id: return from api import config as _cfg with _cfg.PROCESS_SESSION_INDEX_LOCK: _cfg.PROCESS_SESSION_INDEX[str(session_key)] = str(session_id) def unregister_process_session(session_key: str) -> None: if not session_key: return from api import config as _cfg with _cfg.PROCESS_SESSION_INDEX_LOCK: _cfg.PROCESS_SESSION_INDEX.pop(str(session_key), None) def start_drain_thread() -> bool: """Start the background drain thread idempotently. Returns True on first start.""" global _DRAIN_THREAD if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive(): return False _DRAIN_STOP.clear() _DRAIN_THREAD = threading.Thread( target=_drain_loop, name="hermes-webui-bg-task-complete-drain", daemon=True, ) _DRAIN_THREAD.start() return True def stop_drain_thread(timeout: float = 2.0) -> None: _DRAIN_STOP.set() th = _DRAIN_THREAD if th is not None and th.is_alive(): th.join(timeout=timeout)