""" Process Registry -- In-memory registry for managed background processes. Tracks processes spawned via terminal(background=true), providing: - Output buffering (rolling 200KB window) - Status polling and log retrieval - Blocking wait with interrupt support - Process killing - Crash recovery via JSON checkpoint file - Session-scoped tracking for gateway reset protection Background processes execute THROUGH the environment interface -- nothing runs on the host machine unless TERMINAL_ENV=local. For Docker, Singularity, Modal, Daytona, and SSH backends, the command runs inside the sandbox. Usage: from tools.process_registry import process_registry # Spawn a background process (called from terminal_tool) session = process_registry.spawn(env, "pytest -v", task_id="task_123") # Poll for status result = process_registry.poll(session.id) # Block until done result = process_registry.wait(session.id, timeout=300) # Kill it process_registry.kill(session.id) """ import codecs import json import logging import os import platform import shlex import signal import subprocess import threading import time import uuid from pathlib import Path _IS_WINDOWS = platform.system() == "Windows" # systemd transient scopes exist only on Linux. Gate every scope-path branch # on this constant (not merely "not Windows") so macOS and other POSIX # platforms provably never touch systemd code (#70716 cross-platform audit). _IS_LINUX = platform.system() == "Linux" from tools.environments.local import _find_shell, _resolve_safe_cwd, _sanitize_subprocess_env from hermes_cli._subprocess_compat import windows_hide_flags from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from hermes_cli.config import get_hermes_home from agent.redact import redact_sensitive_text logger = logging.getLogger(__name__) # Checkpoint file for crash recovery (gateway only) CHECKPOINT_PATH = get_hermes_home() / "processes.json" # Limits MAX_OUTPUT_CHARS = 200_000 # 200KB rolling output buffer FINISHED_TTL_SECONDS = 1800 # Keep finished processes for 30 minutes MAX_PROCESSES = 64 # Max concurrent tracked processes (LRU pruning) MAX_ACTIVE_PROCESS_AGE = 86400 # 24h default — see session_reset.bg_process_max_age_hours (#29177) # Watch pattern rate limiting — PER SESSION. # Hard rule: at most ONE watch-match notification every WATCH_MIN_INTERVAL_SECONDS. # Any match arriving inside that cooldown window is dropped and counted as a strike. # After WATCH_STRIKE_LIMIT consecutive strike windows, watch_patterns for that # session is permanently disabled and the session falls back to notify_on_complete # semantics (one notification when the process actually exits). WATCH_MIN_INTERVAL_SECONDS = 15 # Minimum spacing between consecutive watch matches WATCH_STRIKE_LIMIT = 3 # Strikes in a row → disable watch + promote to notify_on_complete # Lifetime cap — independent of the strike counter above. A process whose # pattern recurs at a cadence just above WATCH_MIN_INTERVAL_SECONDS (e.g. a # service restarted repeatedly over a day) never trips the consecutive-strike # limit, since each match lands in its own clean cooldown window, yet still # forces a full-context agent turn every single time (#93513). watch_patterns # is documented as "ONLY for rare one-shot mid-process signals", so once a # session has delivered this many matches over its whole life we disable it # and fall back to notify_on_complete, same as the strike-limit path. WATCH_LIFETIME_MAX_HITS = 8 # Global circuit breaker — across all sessions. Secondary safety net so concurrent # siblings can't collectively flood the user even when each is under its own cap. WATCH_GLOBAL_MAX_PER_WINDOW = 15 WATCH_GLOBAL_WINDOW_SECONDS = 10 WATCH_GLOBAL_COOLDOWN_SECONDS = 30 # --------------------------------------------------------------------------- # systemd cgroup isolation for gateway-spawned local executors (#70716) # --------------------------------------------------------------------------- # When Hermes runs as a systemd gateway with MemoryHigh/MemoryMax limits, # local background terminal commands inherit the gateway's cgroup. A # memory-heavy executor (Codex, tests, Node) can push the whole cgroup past # MemoryMax and trigger systemd-oomd to kill the ENTIRE gateway — taking down # the messaging control plane and silently losing the active turn. # # Wrapping the spawn in ``systemd-run --user --scope --unit=hermes-worker-`` # places the worker in its own transient cgroup so an OOM in the worker kills # only the worker, not the gateway. We probe *once* whether # ``systemd-run --user --scope`` is actually usable (the binary can exist on # the PATH while the user D-Bus session is unavailable — common for system # services and containers), and cache the result for the process lifetime. _SYSTEMD_SCOPE_AVAILABLE: Optional[bool] = None _SYSTEMD_SCOPE_PROBE_LOCK = threading.Lock() _SYSTEMD_SCOPE_PROBED_AT = 0.0 _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS = 60.0 _MIN_WORKER_MEMORY_MAX_BYTES = 64 * 1024 * 1024 _DEFAULT_WORKER_MEMORY_MAX_BYTES = 1024 * 1024 * 1024 _WORKER_MEMORY_MAX_CAP_BYTES = 4 * 1024 * 1024 * 1024 def _worker_memory_max_bytes() -> int: """Return a finite per-worker cgroup limit without widening host risk. The proposed local-memory-guard environment override is honored when it tightens the safe bound, so this isolation composes with PR #57121 instead of inventing a second knob. An oversized override cannot widen host risk. Otherwise retain the tighter of the gateway's current cgroup-v2 ``memory.max`` and half of physical RAM, capped at 4 GiB. This keeps the sibling worker outside the gateway cgroup while ensuring the worker cannot consume memory up to the enclosing user slice or host limit. """ override_bound: Optional[int] = None override = os.getenv("TERMINAL_LOCAL_MEMORY_MAX_MB", "").strip() if override: override_valid = False try: parsed = int(override) * 1024 * 1024 if parsed >= _MIN_WORKER_MEMORY_MAX_BYTES: override_bound = parsed override_valid = True except ValueError: pass if not override_valid: logger.warning( "Ignoring invalid TERMINAL_LOCAL_MEMORY_MAX_MB=%r; " "expected an integer representing at least %d MiB", override, _MIN_WORKER_MEMORY_MAX_BYTES // (1024 * 1024), ) candidates: List[int] = [] try: for line in Path("/proc/self/cgroup").read_text(encoding="utf-8").splitlines(): if line.startswith("0::"): relative = line.partition("::")[2].lstrip("/") raw_limit = ( Path("/sys/fs/cgroup") / relative / "memory.max" ).read_text(encoding="utf-8").strip() if raw_limit.isdigit(): cgroup_limit = int(raw_limit) if cgroup_limit >= _MIN_WORKER_MEMORY_MAX_BYTES: candidates.append(cgroup_limit) break except (OSError, ValueError): pass try: physical_bytes = int(os.sysconf("SC_PHYS_PAGES")) * int( os.sysconf("SC_PAGE_SIZE") ) physical_bound = min( _WORKER_MEMORY_MAX_CAP_BYTES, max(_MIN_WORKER_MEMORY_MAX_BYTES, physical_bytes // 2), ) candidates.append(physical_bound) except (OSError, ValueError, TypeError): pass safe_bound = min(candidates) if candidates else _DEFAULT_WORKER_MEMORY_MAX_BYTES return min(override_bound, safe_bound) if override_bound else safe_bound def _systemd_run_user_scope_available() -> bool: """Return True if ``systemd-run --user --scope`` can create a cgroup. Cached after the first probe. ``shutil.which`` alone is insufficient: in system-service deployments (and containers) the user D-Bus session bus that ``systemd-run --user`` needs may be absent even though the binary is on PATH, causing every spawn to fail with ``Failed to connect to user bus``. We do a cheap no-op probe (``systemd-run --user --scope --unit=… -- /bin/true``) and remember the outcome. """ global _SYSTEMD_SCOPE_AVAILABLE, _SYSTEMD_SCOPE_PROBED_AT cached = _SYSTEMD_SCOPE_AVAILABLE now = time.monotonic() if cached is True: return True if ( cached is False and now - _SYSTEMD_SCOPE_PROBED_AT < _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS ): return False # Double-checked locking keeps concurrent first-use spawns from observing # a temporary False while the definitive probe is still in flight. Such a # race would launch the losing workload back inside the gateway cgroup. with _SYSTEMD_SCOPE_PROBE_LOCK: cached = _SYSTEMD_SCOPE_AVAILABLE now = time.monotonic() if cached is True: return True if ( cached is False and now - _SYSTEMD_SCOPE_PROBED_AT < _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS ): return False available = False if _IS_LINUX: try: import shutil binary = shutil.which("systemd-run") if binary: # Probe: create a transient scope that immediately exits. # A unique unit avoids collisions; timeout bounds D-Bus. probe_unit = f"hermes-probe-scope-{os.getpid()}-{uuid.uuid4().hex[:8]}" result = subprocess.run( [ binary, "--user", "--scope", "--quiet", "--unit", probe_unit, "--collect", "--property", "MemoryAccounting=yes", "--property", f"MemoryMax={_worker_memory_max_bytes()}", "--property", "OOMPolicy=kill", "--", "/bin/true", ], capture_output=True, timeout=3, ) available = result.returncode == 0 if not available: logger.debug( "systemd-run --user --scope probe failed (rc=%s): %s", result.returncode, (result.stderr or b"").decode( "utf-8", "replace" ).strip(), ) except Exception as exc: logger.debug("systemd-run --user --scope probe error: %s", exc) _SYSTEMD_SCOPE_AVAILABLE = available _SYSTEMD_SCOPE_PROBED_AT = time.monotonic() return available def _is_supervised_gateway_process() -> bool: """Return whether this process is in a supervised Hermes gateway runtime. Both supervisor markers and ``_HERMES_GATEWAY`` are inherited by every descendant, and importing ``gateway.run`` also sets the latter. Require this process to own the live gateway PID file as well. That keeps transient systemd scopes limited to the gateway itself instead of terminal children or unrelated interactive CLIs in the same supervised process tree. """ if os.environ.get("_HERMES_GATEWAY") != "1": return False try: from gateway.restart import is_gateway_supervisor_process from gateway.status import get_running_pid return ( is_gateway_supervisor_process() and get_running_pid(cleanup_stale=False) == os.getpid() ) except Exception as exc: logger.debug("Could not verify supervised gateway process identity: %s", exc) return False def _build_systemd_scope_argv( shell_argv: List[str], unit_suffix: str, ) -> List[str]: """Wrap *shell_argv* in a ``systemd-run --user --scope`` invocation. The resulting cgroup gets its own memory accounting so an OOM in the worker does not kill the gateway cgroup (#70716). ``--collect`` makes the transient scope self-clean after exit; ``--unit`` gives it a recognisable name for ``systemctl --user status`` / journalctl. """ import shutil binary = shutil.which("systemd-run") if binary is None: # Caller should have checked _systemd_run_user_scope_available(); # guard anyway so we never pass None into Popen. return shell_argv unit_name = f"hermes-worker-{unit_suffix}" memory_max = _worker_memory_max_bytes() return [ binary, "--user", "--scope", "--quiet", "--unit", unit_name, "--collect", "--property", "MemoryAccounting=yes", "--property", f"MemoryMax={memory_max}", "--property", "OOMPolicy=kill", "--", *shell_argv, ] def restart_safe_gateway_child_argv( command: List[str], *, unit_suffix: str ) -> List[str]: """Place a managed-systemd gateway child outside the gateway cgroup. Children that must survive an intentional gateway restart cannot rely on ``start_new_session`` alone: systemd still kills every process in the service cgroup. In that topology, require a transient user scope and fail closed if it cannot be established. Standalone processes, non-systemd supervisors, and non-Linux hosts retain the direct command. """ if not _IS_LINUX: return command if not _is_supervised_gateway_process() or not os.environ.get("INVOCATION_ID"): return command if not _systemd_run_user_scope_available(): raise RuntimeError( "cannot create restart-safe systemd scope for gateway child: " "systemd-run --user --scope is unavailable" ) scoped = _build_systemd_scope_argv(command, unit_suffix=unit_suffix) if scoped == command: raise RuntimeError( "cannot create restart-safe systemd scope for gateway child: " "systemd-run disappeared after the availability probe" ) return scoped def _stop_systemd_unit(unit_name: str) -> bool: """Stop a transient systemd user scope by unit name. This reaps the *entire* cgroup — catching double-forked descendants that survive a plain PID signal because they were reparented to init inside the scope (issue #70716, reviewer gap #2). ``systemctl --user stop`` sends SIGTERM to every process in the unit's cgroup and escalates to SIGKILL after the unit's ``TimeoutStopSec``. Returns True if the unit was successfully stopped (or was already gone), False if ``systemctl`` is unavailable or the stop command failed. """ import shutil binary = shutil.which("systemctl") if binary is None: return False try: result = subprocess.run( [binary, "--user", "stop", unit_name], capture_output=True, timeout=15, ) if result.returncode != 0: stderr = (result.stderr or b"").decode(errors="replace").strip() stderr_lower = stderr.lower() if any( marker in stderr_lower for marker in ("not loaded", "not found", "does not exist") ): return True logger.debug( "systemctl --user stop %s exited %d: %s", unit_name, result.returncode, stderr, ) return False return True except Exception as exc: logger.debug("systemctl --user stop %s failed: %s", unit_name, exc) return False def format_uptime_short(seconds: int) -> str: s = max(0, int(seconds)) if s < 60: return f"{s}s" mins, secs = divmod(s, 60) if mins < 60: return f"{mins}m {secs}s" hours, mins = divmod(mins, 60) return f"{hours}h {mins}m" @dataclass class ProcessSession: """A tracked background process with output buffering.""" id: str # Unique session ID ("proc_xxxxxxxxxxxx") command: str # Original command string task_id: str = "" # Task/sandbox isolation key owner_task_id: str = "" # RAW spawning task id (e.g. subagent "sa-..."); # task_id is the CONTAINER key and may be collapsed # to "default"/session key by _resolve_container_task_id, # so ownership checks must use this field (#child-notify) session_key: str = "" # Gateway session key (for reset protection) pid: Optional[int] = None # OS process ID process: Optional[subprocess.Popen] = None # Popen handle (local only) env_ref: Any = None # Reference to the environment object cwd: Optional[str] = None # Working directory started_at: float = 0.0 # time.time() of spawn (wall clock) host_start_time: Optional[int] = None # kernel start ticks (/proc//stat f22) — PID-reuse guard exited: bool = False # Whether the process has finished exit_code: Optional[int] = None # Exit code (None if still running) completion_reason: str = "exited" # exited|killed|lost|failed_start|already_exited termination_source: str = "" # process.kill|kill_all|backend_lost|failed_start output_buffer: str = "" # Rolling output (last MAX_OUTPUT_CHARS) max_output_chars: int = MAX_OUTPUT_CHARS detached: bool = False # True if recovered from crash (no pipe) pid_scope: str = "host" # "host" for local/PTY PIDs, "sandbox" for env-local PIDs systemd_unit: str = "" # transient scope unit name when spawned under systemd-run (#70716) # Watcher/notification metadata (persisted for crash recovery) watcher_platform: str = "" watcher_chat_id: str = "" watcher_user_id: str = "" watcher_user_name: str = "" watcher_thread_id: str = "" watcher_message_id: str = "" # Triggering message id — reply anchor for topic routing watcher_interval: int = 0 # 0 = no watcher configured # Session-db id of the conversation that spawned this process. Lets the # gateway's completion pre-flight (_classify_completion_target) drop # notifications whose spawning session was closed at an explicit user # boundary (/new), instead of injecting them into the chat's NEW session. parent_session_id: str = "" notify_on_complete: bool = False # Queue agent notification on exit # Watch patterns — trigger agent notification when output matches any pattern watch_patterns: List[str] = field(default_factory=list) _watch_hits: int = field(default=0, repr=False) # total matches delivered _watch_suppressed: int = field(default=0, repr=False) # matches dropped by rate limit _watch_disabled: bool = field(default=False, repr=False) # permanently killed after strike limit # Per-session rate limit state: at most one match every WATCH_MIN_INTERVAL_SECONDS. # When an emission happens, _watch_cooldown_until is set to now + interval and # _watch_strike_candidate becomes True. The next match to arrive before that # deadline counts as one strike (regardless of how many matches were dropped in # between — a strike is a window, not a match). After WATCH_STRIKE_LIMIT strikes # in a row, watch_patterns is disabled and the session promotes to # notify_on_complete. _watch_last_emit_at: float = field(default=0.0, repr=False) _watch_cooldown_until: float = field(default=0.0, repr=False) _watch_strike_candidate: bool = field(default=False, repr=False) _watch_consecutive_strikes: int = field(default=0, repr=False) _completion_event: threading.Event = field(default_factory=threading.Event, repr=False) _lock: threading.Lock = field(default_factory=threading.Lock) _reader_thread: Optional[threading.Thread] = field(default=None, repr=False) _pty: Any = field(default=None, repr=False) # ptyprocess handle (when use_pty=True) class ProcessRegistry: """ In-memory registry of running and finished background processes. Thread-safe. Accessed from: - Executor threads (terminal_tool, process tool handlers) - Gateway asyncio loop (watcher tasks, session reset checks) - Cleanup thread (sandbox reaping coordination) """ _SHELL_NOISE_SUBSTRINGS = ( "bash: cannot set terminal process group", "bash: no job control in this shell", "no job control in this shell", "cannot set terminal process group", "tcsetattr: Inappropriate ioctl for device", ) def __init__(self): self._running: Dict[str, ProcessSession] = {} self._finished: Dict[str, ProcessSession] = {} self._lock = threading.Lock() # Side-channel for check_interval watchers (gateway reads after agent run) self.pending_watchers: List[Dict[str, Any]] = [] # Notification queue — unified queue for all background process events. # Completion notifications (notify_on_complete) and watch pattern matches # both land here, distinguished by "type" field. CLI process_loop and # gateway drain this after each agent turn to auto-trigger new turns. import queue as _queue_mod self.completion_queue: _queue_mod.Queue = _queue_mod.Queue() # Rehydrate durable delegation completions only at registry startup. # Consumers still inject them as fresh turns through this existing rail. try: from tools.async_delegation import restore_undelivered_completions restore_undelivered_completions(self.completion_queue) except Exception as exc: logger.warning("Could not restore async delegation completions: %s", exc) # Track sessions whose completion was already consumed by the agent # via wait/log. Drain loops AND gateway/tui watchers skip notifications # for these — a blocking wait() or a full read_log() means the agent # has the output in hand and is acting on it this turn. self._completion_consumed: set = set() # Track sessions the agent merely *observed* exited via poll(). poll() # is a read-only status check, so it does NOT mark _completion_consumed # (that would let a status check suppress the gateway/tui watcher's # autonomous delivery turn — #10156). But on the CLI the poll result # is returned inline in the same turn, so the idle/post-turn drain must # still skip the queued completion to avoid a duplicate [SYSTEM: ...] # injection (the bug #8228 originally fixed). drain_notifications() # consults this set; the gateway/tui watchers deliberately do NOT. self._poll_observed: set = set() # Global watch-match circuit breaker — across all sessions. # Prevents sibling processes from collectively flooding the user even # when each stays under its own per-session cap. self._global_watch_lock = threading.Lock() self._global_watch_window_start: float = 0.0 self._global_watch_window_hits: int = 0 self._global_watch_tripped_until: float = 0.0 self._global_watch_suppressed_during_trip: int = 0 # Live-output sink set by a driver (e.g. the desktop gateway): called from # reader threads with (session, chunk) to stream output to a UI in # real time, instead of polling the output tail. self.on_output = None # Close-view sink set by a driver (desktop gateway): called with # (session_or_none, process_id) when the agent asks to close a read-only # terminal tab. Distinct from kill — the process keeps running; only the # UI view is dropped (the user can reopen it from the status stack). self.on_close = None @staticmethod def _clean_shell_noise(text: str) -> str: """Strip shell startup warnings from the beginning of output.""" lines = text.split("\n") while lines and any(noise in lines[0] for noise in ProcessRegistry._SHELL_NOISE_SUBSTRINGS): lines.pop(0) return "\n".join(lines) def _emit_output(self, session: ProcessSession, chunk: str) -> None: """Forward a freshly-read chunk to the live-output sink, if one is set. Called from reader threads; never raise into the read loop.""" sink = self.on_output if sink is None or not chunk: return try: sink(session, chunk) except Exception: pass def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: """Scan new output for watch patterns and queue notifications. Called from reader threads with new_text being the freshly-read chunk. Per-session rate limit: at most ONE watch-match notification per WATCH_MIN_INTERVAL_SECONDS. Any match arriving inside the cooldown window is dropped and counts as ONE strike for that window. After WATCH_STRIKE_LIMIT consecutive strike windows, watch_patterns is disabled for this session and the session is promoted to notify_on_complete semantics — one notification when the process actually exits, no more mid-process spam. Independently, WATCH_LIFETIME_MAX_HITS caps the total number of matches ever delivered for a session, so a pattern that keeps recurring at a cadence just above the cooldown (e.g. a service restarted repeatedly over a day) still gets disabled instead of forcing a full-context agent turn indefinitely. """ if not session.watch_patterns or session._watch_disabled: return # Suppress-after-exit: once the reader loop has declared the process # exited, any late chunk we still see is post-exit noise. Dropping these # prevents the "stale notifications delivered minutes after the process # ended" spam when completion_queue consumers run async. if session.exited: return # Scan new text line-by-line for pattern matches matched_lines = [] matched_pattern = None for line in new_text.splitlines(): for pat in session.watch_patterns: if pat in line: matched_lines.append(line.rstrip()) if matched_pattern is None: matched_pattern = pat break # one match per line is enough if not matched_lines: return now = time.time() should_disable = False lifetime_exhausted = False with session._lock: # Case 1: still inside the cooldown from the last emission. # Count this as a strike for the current window (only once per window) # and drop the event. If we've hit the strike limit, disable watch # and promote to notify_on_complete. if session._watch_cooldown_until and now < session._watch_cooldown_until: session._watch_suppressed += len(matched_lines) if not session._watch_strike_candidate: # First drop in this window — count one strike. session._watch_strike_candidate = True session._watch_consecutive_strikes += 1 if session._watch_consecutive_strikes >= WATCH_STRIKE_LIMIT: session._watch_disabled = True # Promote to notify_on_complete so the agent still gets # exactly one notification when the process actually ends. session.notify_on_complete = True should_disable = True return_early = True else: # Case 2: cooldown has expired. # Decide whether this window was a "clean" one (no drops) or a # strike window. If no strike candidate was set during the prior # cooldown, reset the consecutive-strike counter — we're back to # healthy emission cadence. if ( session._watch_cooldown_until and not session._watch_strike_candidate ): session._watch_consecutive_strikes = 0 session._watch_strike_candidate = False # Emit the notification and start a new cooldown window. session._watch_last_emit_at = now session._watch_cooldown_until = now + WATCH_MIN_INTERVAL_SECONDS session._watch_hits += 1 suppressed = session._watch_suppressed session._watch_suppressed = 0 return_early = False # Lifetime cap: this match is delivered (it already earned it), # but disable further ones regardless of how cleanly spaced # they are — see WATCH_LIFETIME_MAX_HITS above. lifetime_exhausted = session._watch_hits >= WATCH_LIFETIME_MAX_HITS if lifetime_exhausted: session._watch_disabled = True session.notify_on_complete = True if return_early: if should_disable: # Emit exactly one "watch disabled, falling back to notify_on_complete" # summary event so the agent/user sees why things went quiet. self.completion_queue.put({ "session_id": session.id, "session_key": session.session_key, "task_id": session.task_id, "owner_task_id": session.owner_task_id or session.task_id, "command": session.command, "type": "watch_disabled", "suppressed": session._watch_suppressed, "platform": session.watcher_platform, "chat_id": session.watcher_chat_id, "user_id": session.watcher_user_id, "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, "message": ( f"Watch patterns disabled for process {session.id} — " f"{WATCH_STRIKE_LIMIT} consecutive rate-limit windows triggered " f"(min spacing {WATCH_MIN_INTERVAL_SECONDS}s). " f"Falling back to notify_on_complete semantics; you'll get " f"exactly one notification when the process exits." ), }) return # Trim matched output to a reasonable size output = "\n".join(matched_lines[:20]) if len(output) > 2000: output = output[:2000] + "\n...(truncated)" # Global circuit breaker — across all sessions (secondary safety net). if not self._global_watch_admit(now): if lifetime_exhausted: # The final match was dropped by the global breaker, but the # session is already disabled — still tell the user why things # went quiet (the strike path emits its summary unconditionally # too). self._emit_lifetime_watch_disabled(session) return notification = { "session_id": session.id, "session_key": session.session_key, "task_id": session.task_id, "owner_task_id": session.owner_task_id or session.task_id, "command": session.command, "type": "watch_match", "pattern": matched_pattern, "output": output, "suppressed": suppressed, "platform": session.watcher_platform, "chat_id": session.watcher_chat_id, "user_id": session.watcher_user_id, "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, } _redact_process_result(notification) self.completion_queue.put(notification) if lifetime_exhausted: # Same "why things went quiet" summary as the strike-limit path, # queued right after the final delivered match. self._emit_lifetime_watch_disabled(session) def _emit_lifetime_watch_disabled(self, session: ProcessSession) -> None: """Queue the watch_disabled summary for the lifetime-cap path (#93513).""" self.completion_queue.put({ "session_id": session.id, "session_key": session.session_key, "task_id": session.task_id, "owner_task_id": session.owner_task_id or session.task_id, "command": session.command, "type": "watch_disabled", "suppressed": 0, "platform": session.watcher_platform, "chat_id": session.watcher_chat_id, "user_id": session.watcher_user_id, "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, "message": ( f"Watch patterns disabled for process {session.id} — " f"reached the lifetime cap of {WATCH_LIFETIME_MAX_HITS} delivered " f"matches. Falling back to notify_on_complete semantics; you'll get " f"exactly one notification when the process exits." ), }) def _global_watch_admit(self, now: float) -> bool: """Return True if this watch_match event is allowed through the global breaker. Semantics: - If we're currently in a cooldown period, drop the event and count it. - Otherwise, slide the rolling window and check the global cap. - If the cap is exceeded, trip the breaker for WATCH_GLOBAL_COOLDOWN_SECONDS and emit ONE summary event so the agent/user sees "N notifications were suppressed" instead of getting them individually. - When the cooldown ends, emit a release summary and reset counters. """ with self._global_watch_lock: # Handle cooldown expiry first so we can emit the release summary. if self._global_watch_tripped_until and now >= self._global_watch_tripped_until: suppressed = self._global_watch_suppressed_during_trip self._global_watch_tripped_until = 0.0 self._global_watch_suppressed_during_trip = 0 self._global_watch_window_start = now self._global_watch_window_hits = 0 if suppressed > 0: # Queue a summary event outside the lock (below). release_msg = { "session_id": "", "session_key": "", "command": "", "type": "watch_overflow_released", "suppressed": suppressed, "message": ( f"Watch-pattern notifications resumed. " f"{suppressed} match event(s) were suppressed during the flood." ), "platform": "", "chat_id": "", "user_id": "", "user_name": "", "thread_id": "", } else: release_msg = None else: release_msg = None # Still in cooldown — drop and count. if self._global_watch_tripped_until and now < self._global_watch_tripped_until: self._global_watch_suppressed_during_trip += 1 admit = False trip_now = None else: # Slide the window. if now - self._global_watch_window_start >= WATCH_GLOBAL_WINDOW_SECONDS: self._global_watch_window_start = now self._global_watch_window_hits = 0 if self._global_watch_window_hits >= WATCH_GLOBAL_MAX_PER_WINDOW: # Trip the breaker. self._global_watch_tripped_until = now + WATCH_GLOBAL_COOLDOWN_SECONDS self._global_watch_suppressed_during_trip += 1 trip_now = now admit = False else: self._global_watch_window_hits += 1 trip_now = None admit = True # Queue summary events outside the lock. if release_msg is not None: self.completion_queue.put(release_msg) if trip_now is not None: self.completion_queue.put({ "session_id": "", "session_key": "", "command": "", "type": "watch_overflow_tripped", "message": ( f"Watch-pattern overflow: >{WATCH_GLOBAL_MAX_PER_WINDOW} " f"notifications in {WATCH_GLOBAL_WINDOW_SECONDS}s across all processes. " f"Suppressing further watch_match events for " f"{WATCH_GLOBAL_COOLDOWN_SECONDS}s." ), "platform": "", "chat_id": "", "user_id": "", "user_name": "", "thread_id": "", }) return admit @staticmethod def _is_host_pid_alive(pid: Optional[int]) -> bool: """Best-effort liveness check for host-visible PIDs.""" if not pid: return False # ``os.kill(pid, 0)`` is NOT a no-op on Windows (bpo-14484) — use # the cross-platform existence check. from gateway.status import _pid_exists return _pid_exists(pid) @staticmethod def _safe_host_start_time(pid: Optional[int]) -> Optional[int]: """Kernel start ticks for a host PID, or None when unavailable.""" if not pid: return None try: from gateway.status import get_process_start_time return get_process_start_time(pid) except Exception: return None @classmethod def _host_pid_is_ours(cls, pid: Optional[int], expected_start: Optional[int]) -> bool: """True only if ``pid`` is alive AND still the process we spawned. The kernel recycles PID/PGID numbers once a process exits and is reaped, so a stored PID can later name an *unrelated* process — observed in the wild as a recycled number landing on a desktop browser's session leader, which our tree-kill then SIGTERMs (Firefox dying at irregular intervals). We compare the kernel start time captured at spawn against the live one; a mismatch means the number was recycled and must never be signalled. When no baseline was captured (legacy checkpoints, or platforms without ``/proc``) we degrade to a bare liveness check rather than refusing to act, preserving prior best-effort behaviour. """ if not cls._is_host_pid_alive(pid): return False if expected_start is None: return True return cls._safe_host_start_time(pid) == expected_start def _refresh_detached_session(self, session: Optional[ProcessSession]) -> Optional[ProcessSession]: """Update recovered host-PID sessions when the underlying process has exited.""" if session is None or session.exited or not session.detached or session.pid_scope != "host": return session # Identity-aware liveness: a recycled PID (alive but a different process # than we spawned) must be treated as "our process exited", so it is # moved to finished and can never be tree-killed by a later kill(). if self._host_pid_is_ours(session.pid, session.host_start_time): return session with session._lock: if session.exited: return session session.exited = True # Recovered sessions no longer have a waitable handle, so the real # exit code is unavailable once the original process object is gone. session.exit_code = None self._move_to_finished(session) return session @staticmethod def _proc_alive(proc) -> bool: """True if a psutil.Process is running and not a zombie. A zombie is already dead (just unreaped), so there's nothing to SIGKILL. """ try: import psutil if not proc.is_running(): return False return proc.status() != psutil.STATUS_ZOMBIE except Exception: return False @staticmethod def _daemon_term_grace_seconds() -> float: """Grace window (s) between SIGTERM and escalated SIGKILL. Read from ``terminal.daemon_term_grace_seconds`` in config.yaml; floored at 0 (0 disables escalation). Falls back to the DEFAULT_CONFIG value if config is unreadable, so callers always get a sane number. """ try: from hermes_cli.config import read_raw_config, cfg_get, DEFAULT_CONFIG cfg = read_raw_config() val = cfg_get(cfg, "terminal", "daemon_term_grace_seconds") if val is None: val = DEFAULT_CONFIG["terminal"]["daemon_term_grace_seconds"] return max(float(val), 0.0) except Exception: return 2.0 @classmethod def _terminate_host_pid(cls, pid: int, expected_start: Optional[int] = None) -> None: """Terminate a host-visible PID and its descendants. ``expected_start`` is the kernel start time captured when we spawned the process. When provided, it is re-validated against the live PID before any signal is sent; a mismatch (or a dead PID) means the number was recycled onto an unrelated process and we refuse to touch it, so a stale background-session PID can never tree-kill a browser or other stranger. POSIX: walks the process tree with ``psutil`` and SIGTERMs children before the parent so subprocess trees (e.g. Chromium renderers/GPU helpers spawned by an ``agent-browser`` daemon) don't get reparented to init and survive cleanup. After a bounded grace window (``terminal.daemon_term_grace_seconds``) any tree member that ignored SIGTERM — a daemon stalled in its signal handler — is escalated to SIGKILL so it can't leak indefinitely. Set the grace to 0 to disable escalation (SIGTERM only). Windows: shells out to ``taskkill /PID /T /F``. This is the documented Microsoft primitive for tree-kill and matches the existing convention in ``gateway.status.terminate_pid``. ``/F`` is already a hard kill, so no separate escalation step is needed. We can't reuse the POSIX psutil path on Windows because: 1. Windows doesn't maintain a Unix-style process tree — ``psutil.Process.children(recursive=True)`` walks PPID links that go stale when intermediate processes exit, so enumeration is best-effort and misses orphaned descendants. 2. ``psutil.Process.terminate()`` on Windows is ``TerminateProcess()`` which kills only the target handle and is a hard kill — there is no Windows equivalent of a SIGTERM that cascades through a process group. (See the warning in ``gateway/status.py::terminate_pid``: "os.kill with SIGTERM is not equivalent to a tree-killing hard stop" on Windows.) Headless Chromium has no GUI window, so the softer ``taskkill /T`` without ``/F`` won't reach it either. ``psutil`` is a hard dependency (see ``pyproject.toml``); the bare-``os.kill`` fallback covers OSError / PermissionError on POSIX and a missing ``taskkill.exe`` on Windows (effectively unreachable on real Windows installs, but cheap insurance). """ if expected_start is not None and not cls._host_pid_is_ours(pid, expected_start): # PID was recycled (start time changed) or is gone — never signal a # stranger. A leaked orphan is strictly preferable to killing e.g. # a browser whose session leader reused this dead session's PID. logger.warning( "Refusing to terminate host pid %d: start-time mismatch — " "PID was recycled onto an unrelated process.", pid, ) return if _IS_WINDOWS: try: subprocess.run( ["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=10, creationflags=windows_hide_flags(), stdin=subprocess.DEVNULL, ) except (FileNotFoundError, subprocess.TimeoutExpired, OSError): try: os.kill(pid, signal.SIGTERM) except (OSError, ProcessLookupError, PermissionError): pass return import psutil try: parent = psutil.Process(pid) except psutil.NoSuchProcess: return except (OSError, PermissionError): try: os.kill(pid, signal.SIGTERM) except (OSError, ProcessLookupError, PermissionError): pass return # Snapshot the whole tree (children before parent) and SIGTERM each. try: targets = parent.children(recursive=True) except (psutil.NoSuchProcess, psutil.AccessDenied, OSError): targets = [] targets.append(parent) for proc in targets: try: proc.terminate() except psutil.NoSuchProcess: pass except (psutil.AccessDenied, OSError): pass # Escalate to SIGKILL for anything that ignored SIGTERM within the # grace window — a daemon stalled in its signal handler would otherwise # leak indefinitely. grace = cls._daemon_term_grace_seconds() if grace <= 0: return # Sleep out the grace window, then independently re-probe every target # and SIGKILL any survivor. We deliberately do NOT trust # ``psutil.wait_procs``'s gone/alive partition here: it reaps via # ``Process.wait()`` and can mis-partition when a target transitions # through a zombie state or when reaping is racy across a parent/child # tree, which left survivors un-killed. A direct liveness re-probe is # deterministic. deadline = time.monotonic() + grace while time.monotonic() < deadline: if not any(cls._proc_alive(_p) for _p in targets): break time.sleep(0.05) for proc in targets: try: if not cls._proc_alive(proc): continue proc.kill() # SIGKILL on POSIX logger.info( "Escalated to SIGKILL for pid %d (ignored SIGTERM within " "%.1fs grace)", proc.pid, grace, ) except psutil.NoSuchProcess: pass except (psutil.AccessDenied, OSError): pass # ----- Spawn ----- @staticmethod def _env_temp_dir(env: Any) -> str: """Return the writable sandbox temp dir for env-backed background tasks.""" get_temp_dir = getattr(env, "get_temp_dir", None) if callable(get_temp_dir): try: temp_dir = get_temp_dir() if isinstance(temp_dir, str) and temp_dir.startswith("/"): return temp_dir.rstrip("/") or "/" except Exception as exc: logger.debug("Could not resolve environment temp dir: %s", exc) return "/tmp" def spawn_local( self, command: str, cwd: str = None, task_id: str = "", session_key: str = "", env_vars: dict = None, use_pty: bool = False, owner_task_id: str = "", ) -> ProcessSession: """ Spawn a background process locally. Only for TERMINAL_ENV=local. Other backends use spawn_via_env(). Args: use_pty: If True, use a pseudo-terminal via ptyprocess for interactive CLI tools (Codex, Claude Code, Python REPL). Falls back to subprocess.Popen if ptyprocess is not installed. """ # Guard against the `A && B &` subshell-wait trap (issue #68915). # Bash parses ``A && B &`` as ``(A && B) &`` — a subshell that holds # the stdout pipe open forever when B is a long-running server. # The rewriter wraps it to ``A && { B & }`` so no subshell fork. # Lazy import avoids circular dependency (terminal_tool imports this). from tools.terminal_tool import _rewrite_compound_background as _rewrite_bg safe_command = _rewrite_bg(command) session = ProcessSession( id=f"proc_{uuid.uuid4().hex[:12]}", command=command, task_id=task_id, owner_task_id=owner_task_id or task_id, session_key=session_key, cwd=_resolve_safe_cwd(cwd or os.getcwd()), started_at=time.time(), ) pty_scope_attempted = False if use_pty: # Try PTY mode for interactive CLI tools try: if _IS_WINDOWS: from winpty import PtyProcess as _PtyProcessCls else: from ptyprocess import PtyProcess as _PtyProcessCls user_shell = _find_shell() pty_env = _sanitize_subprocess_env(os.environ, env_vars) pty_env["PYTHONUNBUFFERED"] = "1" # PTY mode is a real TTY, so pager-happy tools (git log/diff, # man) WILL page and hang waiting for `q` — default them to # cat, honoring any pager the user already exported. pty_env.setdefault("GIT_PAGER", "cat") pty_env.setdefault("PAGER", "cat") pty_argv = [user_shell, "-lic", f"set +m; {safe_command}"] # Cgroup isolation for PTY mode (#70716, reviewer gap #1): # Wrap the PTY command in a systemd scope so interactive # executors get their own cgroup, same as pipe mode. pty_in_supervised_gateway = ( _IS_LINUX and _is_supervised_gateway_process() ) pty_use_systemd_scope = ( pty_in_supervised_gateway and _systemd_run_user_scope_available() ) if pty_use_systemd_scope: pty_argv = _build_systemd_scope_argv( pty_argv, unit_suffix=session.id, ) session.systemd_unit = f"hermes-worker-{session.id}.scope" pty_scope_attempted = True elif pty_in_supervised_gateway: logger.debug( "PTY background executor not isolated in a " "systemd scope (systemd-run --user unavailable); " "worker shares the gateway cgroup." ) pty_proc = _PtyProcessCls.spawn( pty_argv, cwd=session.cwd, env=pty_env, dimensions=(30, 120), ) session.pid = pty_proc.pid session.host_start_time = self._safe_host_start_time(session.pid) # Store the pty handle on the session for read/write session._pty = pty_proc # PTY reader thread reader = threading.Thread( target=self._pty_reader_loop, args=(session,), daemon=True, name=f"proc-pty-reader-{session.id}", ) session._reader_thread = reader reader.start() with self._lock: self._prune_if_needed() self._running[session.id] = session self._write_checkpoint() return session except ImportError: logger.warning("ptyprocess not installed, falling back to pipe mode") except Exception as e: logger.warning("PTY spawn failed (%s), falling back to pipe mode", e) if pty_scope_attempted and session.systemd_unit: if not _stop_systemd_unit(session.systemd_unit): raise RuntimeError( "PTY scope could not be reaped; refusing pipe fallback " "to avoid duplicate command execution" ) from e session.systemd_unit = "" # Standard Popen path (non-PTY or PTY fallback) # Use the user's login shell for consistency with LocalEnvironment -- # ensures rc files are sourced and user tools are available. user_shell = _find_shell() # Force unbuffered output for Python scripts so progress is visible # during background execution (libraries like tqdm/datasets buffer when # stdout is a pipe, hiding output from process(action="poll")). bg_env = _sanitize_subprocess_env(os.environ, env_vars) bg_env["PYTHONUNBUFFERED"] = "1" _popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {} # Cgroup isolation (#70716): when running in the live, supervised # systemd gateway, wrap the worker in its own transient systemd # scope so it gets a separate cgroup. An OOM in the worker then # kills only the worker instead of taking down the whole gateway # cgroup (and the messaging control plane with it). This applies to # both pipe mode and the PTY path above. shell_argv = [user_shell, "-lic", f"set +m; {safe_command}"] in_supervised_gateway = _IS_LINUX and _is_supervised_gateway_process() use_systemd_scope = ( in_supervised_gateway and _systemd_run_user_scope_available() ) if use_systemd_scope: unit_suffix = ( f"{session.id}-pipe-fallback" if pty_scope_attempted else session.id ) spawn_argv = _build_systemd_scope_argv( shell_argv, unit_suffix=unit_suffix, ) session.systemd_unit = f"hermes-worker-{unit_suffix}.scope" # CRITICAL (#70716 regression): systemd-run --scope does NOT give # the worker a new session — the invoked process keeps the # parent's session and inherits its controlling terminal. From an # interactive TUI this drops the worker into the same session as # the foreground process group: background spawns then stop the # whole session (observed as 5 dead TUIs in state T / "Arrêté"). # start_new_session=True gives systemd-run (and the scoped worker # below it) a private session. Cgroup isolation is preserved: # the scope is attached to the invoked process, not to the # spawning session. popen_start_new_session = True else: spawn_argv = shell_argv popen_start_new_session = True if in_supervised_gateway: # Running under a supervisor but could not get a private # cgroup — the worker shares the gateway cgroup, so an OOM # in the worker can still kill the whole gateway (#70716). logger.debug( "Local background executor not isolated in a systemd scope " "(in_supervised_gateway=%s, systemd-run --user available=%s); " "worker shares the gateway cgroup.", in_supervised_gateway, _systemd_run_user_scope_available(), ) proc = subprocess.Popen( spawn_argv, text=True, cwd=session.cwd, env=bg_env, encoding="utf-8", errors="replace", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, start_new_session=popen_start_new_session, **_popen_kwargs, ) session.process = proc session.pid = proc.pid session.host_start_time = self._safe_host_start_time(session.pid) try: # Start output reader thread reader = threading.Thread( target=self._reader_loop, args=(session,), daemon=True, name=f"proc-reader-{session.id}", ) session._reader_thread = reader reader.start() with self._lock: self._prune_if_needed() self._running[session.id] = session self._write_checkpoint() except Exception: # Post-Popen setup failed — kill the orphaned subprocess (and any # descendants spawned via setsid) before re-raising so they do not # leak as untracked background processes. try: if session.systemd_unit: # The worker runs in its own systemd scope and, since the # #70716 session-isolation fix, its own session. Stop the # scope (kills every process in the worker cgroup), then # terminate the systemd-run wrapper PID as fallback. # Never killpg: scope teardown is the authoritative # cleanup for the worker cgroup. _stop_systemd_unit(session.systemd_unit) self._terminate_host_pid(proc.pid, session.host_start_time) elif not _IS_WINDOWS: try: kill_signal = getattr(signal, "SIGKILL", signal.SIGTERM) os.killpg(os.getpgid(proc.pid), kill_signal) # windows-footgun: ok - guarded by _IS_WINDOWS above except (ProcessLookupError, PermissionError, OSError): proc.kill() else: proc.kill() except Exception: pass try: proc.wait(timeout=5) except Exception: pass raise return session def spawn_via_env( self, env: Any, command: str, cwd: str = None, task_id: str = "", session_key: str = "", timeout: int = 10, owner_task_id: str = "", ) -> ProcessSession: """ Spawn a background process through a non-local environment backend. For Docker/Singularity/Modal/Daytona/SSH: runs the command inside the sandbox using the environment's execute() interface. We wrap the command to capture the in-sandbox PID and redirect output to a log file inside the sandbox, then poll the log via subsequent execute() calls. This is less capable than local spawn (no live stdout pipe, no stdin), but it ensures the command runs in the correct sandbox context. """ session = ProcessSession( id=f"proc_{uuid.uuid4().hex[:12]}", command=command, task_id=task_id, owner_task_id=owner_task_id or task_id, session_key=session_key, cwd=cwd, started_at=time.time(), env_ref=env, pid_scope="sandbox", ) # Run the command in the sandbox with output capture temp_dir = self._env_temp_dir(env) log_path = f"{temp_dir}/hermes_bg_{session.id}.log" pid_path = f"{temp_dir}/hermes_bg_{session.id}.pid" exit_path = f"{temp_dir}/hermes_bg_{session.id}.exit" quoted_command = shlex.quote(command) quoted_temp_dir = shlex.quote(temp_dir) quoted_log_path = shlex.quote(log_path) quoted_pid_path = shlex.quote(pid_path) quoted_exit_path = shlex.quote(exit_path) bg_command = ( f"mkdir -p {quoted_temp_dir} && " f"( nohup bash -lc {quoted_command} > {quoted_log_path} 2>&1; " f"rc=$?; printf '%s\\n' \"$rc\" > {quoted_exit_path} ) & " f"echo $! > {quoted_pid_path} && cat {quoted_pid_path}" ) try: result = env.execute( bg_command, timeout=timeout, rewrite_compound_background=False, ) output = result.get("output", "").strip() # Try to extract the PID from the output for line in output.splitlines(): line = line.strip() if line.isdigit(): session.pid = int(line) break # If the wrapper couldn't produce a PID (for example, syntax # error or broken redirect), treat it as a failed launch instead # of exposing a fake running session. if session.pid is None: session.exited = True session.exit_code = int(result.get("returncode", -1)) if session.exit_code == 0: session.exit_code = -1 session.completion_reason = "failed_start" session.termination_source = "failed_start" session.output_buffer = result.get("output", "").strip() except Exception as e: session.exited = True session.exit_code = -1 session.completion_reason = "failed_start" session.termination_source = "failed_start" session.output_buffer = f"Failed to start: {e}" if not session.exited: # Start a poller thread that periodically reads the log file reader = threading.Thread( target=self._env_poller_loop, args=(session, env, log_path, pid_path, exit_path), daemon=True, name=f"proc-poller-{session.id}", ) session._reader_thread = reader reader.start() with self._lock: self._prune_if_needed() if not session.exited: self._running[session.id] = session if not session.exited: self._write_checkpoint() return session # ----- Reader / Poller Threads ----- def _reader_loop(self, session: ProcessSession): """Background thread: read stdout from a local Popen process. IMPORTANT: avoid ``TextIOWrapper.read(4096)`` here. On pipes that call can block until EOF (or a large buffer fills), which makes "live" output land in one burst at process exit. ``buffer.read1(4096)`` yields incremental chunks as bytes become available, then we decode to text. Orphaned-pipe guard (issue #68915): when the user's command backgrounds a long-lived process (``node server.js &``, ``sleep 300 &``), that grandchild inherits the write end of our stdout pipe via ``fork()``. The direct ``bash`` child exits promptly, but the pipe never reaches EOF while the grandchild lives — so a blocking read would park this thread forever, ``session.exited`` would never flip, and ``notify_on_complete`` would never fire (``_reconcile_local_exit`` only runs lazily from poll()/wait(), which an autonomous notification can't rely on). On POSIX we therefore ``select()`` with a short poll interval and stop draining shortly after the direct child exits, even if the pipe hasn't EOF'd — mirroring the foreground fix in ``tools/environments/base.py::_wait_for_process`` (#8340). Windows pipes don't support select(); the blocking path is kept there and the lazy reconcile in poll()/wait() remains the safety net. """ first_chunk = True # Incremental decoder: raw pipe reads can split a multibyte UTF-8 # character across two read1() chunks. A stateless per-chunk # ``bytes.decode(errors="replace")`` turns both halves into U+FFFD # mojibake. The incremental decoder holds the partial sequence until # the continuation bytes arrive — same treatment the foreground path # already has in ``tools/environments/base.py::_wait_for_process``. # (Ported from openclaw/openclaw#112325.) decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") def _append_chunk(chunk: str): nonlocal first_chunk if first_chunk: chunk = self._clean_shell_noise(chunk) first_chunk = False with session._lock: session.output_buffer += chunk if len(session.output_buffer) > session.max_output_chars: session.output_buffer = session.output_buffer[-session.max_output_chars:] self._check_watch_patterns(session, chunk) self._emit_output(session, chunk) try: proc = session.process if proc is None or proc.stdout is None: return stdout = proc.stdout raw_read = getattr(getattr(stdout, "buffer", None), "read1", None) # Resolve a real OS fd for the select() path. Mocked streams # (unit tests, adapters) may lack fileno() — fall back to the # historical blocking loop for those. fd = None if raw_read is not None and not _IS_WINDOWS: fileno = getattr(stdout, "fileno", None) try: candidate = fileno() if callable(fileno) else None except Exception: candidate = None if isinstance(candidate, int) and candidate >= 0: fd = candidate if fd is not None: import select as _select idle_after_exit = 0 while True: try: ready, _, _ = _select.select([fd], [], [], 0.2) except (ValueError, OSError): break # fd already closed if ready: raw = raw_read(4096) if not raw: break # true EOF — all writers closed chunk = decoder.decode(raw) if chunk: _append_chunk(chunk) idle_after_exit = 0 elif proc.poll() is not None: # Direct child is gone and the pipe was idle for # ~200ms. Give it a few more cycles to catch any # buffered tail, then stop — otherwise we would wait # forever on a pipe held open by an orphaned # grandchild (issue #68915). idle_after_exit += 1 if idle_after_exit >= 3: break else: while True: if raw_read is not None: raw = raw_read(4096) if not raw: break chunk = decoder.decode(raw) if not chunk: continue # partial multibyte sequence — wait for more bytes else: # Fallback for mocked/alternate streams without a buffered raw # interface. This may be less "live", but keeps compatibility. chunk = stdout.read(4096) if not chunk: break _append_chunk(chunk) except Exception as e: logger.debug("Process stdout reader ended: %s", e) finally: # Flush any bytes still pending in the incremental decoder (a # truncated multibyte sequence at EOF becomes one U+FFFD instead # of being dropped silently). try: tail = decoder.decode(b"", final=True) if tail: _append_chunk(tail) except Exception: pass # Always reap the child to prevent zombie processes. try: session.process.wait(timeout=5) except Exception as e: logger.debug("Process wait timed out or failed: %s", e) session.exited = True if session.completion_reason != "killed": session.exit_code = session.process.returncode session.completion_reason = "exited" self._move_to_finished(session) @staticmethod def _log_delta_command(quoted_log_path: str, offset: int) -> str: """Build the shell command that reads only new bytes from a log file. The old version ran ``cat`` on the whole file every poll, so a job that keeps writing pays for its entire output again and again. Over a long run that turns into a lot of wasted traffic on the docker/SSH channel, since only the new part is ever used. The command prints one header line, ``" "``, then the bytes between ``offset`` and ``size``. Reading the size first and cutting the tail at that same size keeps the two numbers in step, so a file that grows while the command runs never sends a byte twice. A file that shrank was rotated or truncated, so the offset drops back to 0 and the reader starts over. The end of the window is pulled back to a UTF-8 character boundary: the backend decodes each ``execute()`` result on its own, so a multibyte character straddling two polls would otherwise come back as replacement characters (and break watch patterns near the seam). Up to 3 trailing continuation bytes are held for the next poll; the header reports the trimmed size so the offset stays consistent. """ return ( f"O={offset}; " f"S=$({{ wc -c < {quoted_log_path}; }} 2>/dev/null | tr -dc '0-9'); " f"S=${{S:-0}}; " f'if [ "$S" -lt "$O" ]; then O=0; fi; ' # Hold back an INCOMPLETE trailing UTF-8 sequence for the next # poll. Scan back up to 3 continuation bytes (octal 200-277) to # the lead byte; if the lead byte's declared length (3xx=2, 34x-35x # =3, 36x-37x=4) exceeds the bytes present, trim to before it. # Complete sequences and ASCII tails are left untouched. f'N=0; P=$S; while [ "$P" -gt "$O" ] && [ "$N" -lt 3 ]; do ' f"B=$(tail -c +$P {quoted_log_path} 2>/dev/null | head -c 1 | od -An -to1 | tr -dc '0-9'); " f'case "$B" in 2[0-7][0-7]) P=$((P-1)); N=$((N+1));; *) break;; esac; done; ' f'if [ "$N" -gt 0 ] || [ "$P" -eq "$S" ]; then ' f"B=$(tail -c +$P {quoted_log_path} 2>/dev/null | head -c 1 | od -An -to1 | tr -dc '0-9'); " f'case "$B" in 3[0-3][0-7]) L=2;; 3[4-5][0-7]) L=3;; 3[6-7][0-7]) L=4;; *) L=1;; esac; ' f'if [ "$L" -gt $((N+1)) ]; then S=$((P-1)); fi; fi; ' f'echo "$S $O"; ' f'if [ "$S" -gt "$O" ]; then ' f"tail -c +$((O+1)) {quoted_log_path} 2>/dev/null | head -c $((S-O)); fi" ) def _env_poller_loop( self, session: ProcessSession, env: Any, log_path: str, pid_path: str, exit_path: str ): """Background thread: poll a sandbox log file for non-local backends.""" quoted_log_path = shlex.quote(log_path) quoted_pid_path = shlex.quote(pid_path) quoted_exit_path = shlex.quote(exit_path) # Byte offset already read from the log. Bytes, not characters: the # shell counts bytes, and a log with non-ASCII text has more bytes # than characters. prev_output_bytes = 0 while not session.exited: time.sleep(2) # Poll every 2 seconds try: # Read only the bytes written since the last poll. result = env.execute( self._log_delta_command(quoted_log_path, prev_output_bytes), timeout=10, ) raw = result.get("output", "") header, _, delta = raw.partition("\n") try: size_str, offset_str = header.split() new_size = int(size_str) used_offset = int(offset_str) except ValueError: # No usable header (command failed, shell missing a tool). # Skip this poll rather than act on a half-read value. new_size = None used_offset = None delta = "" if new_size is not None: if used_offset < prev_output_bytes: # The log was rotated or truncated, so what we hold no # longer lines up with the file. Drop it and restart. with session._lock: session.output_buffer = "" prev_output_bytes = new_size if delta: with session._lock: session.output_buffer += delta if len(session.output_buffer) > session.max_output_chars: session.output_buffer = session.output_buffer[-session.max_output_chars:] self._check_watch_patterns(session, delta) self._emit_output(session, delta) # Check if process is still running check = env.execute( f"kill -0 \"$(cat {quoted_pid_path} 2>/dev/null)\" 2>/dev/null; echo $?", timeout=5, ) check_output = check.get("output", "").strip() if check_output and check_output.splitlines()[-1].strip() != "0": # Process has exited -- get exit code captured by the wrapper shell. exit_result = env.execute( f"cat {quoted_exit_path} 2>/dev/null", timeout=5, ) exit_str = exit_result.get("output", "").strip() try: session.exit_code = int(exit_str.splitlines()[-1].strip()) except (ValueError, IndexError): session.exit_code = -1 session.exited = True if session.completion_reason != "killed": session.completion_reason = "exited" self._move_to_finished(session) return except Exception: # Environment might be gone (sandbox reaped, etc.) session.exited = True session.exit_code = -1 session.completion_reason = "lost" session.termination_source = "backend_lost" self._move_to_finished(session) return def _pty_reader_loop(self, session: ProcessSession): """Background thread: read output from a PTY process.""" pty = session._pty # PTY reads can split a multibyte UTF-8 character across chunks just # like pipe reads — hold partial sequences until the rest arrives. # (Ported from openclaw/openclaw#112325.) decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") def _append_text(text: str): with session._lock: session.output_buffer += text if len(session.output_buffer) > session.max_output_chars: session.output_buffer = session.output_buffer[-session.max_output_chars:] self._check_watch_patterns(session, text) self._emit_output(session, text) try: while pty.isalive(): try: chunk = pty.read(4096) if chunk: # ptyprocess returns bytes; pywinpty returns str text = chunk if isinstance(chunk, str) else decoder.decode(chunk) if text: _append_text(text) except EOFError: break except Exception: break except Exception as e: logger.debug("PTY stdout reader ended: %s", e) # Flush any partial multibyte sequence held by the decoder. try: tail = decoder.decode(b"", final=True) if tail: _append_text(tail) except Exception: pass # Process exited try: pty.wait() except Exception as e: logger.debug("PTY wait timed out or failed: %s", e) session.exited = True if session.completion_reason != "killed": session.exit_code = pty.exitstatus if hasattr(pty, 'exitstatus') else -1 session.completion_reason = "exited" self._move_to_finished(session) def _move_to_finished(self, session: ProcessSession): """Move a session from running to finished. Idempotent: if the session was already moved (e.g. kill_process raced with the reader thread), the second call is a no-op — no duplicate completion notification is enqueued. """ with self._lock: was_running = self._running.pop(session.id, None) is not None self._finished[session.id] = session session._completion_event.set() self._write_checkpoint() # Only enqueue completion notification on the FIRST move. Without # this guard, kill_process() and the reader thread can both call # _move_to_finished(), producing duplicate [IMPORTANT: ...] messages. if was_running and session.notify_on_complete: from tools.ansi_strip import strip_ansi output_tail = strip_ansi(session.output_buffer[-2000:]) if session.output_buffer else "" notification = { "type": "completion", "session_id": session.id, "session_key": session.session_key, "task_id": session.task_id, "owner_task_id": session.owner_task_id or session.task_id, "command": session.command, "exit_code": session.exit_code, "completion_reason": session.completion_reason, "termination_source": session.termination_source, "output": output_tail, # Stable producer identity across checkpoint recovery; unlike # a consumer-observed completion timestamp, this does not vary # based on which watcher notices exit first. "started_at": session.started_at, } _redact_process_result(notification) self.completion_queue.put(notification) # ----- Query Methods ----- def is_completion_consumed(self, session_id: str) -> bool: """Check if a completion notification was already consumed via wait/log.""" return session_id in self._completion_consumed def is_session_waiting(self, session_id: str) -> bool: """Whether a goal loop parked on this session should still be parked. Used by the goal-loop wait barrier (``hermes_cli.goals``) to support waiting on a process's OWN trigger, not just its exit. A session is "still waiting" when: - it is still running, AND - if it has ``watch_patterns``, none has matched yet (so a long-lived watcher that fires a trigger mid-run — and may never exit — unblocks the moment its pattern hits, not on exit). Returns False (don't wait) when the session has exited, its watch pattern has already fired, or the session is unknown — so a stale or already-triggered barrier can never wedge the loop. """ if not session_id: return False with self._lock: session = self._running.get(session_id) or self._finished.get(session_id) if session is None: return False # Refresh detached/remote state so .exited is current. try: self._refresh_detached_session(session) except Exception: pass if session.exited: return False # Watch-pattern process: the trigger is a pattern match, not exit. # Once any match has been delivered, the wait is satisfied even though # the process keeps running (server/daemon/watcher case). if session.watch_patterns and not session._watch_disabled: if session._watch_hits > 0: return False return True def wait_for_pending_completions( self, task_id: Optional[str] = None, *, timeout: float | None = None, poll_interval: float = 1.0, ) -> dict: """Bounded wait for tracked ``notify_on_complete`` background processes. One-shot CLI runs (``hermes -q/-Q/-z``) exit as soon as their single turn ends. Any background process the turn spawned with ``notify_on_complete=True`` — a bounded task whose completion the caller explicitly cares about — still holds a stdout pipe owned by the dying parent, so it is killed by SIGPIPE on its next write a few seconds later. Bot Mode handoff REPLIES are the visible casualty (#90879): a recipient invoked as ``hermes -p chat -Q --query-file ...`` dispatches its reply via ``message_agent`` / ``bot_relay`` exactly this way, then exits, and the reply process is destroyed ~3s later. The sender waits forever for a reply that was already killed. Called from the one-shot exit paths so the parent lingers (bounded) until those deliveries actually finish. This fixes the class — ANY bounded background task in a one-shot run, not just DMs: bot_mode_dm deliveries, bot_relay waiter processes, and plain ``terminal(background=true, notify_on_complete=true)`` jobs. Only ``notify_on_complete`` processes are waited on. Plain background processes (servers, daemons, watch-pattern monitors) carry no completion contract and are not the parent's to wait for. Args: task_id: restrict to processes spawned for this task; ``None`` waits on every tracked process (a one-shot CLI process hosts exactly one agent, so its registry is private to that run). timeout: max seconds to linger. ``None`` reads ``terminal.oneshot_completion_wait_seconds`` from config (default 600). ``<= 0`` disables the wait entirely. poll_interval: per-pass event-wait bound; each pass re-reconciles child state so an orphaned-pipe exit (#17327) can't wedge the linger for the full timeout. Returns: ``{"waited": [...], "completed": [...], "timed_out": [...]}`` (session ids). All lists empty when there was nothing to wait on. """ if timeout is None: timeout = self._oneshot_completion_wait_seconds() result: dict = {"waited": [], "completed": [], "timed_out": []} with self._lock: pending = [ s for s in self._running.values() if s.notify_on_complete and not s.exited and (task_id is None or s.task_id == task_id) ] if not pending or timeout <= 0: return result result["waited"] = [s.id for s in pending] logger.info( "One-shot exit lingering (bounded %ss) for %d notify_on_complete " "background process(es): %s", timeout, len(pending), ", ".join(s.id for s in pending), ) deadline = time.monotonic() + max(float(timeout), 0.0) interval = max(float(poll_interval), 0.05) try: from tools.interrupt import is_interrupted as _is_interrupted except Exception: def _is_interrupted() -> bool: return False interrupted = False for session in pending: try: while not session.exited: if interrupted or _is_interrupted(): interrupted = True break remaining = deadline - time.monotonic() if remaining <= 0: break # Reconcile first: catches direct-child exits whose reader # is blocked on a pipe held open by a descendant (#17327) # and detached/env sessions, so the event actually fires. try: self._reconcile_local_exit(session) self._refresh_detached_session(session) except Exception: pass if session.exited: break session._completion_event.wait(min(remaining, interval)) except KeyboardInterrupt: # User aborted the linger — stop waiting on everything but # never let the interrupt skip the caller's durable teardown # (session flush, end_session) that follows this wait. interrupted = True if session.exited: result["completed"].append(session.id) else: result["timed_out"].append(session.id) if result["timed_out"]: logger.warning( "One-shot exit linger timed out after %ss with %d background " "process(es) still running: %s — they may be killed when this " "process exits.", timeout, len(result["timed_out"]), ", ".join(result["timed_out"]), ) return result @staticmethod def _oneshot_completion_wait_seconds() -> float: """Bounded linger (s) for one-shot exits with pending notify_on_complete processes. Read from ``terminal.oneshot_completion_wait_seconds``; 0 disables. Falls back to the DEFAULT_CONFIG value (600) when config is unreadable so callers always get a sane bound. """ try: from hermes_cli.config import DEFAULT_CONFIG, cfg_get, read_raw_config cfg = read_raw_config() val = cfg_get(cfg, "terminal", "oneshot_completion_wait_seconds") if val is None: val = DEFAULT_CONFIG["terminal"]["oneshot_completion_wait_seconds"] return max(float(val), 0.0) except Exception: return 600.0 def _drain_should_skip( self, session_id: str, *, skip_poll_observed: bool = True ) -> bool: """Whether this drain should skip a completion event for this session. Skips when the agent has either truly consumed the output (wait/log → ``_completion_consumed``) or observed the exit inline via poll() (``_poll_observed``). In both cases the CLI agent already has the result this turn, so injecting a [SYSTEM: ...] completion would be a duplicate (#8228). The gateway/tui watchers do NOT use this — they check only ``is_completion_consumed`` so a read-only poll never suppresses their autonomous delivery turn (#10156). """ return session_id in self._completion_consumed or ( skip_poll_observed and session_id in self._poll_observed ) @staticmethod def _surface_child_process_notifications() -> bool: """Whether subagent-owned process notifications surface in the parent. Read from ``delegation.surface_child_process_notifications`` in config.yaml (default false = suppress). On any config read error the DEFAULT applies (suppress) — never crash the drain loop. """ try: from hermes_cli.config import DEFAULT_CONFIG, cfg_get, read_raw_config cfg = read_raw_config() val = cfg_get(cfg, "delegation", "surface_child_process_notifications") if val is None: val = DEFAULT_CONFIG["delegation"][ "surface_child_process_notifications" ] return bool(val) except Exception: return False def drain_notifications( self, session_key: str = "", owns_event=None, *, skip_poll_observed: bool = True, ) -> "list[tuple[dict, str]]": """Pop all pending notification events and return formatted pairs. Returns a list of (raw_event, formatted_text) tuples. Skips completion events the agent already consumed via wait/log or observed inline via poll() (see ``_drain_should_skip``). Gateway/TUI callers pass ``skip_poll_observed=False`` because read-only polling must not suppress autonomous delivery there. When a routing filter is supplied, addressed notifications must not be drained into the wrong session. Async-delegation events always require conversation payload; ordinary notifications require routing when they carry ``session_key`` or ``origin_ui_session_id`` metadata. Two filter modes are supported, strongest first: - ``owns_event(evt) -> bool``: positive-proof ownership callback. When provided, a routed event is consumed ONLY if the callback returns True; everything else is re-queued for its owner. The TUI passes its compression-chain-aware ownership check here so a post-compression session still claims its own pre-compression dispatches. - ``session_key``: plain key equality (CLI and other single-session callers). Non-matching addressed events are re-queued. With neither set, all events are consumed (legacy single-session behavior, backward compatible). Ownerless ordinary notifications also retain that legacy behavior even when a filter is provided. When a filter is provided, ownerless async-delegation events remain fail-closed and require positive proof. """ results: "list[tuple[dict, str]]" = [] requeue: "list[dict]" = [] # Lazily-read flag for subagent-owned process notifications # (delegation.surface_child_process_notifications, default false). # Read at most once per drain, and only when an sa- event shows up. surface_child: "bool | None" = None while not self.completion_queue.empty(): try: evt = self.completion_queue.get_nowait() except Exception: break # Positive-proof ownership beats bare key equality. Delegation # payloads always require proof; ordinary events require it once # they carry routing metadata. Ownerless ordinary events preserve # legacy single-session delivery. is_async_delegation = evt.get("type") == "async_delegation" evt_session_key = str(evt.get("session_key") or "") evt_origin_sid = str(evt.get("origin_ui_session_id") or "") requires_positive_proof = is_async_delegation or bool( evt_session_key or evt_origin_sid ) if owns_event is not None and requires_positive_proof: try: owned = bool(owns_event(evt)) except Exception: owned = False # fail closed — never leak on a broken check if not owned: requeue.append(evt) continue elif session_key and requires_positive_proof: if evt_session_key != session_key: requeue.append(evt) continue elif is_async_delegation and evt.get("restored"): # Durable restore can enqueue previous-process payloads into a # fresh registry. An unfiltered legacy drain cannot prove # ownership, so leave those events queued for the owner. requeue.append(evt) continue # Local consumed/observed state may suppress only events this # session owns (or legacy ownerless ordinary events). Routing must # happen first so a foreign session cannot drop the owner's event. _evt_sid = evt.get("session_id", "") if evt.get("type") == "completion" and self._drain_should_skip( _evt_sid, skip_poll_observed=skip_poll_observed ): continue # Subagent-owned process notifications are suppressed from the # parent conversation by default — the child's consolidated # delegation result is the deliverable; "npm ci finished" walls # mid-chat are noise. Ownership is judged on owner_task_id (the # RAW spawning task id): the container key in task_id is # deliberately collapsed to "default"/the session key by # _resolve_container_task_id, which previously let child events # bypass this gate. Dropped, NOT requeued (children never drain # notify events, so requeueing would pin them in the queue # forever). Type 'async_delegation' is the delegation result # itself and is NEVER suppressed. _evt_task_id = str( evt.get("owner_task_id") or evt.get("task_id") or "" ) if not is_async_delegation and _evt_task_id.startswith("sa-"): if surface_child is None: surface_child = self._surface_child_process_notifications() if not surface_child: logger.debug( "Suppressed subagent-owned process notification " "(delegation.surface_child_process_notifications=false): " "type=%s session_id=%s task_id=%s", evt.get("type", "completion"), _evt_sid, _evt_task_id, ) continue text = format_process_notification(evt) if text: results.append((evt, text)) for evt in requeue: self.completion_queue.put(evt) return results # Minimum characters of the random suffix required for prefix resolution. # Short prefixes ("p", "pr", "proc_1") are too collision-prone to act on. _MIN_PREFIX_CHARS = 4 def get(self, session_id: str) -> Optional[ProcessSession]: """Get a session by ID (running or finished). Accepts either the full ID or a unique ID prefix (inspired by Factory Droid's task-ID prefixes, and the same UX as ``git``/``docker`` short hashes): ``proc_4dae`` — or just the bare suffix ``4dae`` — resolves to ``proc_4dae56ca81f6`` when exactly one session matches. Ambiguous or too-short prefixes resolve to None (callers already report "No process with ID ..."), never to an arbitrary pick. """ with self._lock: session = self._running.get(session_id) or self._finished.get(session_id) if session is None: session = self._resolve_prefix(session_id) return self._refresh_detached_session(session) def _resolve_prefix(self, session_id: str) -> Optional[ProcessSession]: """Resolve a unique session-ID prefix to its session, else None. Exact lookups happen in :meth:`get` before this runs, so a full ID never pays the scan. Matching is prefix-only (no substring) and requires a unique hit; a bare suffix without the ``proc_`` lead is normalized so users can paste just the hex tail. """ if not session_id or not isinstance(session_id, str): return None query = session_id.strip() if not query: return None # Allow the bare suffix form: "4dae56" -> "proc_4dae56". if not query.startswith("proc_"): query = f"proc_{query}" suffix = query[len("proc_"):] if len(suffix) < self._MIN_PREFIX_CHARS: return None with self._lock: matches = [ s for store in (self._running, self._finished) for sid, s in store.items() if sid.startswith(query) ] if len(matches) == 1: return matches[0] return None def _reconcile_local_exit(self, session: "ProcessSession") -> None: """Reconcile session.exited against the real child process state. The reader thread (`_reader_loop`) sets `session.exited = True` only in its `finally` block, which runs when `stdout.read()` returns EOF. If the direct `Popen` child has exited but a descendant process (e.g. a daemon spawned by `hermes update` restarting the gateway) is still holding the stdout pipe open, the reader blocks forever and poll() keeps returning "running" indefinitely (issue #17327 — 74 polls over 7 minutes on Feishu). This helper closes that window: when `session.exited` is still False but the direct child's `Popen.poll()` reports an exit code, drain any readable bytes non-blocking and flip `session.exited`. The orphaned reader thread remains stuck on its blocking `read()` but is a daemon thread and will be reaped with the process. Safe no-op on sessions without a local `Popen` (env/PTY), already- exited sessions, and detached-recovered sessions. """ if session is None or session.exited: return proc = getattr(session, "process", None) if proc is None: return try: rc = proc.poll() except Exception: return if rc is None: return # Direct child still running — reader block is legitimate. # Direct child exited. Try to drain any bytes the reader hasn't # consumed yet. This is best-effort: if the pipe is held open by a # descendant, the non-blocking read returns what's immediately # available and we stop. drained = "" stdout = getattr(proc, "stdout", None) if stdout is not None and not _IS_WINDOWS: try: import fcntl fd = stdout.fileno() flags = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) try: chunk = stdout.read() if chunk: drained = chunk if isinstance(chunk, str) else chunk.decode("utf-8", errors="replace") except (BlockingIOError, OSError, ValueError): pass finally: try: fcntl.fcntl(fd, fcntl.F_SETFL, flags) except Exception: pass except Exception as e: logger.debug("Non-blocking drain failed for %s: %s", session.id, e) with session._lock: if drained: session.output_buffer += drained if len(session.output_buffer) > session.max_output_chars: session.output_buffer = session.output_buffer[-session.max_output_chars:] session.exited = True if session.completion_reason != "killed": session.exit_code = rc session.completion_reason = "exited" logger.info( "Reconciled session %s: direct child exited with code %s but reader " "was still blocked (orphaned pipe). Flipped to exited.", session.id, rc, ) self._move_to_finished(session) def poll(self, session_id: str) -> dict: """Check status and get new output for a background process.""" from tools.ansi_strip import strip_ansi session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} # Reconcile against real child state before reading session.exited. # Guards against orphaned-pipe reader hangs (issue #17327). self._reconcile_local_exit(session) with session._lock: output_preview = strip_ansi(session.output_buffer[-1000:]) if session.output_buffer else "" result = { "session_id": session.id, "command": session.command, "status": "exited" if session.exited else "running", "pid": session.pid, "uptime_seconds": int(time.time() - session.started_at), "output_preview": output_preview, } if session.exited: result["exit_code"] = session.exit_code result["completion_reason"] = session.completion_reason result["termination_source"] = session.termination_source # NOTE: poll() is a read-only status query and deliberately does # NOT mark the session _completion_consumed. wait()/read_log() # represent actual output consumption and do mark it. Marking # consumed here would let a status check silently suppress the # notify_on_complete watcher's autonomous delivery turn (#10156). # # We DO record it in _poll_observed so the CLI's inline drain still # dedups (the agent already saw the exit in this turn's poll result) # without affecting the gateway/tui watchers, which only consult # _completion_consumed. self._poll_observed.add(session_id) if session.detached: result["detached"] = True result["note"] = "Process recovered after restart -- output history unavailable" return result def read_log(self, session_id: str, offset: int | None = None, limit: int = 200) -> dict: """Read the full output log with optional pagination by lines.""" from tools.ansi_strip import strip_ansi session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} with session._lock: full_output = strip_ansi(session.output_buffer) lines = full_output.splitlines() total_lines = len(lines) # Default (offset=None): last N lines. An explicit offset=0 means # "start from the first line" — previously it was conflated with # the default and silently returned the TAIL instead of the head # (same falsy-coercion class as the wait() timeout guard; salvaged # from PR #60004, credit @isheng-eqi). if offset is None and limit > 0: selected = lines[-limit:] observed_completion_output = bool(selected) or total_lines == 0 else: offset = offset or 0 selected = lines[offset:offset + limit] stop = slice(offset, offset + limit).indices(total_lines)[1] observed_completion_output = ( total_lines == 0 or (bool(selected) and stop == total_lines) ) result = { "session_id": session.id, "command": session.command, "status": "exited" if session.exited else "running", "output": "\n".join(selected), "total_lines": total_lines, "showing": f"{len(selected)} lines", } if session.exited and observed_completion_output: self._completion_consumed.add(session_id) return result def wait(self, session_id: str, timeout: int = None) -> dict: """ Block until a process exits, timeout, or interrupt. Args: session_id: The process to wait for. timeout: Max seconds to block. Falls back to TERMINAL_TIMEOUT config. Returns: dict with status ("exited", "timeout", "interrupted", "not_found") and output snapshot. """ from tools.ansi_strip import strip_ansi from tools.interrupt import is_interrupted as _is_interrupted try: default_timeout = int(os.getenv("TERMINAL_TIMEOUT", "180")) except (ValueError, TypeError): default_timeout = 180 max_timeout = default_timeout requested_timeout = timeout timeout_note = None # Reject non-positive timeouts — the schema declares minimum=1, but # not every caller enforces schemas before dispatch. timeout=0 is # falsy, so without this guard it silently fell through # (`0 or max_timeout`) to the DEFAULT wait instead of erroring. # Salvaged from PR #60004 (credit @isheng-eqi). if requested_timeout is not None and requested_timeout <= 0: return { "status": "error", "error": f"timeout must be positive (got {requested_timeout})", } if requested_timeout and requested_timeout > max_timeout: effective_timeout = max_timeout timeout_note = ( f"Requested wait of {requested_timeout}s was clamped " f"to configured limit of {max_timeout}s" ) else: effective_timeout = requested_timeout or max_timeout session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} deadline = time.monotonic() + effective_timeout while time.monotonic() < deadline: session = self._refresh_detached_session(session) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} # Reconcile against real child state — guards against orphaned- # pipe reader hangs where the reader is blocked but the direct # child has already exited (issue #17327). self._reconcile_local_exit(session) if session.exited: self._completion_consumed.add(session_id) result = { "status": "exited", "command": session.command, "exit_code": session.exit_code, "completion_reason": session.completion_reason, "termination_source": session.termination_source, "output": strip_ansi(session.output_buffer[-2000:]), } if timeout_note: result["timeout_note"] = timeout_note return result if _is_interrupted(): result = { "status": "interrupted", "command": session.command, "output": strip_ansi(session.output_buffer[-1000:]), "note": "User sent a new message -- wait interrupted", } if timeout_note: result["timeout_note"] = timeout_note return result remaining = deadline - time.monotonic() if remaining <= 0: break session._completion_event.wait(timeout=min(1.0, remaining)) result = { "status": "timeout", "command": session.command, "output": strip_ansi(session.output_buffer[-1000:]), # A wait window elapsing is NOT a failure — 511 exact-duplicate # process calls in a production window show models re-issuing # identical waits after misreading this result as an error. "process_running": True, } uptime = time.time() - session.started_at if session.started_at else None base_note = ( f"Wait window of {effective_timeout}s elapsed — the process is " "still running. This is not an error." ) if uptime is not None: base_note += f" Uptime: {int(uptime)}s." if session.notify_on_complete: base_note += ( " notify_on_complete is set: you will be notified on exit — " "do more work instead of waiting again." ) else: base_note += ( " Poll again later or use terminal(background=true, " "notify_on_complete=true) next time for automatic notification." ) if timeout_note: result["timeout_note"] = f"{timeout_note}. {base_note}" else: result["timeout_note"] = base_note return result def kill_process( self, session_id: str, *, source: str = "process.kill", consume_output: bool = True, ) -> dict: """Kill a background process and return its output snapshot. ``consume_output`` is true for explicit tool/RPC kills because their caller observes the returned output. Bulk cleanup passes false: it discards each result and therefore must not suppress an autonomous output-bearing completion notification. Exception: abandoned-turn reaping (``kill_started_since``) is bulk cleanup that deliberately passes true — a killed abandoned process must not enqueue a synthetic follow-up that revives work the timeout/interrupt stopped. """ from tools.ansi_strip import strip_ansi session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} if session.exited: # Even if the main process already exited, a double-forked # descendant may still be alive in the systemd scope (#70716, # reviewer gap #2 — the ``already_exited`` early return skipped # unit cleanup). Stop the scope to reap any survivors. if session.systemd_unit: _stop_systemd_unit(session.systemd_unit) with session._lock: result = { "status": "already_exited", "command": session.command, "exit_code": session.exit_code, "completion_reason": session.completion_reason, "termination_source": session.termination_source, "output": strip_ansi(session.output_buffer[-2000:]), } # Only suppress the autonomous turn after its output is present in # the explicit kill result, matching wait/log consumption. if consume_output: self._completion_consumed.add(session_id) return result # Kill via PTY, Popen (local), or env execute (non-local) try: if session._pty: # PTY process -- terminate via ptyprocess try: session._pty.terminate(force=True) except Exception: if session.pid: os.kill(session.pid, signal.SIGTERM) elif session.process: # Local process -- kill the process tree. On Windows this # must be taskkill /T /F; Popen.terminate() only kills the # shell wrapper and leaves Git Bash descendants behind. self._terminate_host_pid(session.process.pid, session.host_start_time) elif session.env_ref and session.pid: # Non-local -- kill inside sandbox session.env_ref.execute(f"kill {session.pid} 2>/dev/null", timeout=5) elif session.detached and session.pid_scope == "host" and session.pid: # Identity check, not bare liveness: if the PID is gone OR was # recycled onto an unrelated process, treat our process as # exited and never tree-kill the stranger. If this recovered # session also carries an owned systemd scope, stop that scope # before returning: a daemonized descendant may still be alive # there even though the wrapper PID exited or was recycled # across the gateway restart (#70716, teknium1 review). if not self._host_pid_is_ours(session.pid, session.host_start_time): if session.systemd_unit: _stop_systemd_unit(session.systemd_unit) with session._lock: session.exited = True session.exit_code = None output = strip_ansi(session.output_buffer[-2000:]) if consume_output: self._completion_consumed.add(session_id) self._move_to_finished(session) return { "status": "already_exited", "exit_code": session.exit_code, "output": output, } self._terminate_host_pid(session.pid, session.host_start_time) else: return { "status": "error", "error": ( "Recovered process cannot be killed after restart because " "its original runtime handle is no longer available" ), } # If the worker was spawned in its own systemd scope (#70716), # stop the entire unit to reap any double-forked descendants that # were reparented inside the scope and survived the PID signal # above (reviewer gap #2). ``systemctl --user stop`` sends # SIGTERM to every process in the cgroup and escalates to SIGKILL # after TimeoutStopSec. This is additive — the PID-based kill # above already handled the main process; this catches stragglers. if session.systemd_unit: _stop_systemd_unit(session.systemd_unit) # Capture output before marking consumed, then mark consumed before # exposing ``exited`` to watcher tasks. This closes the delayed # notification race without discarding the terminal transcript. with session._lock: output = strip_ansi(session.output_buffer[-2000:]) if consume_output: self._completion_consumed.add(session_id) session.exited = True session.exit_code = -15 # SIGTERM session.completion_reason = "killed" session.termination_source = source self._move_to_finished(session) self._write_checkpoint() return { "status": "killed", "session_id": session.id, "completion_reason": session.completion_reason, "termination_source": session.termination_source, "output": output, } except Exception as e: return {"status": "error", "error": str(e)} def write_stdin(self, session_id: str, data: str) -> dict: """Send raw data to a running process's stdin (no newline appended).""" session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} if session.exited: return {"status": "already_exited", "error": "Process has already finished"} # PTY mode -- write through pty handle. if hasattr(session, '_pty') and session._pty: try: # pywinpty expects str on Windows; ptyprocess expects bytes on POSIX. if _IS_WINDOWS: pty_data = data.decode("utf-8") if isinstance(data, bytes) else str(data) else: # surrogateescape: a PTY is a byte stream — round-trip the # original bytes instead of crashing on surrogate content. pty_data = data.encode("utf-8", "surrogateescape") if isinstance(data, str) else data session._pty.write(pty_data) return {"status": "ok", "bytes_written": len(data)} except Exception as e: return {"status": "error", "error": str(e)} # Popen mode -- write through stdin pipe if not session.process or not session.process.stdin: return {"status": "error", "error": "Process stdin not available (non-local backend or stdin closed)"} try: session.process.stdin.write(data) session.process.stdin.flush() return {"status": "ok", "bytes_written": len(data)} except Exception as e: return {"status": "error", "error": str(e)} def submit_stdin(self, session_id: str, data: str = "") -> dict: """Send data + newline to a running process's stdin (like pressing Enter). On a Windows PTY session the Enter key is a carriage return: ConPTY cooked input treats ``\\r`` as end-of-line, and a bare ``\\n`` written through pywinpty is NOT delivered as a line terminator — the child's blocking line read (Python ``readline()``, Go ``bufio.Scanner`` as in ``gh auth login``'s "Press Enter to open the browser" prompt) simply never returns and the process hangs while looking healthy. Verified empirically via pywinpty 2.0.15: ``\\n`` -> no line, ``\\r`` / ``\\r\\n`` -> line delivered. Use ``\\r\\n`` so the child sees both the Enter keypress and a conventional newline; POSIX PTYs and Popen pipes keep the plain ``\\n``. """ session = self.get(session_id) is_windows_pty = bool( _IS_WINDOWS and session is not None and getattr(session, "_pty", None) ) line_ending = "\r\n" if is_windows_pty else "\n" return self.write_stdin(session_id, data + line_ending) def request_close_terminal(self, session_id: str) -> dict: """Ask the desktop GUI to close the read-only terminal tab mirroring this background process. This does NOT kill the process — it only drops the view. Output keeps streaming into the (capped) buffer and the user can reopen the tab from the status stack. Desktop-only: returns an error if no UI close sink is wired (e.g. CLI / messaging).""" sink = self.on_close if sink is None: return { "status": "error", "error": "close_terminal is only available in the Hermes desktop app.", } # The session may already be finished (or pruned) — the tab can still # linger and be closed, so a missing session is not an error here. session = self.get(session_id) try: sink(session, session_id) except Exception as e: return {"status": "error", "error": str(e)} return { "status": "ok", "closed": session_id, "note": ( "Closed the read-only terminal tab. The process was not killed; " "its output remains available and the user can reopen the tab " "from the status stack." ), } def close_stdin(self, session_id: str) -> dict: """Close a running process's stdin / send EOF without killing the process.""" session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"} if session.exited: return {"status": "already_exited", "error": "Process has already finished"} if hasattr(session, '_pty') and session._pty: try: session._pty.sendeof() return {"status": "ok", "message": "EOF sent"} except Exception as e: return {"status": "error", "error": str(e)} if not session.process or not session.process.stdin: return {"status": "error", "error": "Process stdin not available (non-local backend or stdin closed)"} try: session.process.stdin.close() return {"status": "ok", "message": "stdin closed"} except Exception as e: return {"status": "error", "error": str(e)} def count_running(self) -> int: """Return the count of currently-running background processes. Cheap O(1) read of the running dict, suitable for status-bar polling on every render tick. CPython dict ``len()`` is atomic; callers do not need to hold ``self._lock``. Reflects ``_running`` only: sessions are moved to ``_finished`` when their subprocess exits. """ try: return len(self._running) except Exception: return 0 def list_sessions(self, task_id: str = None, session_key: str = None) -> list: """List all running and recently-finished processes. When ``task_id`` is given, processes for that task are included. When ``session_key`` is also given, session-scoped background processes (``background: true``) registered under that gateway session are surfaced too, even if they belong to a different task — so the agent can discover a forgotten preview server that is blocking session reset (#29177). Such cross-task entries are flagged with ``"session_scoped": true``. """ with self._lock: all_sessions = list(self._running.values()) + list(self._finished.values()) all_sessions = [self._refresh_detached_session(s) for s in all_sessions] if task_id or session_key: all_sessions = [ s for s in all_sessions if (task_id and s.task_id == task_id) or (session_key and s.session_key == session_key) ] result = [] for s in all_sessions: entry = { "session_id": s.id, "command": s.command[:200], "cwd": s.cwd, "pid": s.pid, "started_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(s.started_at)), "uptime_seconds": int(time.time() - s.started_at), "status": "exited" if s.exited else "running", "output_preview": s.output_buffer[-200:] if s.output_buffer else "", } # Flag processes surfaced only because they share the gateway # session (not the current task) — these are the long-lived # background processes a user may have forgotten about (#29177). if task_id and session_key and s.task_id != task_id and s.session_key == session_key: entry["session_scoped"] = True # Trigger metadata so a goal-loop judge can decide to wait on this # process's OWN signal (a watch-pattern match or completion), not # just its exit. A watcher with watch_patterns may never exit. if s.watch_patterns and not s._watch_disabled: entry["watch_patterns"] = list(s.watch_patterns) entry["watch_hit"] = s._watch_hits > 0 if s.notify_on_complete: entry["notify_on_complete"] = True if s.exited: entry["exit_code"] = s.exit_code if s.detached: entry["detached"] = True result.append(entry) return result # ----- Session/Task Queries (for gateway integration) ----- def has_active_processes(self, task_id: str) -> bool: """Check if there are active (running) processes for a task_id.""" with self._lock: sessions = list(self._running.values()) for session in sessions: self._refresh_detached_session(session) with self._lock: return any( s.task_id == task_id and not s.exited for s in self._running.values() ) def has_active_for_session( self, session_key: str, max_active_age: Optional[float] = None, ) -> bool: """Check if there are active processes for a gateway session key. When *max_active_age* is set (seconds), processes that started more than that many seconds ago are **ignored** — they are still running but are considered stale and must not block session idle / daily reset. This prevents a forgotten ``http.server`` (or any long-lived preview process) from permanently freezing the session lifecycle. Args: session_key: Gateway session key to check. max_active_age: If set, ignore processes older than this many seconds. ``None`` retains the legacy behaviour (any running process blocks). """ with self._lock: sessions = list(self._running.values()) for session in sessions: self._refresh_detached_session(session) now = time.time() with self._lock: return any( s.session_key == session_key and not s.exited and (max_active_age is None or (now - s.started_at) < max_active_age) for s in self._running.values() ) def has_any_active(self) -> bool: """Whether ANY background process is still running (across all sessions). Used by scale-to-zero idle detection (gateway/scale_to_zero): a gateway with a live background process (terminal background=true) is NOT idle and must not be suspended, or the process is lost. Refreshes detached sessions first so a finished-but-unreaped process reads as inactive. """ with self._lock: sessions = list(self._running.values()) for session in sessions: self._refresh_detached_session(session) with self._lock: return any(not s.exited for s in self._running.values()) def snapshot_running_ids(self, task_id: str) -> frozenset[str]: """Capture running process IDs owned by ``task_id``. Gateway turns use this as a boundary marker: if a turn times out, only processes absent from its starting snapshot belong to the abandoned turn. Older session processes must survive because background tasks intentionally span successful turns. """ with self._lock: return frozenset( s.id for s in self._running.values() if s.task_id == task_id and not s.exited ) def kill_started_since( self, task_id: str, baseline_ids, *, source: str, ) -> int: """Kill processes created for ``task_id`` after a prior snapshot. ``consume_output`` is forced on: abandoned-turn output must not enqueue a synthetic follow-up that revives work the timeout deliberately stopped. """ return self.kill_all( task_id, exclude_ids=frozenset(baseline_ids or ()), source=source, consume_output=True, ) def kill_all( self, task_id: Optional[str] = None, *, exclude_ids: frozenset = frozenset(), source: str = "kill_all", consume_output: bool = False, ) -> int: """Kill all running processes, optionally filtered by task_id. Returns count killed.""" with self._lock: targets = [ s for s in self._running.values() if (task_id is None or s.task_id == task_id) and s.id not in exclude_ids and not s.exited ] killed = 0 for session in targets: result = self.kill_process( session.id, source=source, consume_output=consume_output, ) if result.get("status") in {"killed", "already_exited"}: killed += 1 return killed # ----- Cleanup / Pruning ----- def _prune_if_needed(self): """Remove oldest finished sessions if over MAX_PROCESSES. Must hold _lock.""" # First prune expired finished sessions now = time.time() expired = [ sid for sid, s in self._finished.items() if (now - s.started_at) > FINISHED_TTL_SECONDS ] for sid in expired: del self._finished[sid] self._completion_consumed.discard(sid) self._poll_observed.discard(sid) # If still over limit, remove oldest finished total = len(self._running) + len(self._finished) if total >= MAX_PROCESSES and self._finished: oldest_id = min(self._finished, key=lambda sid: self._finished[sid].started_at) del self._finished[oldest_id] self._completion_consumed.discard(oldest_id) self._poll_observed.discard(oldest_id) # Drop any _completion_consumed / _poll_observed entries whose sessions # are no longer tracked at all — belt-and-suspenders against # module-lifetime growth on registry lookup paths that don't reach the # dict prunes. tracked = self._running.keys() | self._finished.keys() stale = self._completion_consumed - tracked if stale: self._completion_consumed -= stale stale_polls = self._poll_observed - tracked if stale_polls: self._poll_observed -= stale_polls # ----- Checkpoint (crash recovery) ----- def _write_checkpoint( self, extra_entries: Optional[List[Dict[str, Any]]] = None, ): """Write running process metadata to checkpoint file atomically.""" try: with self._lock: entries = [] for s in self._running.values(): if not s.exited: # Lazily backfill the kernel start time for host PIDs so # recovery after restart can detect PID recycling even # for sessions spawned before this field existed. if s.host_start_time is None and s.pid_scope == "host" and s.pid: s.host_start_time = self._safe_host_start_time(s.pid) entries.append({ "session_id": s.id, # Redact inline credentials before persisting to # disk — the checkpoint file lives under # ~/.hermes/processes.json with the raw command # (issue #77484). Recovery only uses command for # display/logging (the process is already running; # adoption re-validates the PID, never re-runs the # command), so masking is lossless. "command": redact_sensitive_text(s.command, code_file=True), "pid": s.pid, "pid_scope": s.pid_scope, "host_start_time": s.host_start_time, "systemd_unit": s.systemd_unit, "cwd": s.cwd, "started_at": s.started_at, "task_id": s.task_id, "owner_task_id": s.owner_task_id or s.task_id, "session_key": s.session_key, "watcher_platform": s.watcher_platform, "watcher_chat_id": s.watcher_chat_id, "watcher_user_id": s.watcher_user_id, "watcher_user_name": s.watcher_user_name, "watcher_thread_id": s.watcher_thread_id, "watcher_message_id": s.watcher_message_id, "watcher_interval": s.watcher_interval, "parent_session_id": s.parent_session_id, "notify_on_complete": s.notify_on_complete, "watch_patterns": s.watch_patterns, }) if extra_entries: tracked_ids = {item.get("session_id") for item in entries} entries.extend( item for item in extra_entries if item.get("session_id") not in tracked_ids ) # Atomic write to avoid corruption on crash from utils import atomic_json_write atomic_json_write(CHECKPOINT_PATH, entries) except Exception as e: logger.debug("Failed to write checkpoint file: %s", e, exc_info=True) def recover_from_checkpoint(self) -> int: """ On gateway startup, probe PIDs from checkpoint file. Returns the number of processes recovered as detached. """ if not CHECKPOINT_PATH.exists(): return 0 try: entries = json.loads(CHECKPOINT_PATH.read_text(encoding="utf-8")) except Exception: return 0 recovered = 0 unresolved_scope_entries: List[Dict[str, Any]] = [] for entry in entries: pid = entry.get("pid") if not pid: continue pid_scope = entry.get("pid_scope", "host") if pid_scope != "host": # Sandbox-backed processes keep only in-sandbox PIDs in the # checkpoint, which are not meaningful to the restarted host # process once the original environment handle is gone. logger.info( "Skipping recovery for non-host process: %s (pid=%s, scope=%s)", entry.get("command", "unknown")[:60], pid, pid_scope, ) continue # The PID must be alive AND still the same process we spawned. A # bare liveness check is unsafe: across a restart (especially a # reboot or long uptime) the kernel may have recycled this number # onto an unrelated process — adopting it would let a later kill or # watcher tree-kill a stranger (e.g. a browser). Re-validate the # kernel start time recorded in the checkpoint. recorded_start = entry.get("host_start_time") if not self._host_pid_is_ours(pid, recorded_start): if self._is_host_pid_alive(pid): logger.info( "Not recovering session %s: pid %d is alive but its " "start time no longer matches — PID was recycled onto " "an unrelated process; refusing to adopt it.", entry.get("session_id", "?"), pid, ) systemd_unit = entry.get("systemd_unit", "") if systemd_unit and not _stop_systemd_unit(systemd_unit): logger.warning( "Could not reap persisted scope %s for dead wrapper pid %s; " "retaining checkpoint entry for the next startup", systemd_unit, pid, ) unresolved_scope_entries.append(entry) continue session = ProcessSession( id=entry["session_id"], command=entry.get("command", "unknown"), task_id=entry.get("task_id", ""), owner_task_id=entry.get("owner_task_id", "") or entry.get("task_id", ""), session_key=entry.get("session_key", ""), pid=pid, host_start_time=recorded_start, pid_scope=pid_scope, systemd_unit=entry.get("systemd_unit", ""), cwd=entry.get("cwd"), started_at=entry.get("started_at", time.time()), detached=True, # Can't read output, but can report status + kill watcher_platform=entry.get("watcher_platform", ""), watcher_chat_id=entry.get("watcher_chat_id", ""), watcher_user_id=entry.get("watcher_user_id", ""), watcher_user_name=entry.get("watcher_user_name", ""), watcher_thread_id=entry.get("watcher_thread_id", ""), watcher_message_id=entry.get("watcher_message_id", ""), watcher_interval=entry.get("watcher_interval", 0), parent_session_id=entry.get("parent_session_id", ""), notify_on_complete=entry.get("notify_on_complete", False), watch_patterns=entry.get("watch_patterns", []), ) with self._lock: self._running[session.id] = session recovered += 1 logger.info("Recovered detached process: %s (pid=%d)", session.command[:60], pid) # Re-enqueue watcher so gateway can resume notifications if session.watcher_interval > 0: self.pending_watchers.append({ "session_id": session.id, "check_interval": session.watcher_interval, "session_key": session.session_key, "platform": session.watcher_platform, "chat_id": session.watcher_chat_id, "user_id": session.watcher_user_id, "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, "notify_on_complete": session.notify_on_complete, "parent_session_id": session.parent_session_id, }) self._write_checkpoint(extra_entries=unresolved_scope_entries) return recovered # Module-level singleton process_registry = ProcessRegistry() def _format_age(seconds: float) -> str: """Human-friendly elapsed string ('18m', '2h3m', '45s').""" try: s = int(max(0, seconds)) except (TypeError, ValueError): return "?" if s < 60: return f"{s}s" m, s = divmod(s, 60) if m < 60: return f"{m}m" if s == 0 else f"{m}m{s}s" h, m = divmod(m, 60) return f"{h}h" if m == 0 else f"{h}h{m}m" def _model_not_found_patterns() -> "list[str]": """Model-not-found phrases from the failover classifier. Imported from ``agent.error_classifier`` so the batch renderer applies the SAME classification the failover path consumes — no hand-copied pattern list to drift. Fails open to a minimal built-in set so a classifier import problem never hides the per-task blocks. (Import approach from PR #97667 by @liuhao1024.) """ try: from agent.error_classifier import _MODEL_NOT_FOUND_PATTERNS return list(_MODEL_NOT_FOUND_PATTERNS) except Exception: return ["is not a valid model", "model not found", "model_not_found"] def _delegation_config() -> dict: """Load the active delegation config (model/provider/fallbacks), fail-open. Mirrors ``tools.delegate_tool._load_config`` so the renderer sees the same ``model`` / ``provider`` the dispatcher used, without importing the heavy delegation module at import time. Returns ``{}`` on any error so callers fail open to "no notice" rather than dropping the per-task blocks. """ try: from tools.delegate_tool import _load_config as _cfg return _cfg() or {} except Exception: return {} def _delegation_model_not_found(results, config) -> bool: """True when a result entry reflects a config-level model_not_found rejection. Matches when at least one entry's error/summary text contains both a model-not-found phrase AND the name of the currently-configured delegation model — so a stale task failing on a *different* (removed) model is not mis-attributed to the config-level root cause. """ model = (config or {}).get("model") if not model: return False model = str(model).lower() for r in results or []: text = " ".join( str(part) for part in (r.get("error"), r.get("summary")) if part ).lower() if not text or model not in text: continue if any(p in text for p in _model_not_found_patterns()): return True return False def _delegation_model_not_found_notice(results) -> "list[str] | None": """Build the config-level model_not_found notice lines, or None. Returns ``None`` unless at least one result entry shows the configured delegation model being rejected by its provider, in which case a short actionable block is returned. Every failure path fails open to ``None`` so a config hiccup never hides the per-task blocks. Emit once per batch. """ config = _delegation_config() if not _delegation_model_not_found(results, config): return None model = config.get("model") or "?" provider = config.get("provider") or "configured provider" lines = [ "⚠ SUBAGENT MODEL REJECTED: the configured Subagent Model " f'"{model}" was rejected by provider "{provider}" ' "(HTTP 400: not a valid model ID).", "Every task in this batch failed for this reason before doing any work.", "Check Settings → Advanced → Subagent Model (or: " "hermes config get delegation.model).", ] try: from hermes_cli.fallback_config import get_fallback_chain if not get_fallback_chain(config): lines.append( "No fallback chain is configured, so no failover was attempted." ) except Exception: pass return lines def _format_async_delegation(evt: dict) -> str: """Format an async-delegation completion into a self-contained re-injection. Carries the FULL original task source (goal, the context the parent supplied, toolsets, role, model) plus dispatch time, status, and the complete result summary. When this re-enters the conversation the agent may be deep in unrelated context and won't remember why the subagent existed, so the block is written to stand entirely on its own — enough to use the result OR re-dispatch if the world has moved on. """ import time as _time deleg_id = evt.get("delegation_id", "unknown") goal = evt.get("goal", "") or "" context = evt.get("context") toolsets = evt.get("toolsets") role = evt.get("role") or "leaf" model = evt.get("model") or "?" status = evt.get("status") or "completed" summary = evt.get("summary") error = evt.get("error") api_calls = evt.get("api_calls", 0) duration = evt.get("duration_seconds", "?") truncated = evt.get("truncated") or evt.get("exit_reason") == "max_iterations" dispatched_at = evt.get("dispatched_at") completed_at = evt.get("completed_at") or _time.time() # ----- Batch (fan-out) completion: consolidated multi-task block ----- # A whole delegate_task fan-out dispatched as one background unit finishes # together and carries a per-task `results` list. Render every subagent's # summary in one block so the model gets the consolidated outcome at once. batch_results = evt.get("results") if evt.get("is_batch") or isinstance(batch_results, list): results = batch_results or [] goals = evt.get("goals") or [] n = len(results) if results else len(goals) total_dur = evt.get("total_duration_seconds", duration) lines = [ f"[ASYNC DELEGATION BATCH COMPLETE — {deleg_id}]", f"A background fan-out of {n} subagent(s) you dispatched earlier " "has finished. All ran in parallel and waited on each other; their " "consolidated results are below. You may have moved on since " "dispatching — act on these or re-dispatch if things have changed.", "", ] if isinstance(dispatched_at, (int, float)): ts = _time.strftime("%Y-%m-%d %H:%M:%S", _time.localtime(dispatched_at)) age = f" ({_format_age(completed_at - dispatched_at)} ago)" lines.append(f"Dispatched: {ts}{age}") if context: lines.append(f"Context you provided: {context}") if toolsets: lines.append(f"Toolsets: {', '.join(toolsets)}") lines.append(f"Role: {role} Model: {model} Total duration: {total_dur}s") if error and not results: lines.append("--- ERROR ---") lines.append(f"The batch did not complete successfully: {error}") return "\n".join(lines) # Config-level rejection notice BEFORE the per-task wall — a rejected # delegation model fails every task identically before doing any # work, and that signal must not stay buried in the task blocks. _notice = _delegation_model_not_found_notice(results) if _notice: lines.append("") lines.extend(_notice) for r in sorted(results, key=lambda x: x.get("task_index", 0)): idx = r.get("task_index", 0) r_status = r.get("status", "?") r_summary = r.get("summary") r_error = r.get("error") r_goal = goals[idx] if idx < len(goals) else r.get("goal", "") r_truncated = r.get("truncated") or r.get("exit_reason") == "max_iterations" icon = "⚠" if r_truncated else ("✓" if r_status in ("completed", "success") else "✗") lines.append("") header = f"--- {icon} TASK {idx + 1}/{n}" if r_goal: header += f": {r_goal}" header += f" (status={r_status}" if r.get("api_calls"): header += f", api_calls={r['api_calls']}" if r.get("duration_seconds") is not None: header += f", {r['duration_seconds']}s" if r_truncated: header += ", TRUNCATED: hit max_iterations — work may be incomplete" header += ") ---" lines.append(header) if r_status in ("completed", "success") and r_summary: if r_truncated: lines.append( "[TRUNCATED — subagent hit its iteration cap; the " "summary below may be incomplete. Verify before relying " "on it, or re-dispatch the unfinished part.]" ) lines.append(r_summary) elif r_summary: if r_error: lines.append(f"({r_status}: {r_error})") lines.append("Partial output:") lines.append(r_summary) else: lines.append( f"(no summary — status={r_status}" + (f": {r_error}" if r_error else "") + ")" ) r_live = r.get("live_transcript") if r_live: lines.append( f"Full live transcript (complete tool/assistant trace): {r_live}" ) return "\n".join(lines) age = "" if isinstance(dispatched_at, (int, float)): age = f" ({_format_age(completed_at - dispatched_at)} ago)" lines = [ f"[ASYNC DELEGATION COMPLETE — {deleg_id}]", "A background subagent you dispatched earlier has finished. You may " "have moved on since dispatching it; the full task source is below so " "you can act on the result or re-dispatch if things have changed.", "", ] if isinstance(dispatched_at, (int, float)): ts = _time.strftime("%Y-%m-%d %H:%M:%S", _time.localtime(dispatched_at)) lines.append(f"Dispatched: {ts}{age}") lines.append(f"Original goal: {goal}") if context: lines.append(f"Context you provided: {context}") if toolsets: lines.append(f"Toolsets: {', '.join(toolsets)}") lines.append(f"Role: {role} Model: {model}") _notice = _delegation_model_not_found_notice([evt]) if _notice: lines.append("") lines.extend(_notice) _trunc = " [TRUNCATED: hit max_iterations — work may be incomplete]" if truncated else "" lines.append(f"Status: {status} API calls: {api_calls} Duration: {duration}s{_trunc}") lines.append("--- RESULT ---") if status in ("completed", "success") and summary: if truncated: lines.append( "[TRUNCATED — subagent hit its iteration cap; the summary below " "may be incomplete. Verify before relying on it, or re-dispatch " "the unfinished part.]" ) lines.append(summary) elif status == "interrupted": lines.append( "The subagent was interrupted before completing" + (f": {error}" if error else ".") ) if summary: lines.append("Partial output:") lines.append(summary) else: # error / timeout / failed lines.append( f"The subagent did not complete successfully (status={status})." + (f"\n{error}" if error else "") ) if summary: lines.append("Partial output:") lines.append(summary) return "\n".join(lines) def _delegation_attribution_line(evt: dict) -> "str | None": """One-line delegation attribution for a child-originated process event. Subagents run their terminal sessions under ``task_id == subagent_id`` (delegate_tool._run_single_child). When a background process they started completes, its notification is routed to the PARENT conversation by design (children consume their own waits via process(wait); anything that outlives the child must land where a durable consumer exists). Without attribution the parent-facing user sees an anonymous raw output wall mid-conversation with no hint it came from a delegation. Resolve the task_id against the live + recently-finished subagent registry and return a short provenance line, or None for parent-owned processes. """ task_id = str(evt.get("owner_task_id") or evt.get("task_id") or "") if not task_id.startswith("sa-"): return None try: from tools.delegate_tool import get_subagent_attribution info = get_subagent_attribution(task_id) except Exception: info = None if not info: # The task_id shape says "subagent" even when the registry entry has # aged out — still attribute generically rather than anonymously. return f"Started by subagent {task_id} (delegate_task)." goal = str(info.get("goal") or "").strip() if len(goal) > 120: goal = goal[:117] + "..." deleg = info.get("delegation_id") parts = [f"Started by subagent {task_id}"] if deleg: parts.append(f"of delegation {deleg}") line = " ".join(parts) + "." if goal: line += f' Task: "{goal}"' return line def format_process_notification(evt: dict) -> "str | None": """Format a process notification event into a [IMPORTANT: ...] message. Handles completion events (notify_on_complete), watch pattern matches, and watch disabled events from the unified completion_queue. """ evt_type = evt.get("type", "completion") _sid = evt.get("session_id", "unknown") _cmd = evt.get("command", "unknown") _attribution = _delegation_attribution_line(evt) if evt_type == "watch_disabled": return f"[IMPORTANT: {evt.get('message', '')}]" # Overflow events carry their human-readable summary in `message` — # without this case they fall through to the completion formatter and # surface as a phantom "process exited (exit code ?)" notification. if evt_type in ("watch_overflow_tripped", "watch_overflow_released"): return f"[IMPORTANT: {evt.get('message', '')}]" if evt_type == "watch_match": _pat = evt.get("pattern", "?") _out = evt.get("output", "") _sup = evt.get("suppressed", 0) text = ( f"[IMPORTANT: Background process {_sid} matched " f"watch pattern \"{_pat}\".\n" ) if _attribution: text += f"{_attribution}\n" text += ( f"Command: {_cmd}\n" f"Matched output:\n{_out}" ) if _sup: text += f"\n({_sup} earlier matches were suppressed by rate limit)" text += "]" return text if evt_type == "async_delegation": return _format_async_delegation(evt) _exit = evt.get("exit_code", "?") _out = evt.get("output", "") _reason = evt.get("completion_reason") or "exited" _source = evt.get("termination_source") or "" _signal = "" if _exit in {-15, 143, "-15", "143"}: _signal = ", SIGTERM" if _reason == "killed": _status = f"terminated by {_source or 'Hermes'}" elif _reason == "lost": _status = "marked lost because the process backend disappeared" elif _reason == "failed_start": _status = "failed to start" elif _exit == 0: _status = "completed normally" else: _status = "exited" text = ( f"[IMPORTANT: Background process {_sid} {_status} " f"(exit code {_exit}{_signal}).\n" ) if _attribution: text += f"{_attribution}\n" # A subagent-owned process's full output belongs in the child's # transcript/summary, not as a raw wall in the parent conversation — # trim the tail hard while keeping enough to recognise failures. if isinstance(_out, str) and len(_out) > 600: _out = ( "...(output trimmed — subagent-owned process; see the " "delegation's live transcript for full output)\n" + _out[-600:] ) text += ( f"Command: {_cmd}\n" f"Output:\n{_out}]" ) return text # --------------------------------------------------------------------------- # Registry -- the "process" tool schema + handler # --------------------------------------------------------------------------- from tools.registry import registry, tool_error PROCESS_SCHEMA = { "name": "process_manage", # Dieted (#95681): the action enum names the verbs; the description # keeps only non-obvious semantics. write-vs-submit is the tool's one # real trap (a lone \n on a Windows PTY is not a line terminator) — # that teaching gains emphasis rather than losing it. "description": ( "Poll, wait on, or kill background terminal processes (from " "terminal(background=true)). " "poll: status + new output. log: full output, paged. wait: block " "until exit or timeout (partial output on timeout). write vs " "submit: submit appends Enter — use it to answer prompts; write " "sends raw bytes, no newline. close: EOF stdin. kill: terminate." ), "parameters": { "type": "object", "properties": { "action": { "type": "string", "enum": ["list", "poll", "log", "wait", "kill", "write", "submit", "close"] }, "session_id": { "type": "string", "description": "From terminal background output; any unique prefix works ('4dae' for proc_4dae56ca81f6). Required except for 'list'." }, "data": { "type": "string", "description": "Stdin text for write/submit." }, "timeout": { "type": "integer", "description": "Max seconds for 'wait'.", "minimum": 1 }, "offset": { "type": "integer", "description": "Log line offset (default: last 200)." }, "limit": { "type": "integer", "description": "Max log lines.", "minimum": 1 } }, "required": ["action"] } } def _redact_process_result(result: dict) -> dict: """Redact secrets from background-process output before it reaches the model, session.db, and CLI display. Mirrors the foreground ``terminal`` redaction (terminal_tool.py) so the two surfaces can't diverge — issue #43025 (background output was returned verbatim). Respects ``security.redact_secrets`` (no force): output fields pass through ``redact_terminal_output`` which picks ``code_file`` based on the recorded command (env dumps get the ENV-assignment pass). The command string itself is also redacted in case it carried an inline credential. """ if not isinstance(result, dict): return result from agent.redact import redact_sensitive_text, redact_terminal_output command = result.get("command") or "" for field in ("output", "output_preview"): value = result.get(field) if isinstance(value, str) and value: result[field] = redact_terminal_output(value, command) if isinstance(result.get("command"), str) and result["command"]: result["command"] = redact_sensitive_text(result["command"], code_file=True) return result def _handle_process(args, **kw): task_id = kw.get("task_id") action = args.get("action", "") # Coerce to string — some models send session_id as an integer session_id = str(args.get("session_id", "")) if args.get("session_id") is not None else "" if action == "list": # Surface session-scoped background processes (e.g. a forgotten # preview server) in addition to this task's own — they share the # gateway session_key and can block session reset (#29177). try: from tools.approval import get_current_session_key session_key = get_current_session_key(default="") or "" except Exception: session_key = "" return json.dumps( { "processes": [ _redact_process_result(p) for p in process_registry.list_sessions(task_id=task_id, session_key=session_key or None) ] }, ensure_ascii=False, ) elif action in {"poll", "log", "wait", "kill", "write", "submit", "close"}: if not session_id: return tool_error(f"session_id is required for {action}") if action == "poll": return json.dumps(_redact_process_result(process_registry.poll(session_id)), ensure_ascii=False) elif action == "log": return json.dumps(_redact_process_result(process_registry.read_log( session_id, offset=args.get("offset"), limit=args.get("limit", 200))), ensure_ascii=False) elif action == "wait": return json.dumps(_redact_process_result(process_registry.wait(session_id, timeout=args.get("timeout"))), ensure_ascii=False) elif action == "kill": return json.dumps( _redact_process_result(process_registry.kill_process(session_id)), ensure_ascii=False, ) elif action == "write": return json.dumps(process_registry.write_stdin(session_id, str(args.get("data", ""))), ensure_ascii=False) elif action == "submit": return json.dumps(process_registry.submit_stdin(session_id, str(args.get("data", ""))), ensure_ascii=False) elif action == "close": return json.dumps(process_registry.close_stdin(session_id), ensure_ascii=False) return tool_error(f"Unknown process action: {action}. Use: list, poll, log, wait, kill, write, submit, close") registry.register( name="process_manage", toolset="terminal", schema=PROCESS_SCHEMA, handler=_handle_process, emoji="⚙️", )