From cc9ef823550d88c6bc060e75b2aed8dcfc08e98f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 12:15:38 -0700 Subject: [PATCH] refactor(gateway-cli): dispatch-table subcommands and shared service helpers in hermes_cli/gateway.py hermes_cli/gateway.py (9178 -> 6906): - `hermes gateway` subcommand routing: if/elif chain -> _GATEWAY_SUBCOMMANDS dispatch table of _cmd_* handlers; _stop_installed_service and _refuse_from_inside_gateway unify the stop/restart/uninstall service-stop and self-target guards. - One systemd unit template; _systemctl_show, systemctl reset+action, is-active probes, _installed_service_kind ladder (3 sites), launchctl-list PID probe, launchd bootstrap+kickstart helper, legacy-unit removal loop, loop-tick witness ping, ps/wmic line parsers, planned-stop marker helper, _CAPTURE_TEXT subprocess kwargs (21 sites), _gw_windows() accessor (15 lazy imports). - run_gateway startup helpers, reaper exclusion set, Windows process listing extracted; wizard service actions and platform-setup prompts unified (_setup_service_action). - Dead: _windows_scheduled_task_running (test-only), _container_systemd_operational (inlined), whatsapp/email/matrix built-in status branches (those platforms are registry plugins). - try/except-pass around single statements -> contextlib.suppress; comments/docstrings compacted to the WHY (bounded_probe_run vs subprocess.run on Windows, pythonw launcher-stub filtering, raw-record vs validated-probe exclusion in the orphan reaper, Scheduled-Task Ready-vs-Running). - `hermes gateway --help` byte-identical. --- hermes_cli/gateway.py | 5798 +++++++++--------------------- tests/hermes_cli/test_gateway.py | 4 - 2 files changed, 1763 insertions(+), 4039 deletions(-) diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 6b155487cc..f4688751ee 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -1,10 +1,10 @@ -""" -Gateway subcommand for hermes CLI. +"""Gateway subcommand for hermes CLI. Handles: hermes gateway [run|start|stop|restart|status|install|uninstall|setup] """ import asyncio +import contextlib from hermes_cli.cli_output import line_input import json import logging @@ -20,8 +20,7 @@ import time from dataclasses import dataclass from pathlib import Path -# Ensure /bin and /usr/bin are on PATH so launchctl/systemctl are discoverable -# when running under UV's bundled Python which ships a minimal PATH (#3849). +# UV's bundled Python ships a minimal PATH; ensure launchctl/systemctl are discoverable. if os.name == "posix": _sys_dirs = {"/bin", "/usr/bin", "/usr/sbin", "/sbin"} _path_dirs = set(os.environ.get("PATH", "").split(os.pathsep)) @@ -34,7 +33,6 @@ PROJECT_ROOT = Path(__file__).parent.parent.resolve() from gateway.config import coerce_systemd_watchdog_seconds, load_gateway_config from gateway.status import terminate_pid from gateway.restart import ( - DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, EXTERNAL_GATEWAY_SUPERVISOR_ENV, GATEWAY_FATAL_CONFIG_EXIT_CODE, @@ -56,8 +54,7 @@ from hermes_cli.config import ( write_platform_config_field, ) -# display_hermes_home is imported lazily at call sites to avoid ImportError -# when hermes_constants is cached from a pre-update version during `hermes update`. +# display_hermes_home is imported lazily: hermes_constants may be a cached pre-update version. from hermes_cli.setup import ( print_header, print_info, @@ -72,6 +69,9 @@ from hermes_cli.colors import Colors, color logger = logging.getLogger(__name__) +# Shared ``subprocess.run`` kwargs for text-mode probes (stdout/stderr captured, decode-tolerant). +_CAPTURE_TEXT = dict(capture_output=True, text=True, encoding="utf-8", errors="replace") + # ============================================================================= # Process Management (for manual gateway runs) # ============================================================================= @@ -117,48 +117,25 @@ class WindowsGatewayService: def _get_service_pids(all_profiles: bool = False) -> set: - """Return PIDs currently managed by systemd or launchd gateway services. + """Return PIDs managed by systemd/launchd gateway services (excluded from stale-process sweeps). - Used to avoid killing freshly-restarted service processes when sweeping - for stale manual gateway processes after a service restart. Relies on - the service manager having committed the new PID before the restart command - returns (true for both systemd and launchd in practice). - - ``all_profiles`` widens the launchd branch to every installed - ``ai.hermes.gateway*`` LaunchAgent — the update path needs the whole - fleet excluded from its sweep (#41403, #73626): sibling-profile launchd - gateways found by the (BSD-fixed) ps scan must not be misclassified as - manual processes and killed. Default-scope callers (``gateway status``, - cron checks) keep seeing only the current profile's service; the orphan - reaper passes all_profiles=True for the same friendly-fire reason. The - systemd branch mirrors this: default scope filters to the current - profile's exact unit name; ``all_profiles=True`` widens to the - ``hermes-gateway*`` fleet glob. + Relies on the service manager committing the new PID before the restart command returns. + Default scope covers only the current profile's unit/label; ``all_profiles`` widens to the + whole ``hermes-gateway*`` / ``ai.hermes.gateway*`` fleet so the update path and orphan reaper + never misclassify a sibling profile's service gateway as a manual process and kill it. """ pids: set = set() # --- systemd (Linux): user and system scopes --- - # Default scope lists only this profile's unit (the unit name encodes the - # profile via get_service_name()); all_profiles widens to the fleet glob. if supports_systemd_services(): - if all_profiles: - pattern = "hermes-gateway*" - else: - pattern = get_service_name() + pattern = "hermes-gateway*" if all_profiles else get_service_name() for scope_args in [["systemctl", "--user"], ["systemctl"]]: try: result = subprocess.run( scope_args - + [ - "list-units", - pattern, - "--plain", - "--no-legend", - "--no-pager", - ], - capture_output=True, - text=True, encoding='utf-8', errors='replace', + + ["list-units", pattern, "--plain", "--no-legend", "--no-pager"], timeout=5, + **_CAPTURE_TEXT, ) for line in result.stdout.strip().splitlines(): parts = line.split() @@ -168,9 +145,8 @@ def _get_service_pids(all_profiles: bool = False) -> set: try: show = subprocess.run( scope_args + ["show", svc, "--property=MainPID", "--value"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', timeout=5, + **_CAPTURE_TEXT, ) pid = int(show.stdout.strip()) if pid > 0: @@ -184,11 +160,7 @@ def _get_service_pids(all_profiles: bool = False) -> set: if is_macos(): labels = {get_launchd_label()} if all_profiles: - # Every gateway LaunchAgent, not just the invoking profile's — - # mirrors the systemd branch's ``hermes-gateway*`` pattern above. - # The update path restarts the whole fleet, and its stale-process - # sweep must not mistake a sibling service's fresh PID for a - # manual gateway it should kill (#41403). + # Whole fleet, mirroring the systemd ``hermes-gateway*`` glob above. labels.update(launchd_gateway_labels_for_install()) for label in sorted(labels): try: @@ -198,25 +170,14 @@ def _get_service_pids(all_profiles: bool = False) -> set: if pid is not None and pid > 0: pids.add(pid) if all_profiles: - # Belt-and-suspenders for the EXCLUDE use case (#74075): a bare - # ``launchctl list`` prefix scan also catches ai.hermes.gateway* - # agents the label derivation can't map (renamed profiles, other - # installs sharing this user). Over-inclusion is safe here — - # these PIDs are only ever protected from the kill sweep, never - # targeted. Restart paths use the label-derived set only. + # Prefix scan also catches ai.hermes.gateway* agents the label derivation can't map + # (renamed profiles, other installs). Over-inclusion is safe: PIDs are only protected. try: - result = subprocess.run( - ["launchctl", "list"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=5, - ) + result = subprocess.run(["launchctl", "list"], timeout=5, **_CAPTURE_TEXT) if result.returncode == 0: for line in result.stdout.strip().splitlines(): parts = line.split() - if len(parts) >= 3 and parts[-1].startswith( - "ai.hermes.gateway" - ): + if len(parts) >= 3 and parts[-1].startswith("ai.hermes.gateway"): try: pid = int(parts[0]) if pid > 0: @@ -230,14 +191,7 @@ def _get_service_pids(all_profiles: bool = False) -> set: def _get_parent_pid(pid: int) -> int | None: - """Return the parent PID for ``pid``, or ``None`` when unavailable. - - Uses psutil (core dependency) which works on every platform. The - older implementation shelled out to ``ps -o ppid= -p ``, which - silently fails on Windows (no ``ps``) so the ancestor walk terminated - at self — the caller's dedup / exclude logic then couldn't distinguish - "hermes CLI that invoked this scan" from "real gateway process". - """ + """Return the parent PID for ``pid``, or ``None``. psutil first (works on Windows, where ``ps`` doesn't).""" if pid <= 1: return None try: @@ -248,26 +202,17 @@ def _get_parent_pid(pid: int) -> int | None: pass except Exception: return None - # Fallback: shell out to ps (POSIX only). Git Bash installs ``ps.exe`` on - # Windows; running it from the windowless desktop/gateway backend flashes a - # console, and psutil above is the authoritative Windows path anyway. + # ps fallback, POSIX only: Git Bash's ps.exe would flash a console from the windowless backend. if is_windows(): return None if not shutil.which("ps"): return None try: - result = subprocess.run( - ["ps", "-o", "ppid=", "-p", str(pid)], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=5, - ) + result = subprocess.run(["ps", "-o", "ppid=", "-p", str(pid)], timeout=5, **_CAPTURE_TEXT) except (FileNotFoundError, subprocess.TimeoutExpired): return None - if result.returncode != 0: - return None raw = result.stdout.strip() - if not raw: + if result.returncode != 0 or not raw: return None try: parent_pid = int(raw.splitlines()[-1].strip()) @@ -305,30 +250,12 @@ def _request_gateway_self_restart(pid: int) -> bool: def _graceful_restart_via_sigusr1(pid: int, drain_timeout: float) -> bool: - """Send SIGUSR1 to a gateway PID and wait for it to exit gracefully. + """Send SIGUSR1 (drain-aware restart) to a gateway PID and wait for it to exit. - SIGUSR1 is wired in gateway/run.py to ``request_restart(via_service=True)``, - which refuses new turns, waits for in-flight work up to - ``agent.restart_after_turn_timeout``, then runs ``stop()`` (force-interrupt - budget ``agent.restart_drain_timeout``) and exits. Both systemd - (``Restart=always``) and launchd (unconditional KeepAlive) restart on - any exit. - - This is the drain-aware alternative to ``systemctl restart`` / ``SIGTERM``, - which SIGKILL in-flight agents after a short timeout. - - Args: - pid: Gateway process PID (systemd MainPID, launchd PID, or bare - process PID). - drain_timeout: Seconds to wait for the process to exit after sending - SIGUSR1. Must cover the after-turn wait plus the stop()/drain - phase (#77184); callers should pass - ``resolve_restart_exit_wait_budget(...)``. - - Returns: - True if the PID was signalled and exited within the timeout. - False if SIGUSR1 couldn't be sent or the process didn't exit in - time (caller should fall back to a harder restart path). + gateway/run.py maps SIGUSR1 to ``request_restart(via_service=True)``: refuse new turns, wait + for in-flight work, ``stop()``, exit; systemd/launchd then relaunch. ``drain_timeout`` must + cover the after-turn wait plus the drain — pass ``resolve_restart_exit_wait_budget(...)``. + Returns False if the signal couldn't be sent or the process outlived the timeout. """ if not hasattr(signal, "SIGUSR1"): return False @@ -337,34 +264,25 @@ def _graceful_restart_via_sigusr1(pid: int, drain_timeout: float) -> bool: try: os.kill(pid, signal.SIGUSR1) # windows-footgun: ok — POSIX signal, guarded by hasattr(signal, 'SIGUSR1') above except ProcessLookupError: - # Already gone — nothing to drain. return True except (PermissionError, OSError): return False - # Drain-wait: delegate to the shared PID-exit helper (0.5s poll, bounded). return _wait_for_pid_exit(pid, max(drain_timeout, 1.0)) def _wait_for_pid_exit(pid: int, timeout: float) -> bool: - """Wait up to ``timeout`` seconds for ``pid`` to leave the process table. + """Wait up to ``timeout``s for ``pid`` to exit; True once gone, False on timeout. - ``launchctl bootstrap`` of a label whose previous instance is still draining - fails with EIO ("already loaded"), so callers that tear the gateway down - must wait for the old process to actually exit before re-bootstrapping. - - Returns True once the PID is gone (or was never alive), False on timeout. + ``launchctl bootstrap`` fails with EIO while the previous instance is still draining, so + teardown callers must wait for the real exit before re-bootstrapping. """ if pid <= 0: return True import time as _time - # IMPORTANT Windows note: ``os.kill(pid, 0)`` is NOT a no-op on - # Windows — Python's implementation calls ``TerminateProcess(handle, 0)`` - # for sig=0, hard-killing the target. Use the cross-platform - # ``_pid_exists`` helper in gateway.status which does OpenProcess + - # WaitForSingleObject on Windows. + # ``os.kill(pid, 0)`` hard-kills on Windows (TerminateProcess); use _pid_exists instead. from gateway.status import _pid_exists deadline = _time.monotonic() + max(timeout, 0.0) @@ -376,85 +294,38 @@ def _wait_for_pid_exit(pid: int, timeout: float) -> bool: _time.sleep(0.5) -# --- Wedged-gateway detection + bounded escalation (#81642) ----------------- +# --- Wedged-gateway detection + bounded escalation --------------------------- # -# A gateway whose asyncio loop is stalled (e.g. an in-loop compression pass, -# #72707) cannot process SIGTERM/SIGUSR1 shutdown: the drain wait then burns -# the full drain budget (180s by default), warns "still running after 180.0s -# — restart may fail", and `hermes update` can deadlock behind it. The loop -# publishes a liveness signal precisely for this case: an asyncio task -# rewrites ``state/gateway.heartbeat`` every 30s (#66892), so a frozen loop -# stops refreshing the file while a busy-but-alive loop keeps refreshing it. +# A gateway whose asyncio loop is stalled cannot handle SIGTERM/SIGUSR1, so the drain wait burns +# its full budget and `hermes update` can deadlock. Two witnesses classify the loop BEFORE any +# drain wait: the heartbeat file ``state/gateway.heartbeat`` (rewritten every 30s, but on a thread +# — so freshness/staleness alone is not proof) and the loop-tick socket +# ``state/gateway.loop-tick..sock``, answered by the loop itself; the payload records whether +# the socket is armed (``loop_tick_socket``). # -# Since #90502 the heartbeat write runs on a thread (a stalling filesystem -# must not be able to block the loop the watchdog watches), which costs the -# file its status as *proof*: a stalled write or a saturated executor can age -# the file while the loop runs, and an off-loop write can land after the loop -# froze, keeping the file fresh for a dead loop. The loop therefore also arms -# a second witness — ``state/gateway.loop-tick..sock``, a UNIX socket -# answered by the loop itself — and records whether it is armed in the -# heartbeat payload (``loop_tick_socket``). +# - ``alive`` — socket answered, or file fresh and not contradicted. Normal graceful drain, +# which honours the in-flight cron drain floor. +# - ``wedged`` — heartbeat is this PID's, stale past several beats, AND the armed socket stays +# silent across ``tick_strikes`` consecutive misses. Only then may callers +# escalate via ``_escalate_wedged_gateway``; one silent probe is never authority. +# - ``unknown`` — no/unreadable heartbeat, PID mismatch, or witness conflict. Treated as alive: +# never escalate on ambiguity. # -# ``probe_gateway_loop_liveness`` reads both signals (a local stat + JSON -# read + a bounded socket ping, repeated up to ``tick_strikes`` times when a -# wedge is suspected — worst case ~3.4s, still far inside the 10s query tier -# of the subprocess timeout doc) and classifies the gateway BEFORE any drain -# wait begins: -# -# - ``alive`` — the loop answered the tick socket, or the file is fresh and -# the loop is not contradicted by the socket. Callers must -# take the normal graceful-drain path, which honours the -# in-flight cron drain floor (#86684). -# - ``wedged`` — the heartbeat belongs to this PID, is stale well past -# several missed beats, AND the tick socket is armed but -# stays silent across a sustained window of consecutive -# misses (default 3): both witnesses agree, sustained, that -# the loop is provably dead. One silent probe is never -# destructive authority — a transient synchronous stall can -# outlast a single recv timeout, so a lone miss falls to -# ``unknown``. Draining is pointless for a provably dead -# loop (nothing can run the drain), so callers may escalate -# immediately via ``_escalate_wedged_gateway``. -# - ``unknown`` — no heartbeat / unreadable / PID mismatch / witness conflict -# (fresh file with a silent loop, armed socket unreachable). -# Treated like ``alive``: never escalate on ambiguity. -# -# The distinction matters: only a *provably dead* loop may bypass the cron -# drain floor. A merely busy gateway still answers the probe (socket ping) -# and keeps its full drain budget — even when the filesystem is stalling the -# heartbeat write (the incident that motivated #90502). -# -# Legacy gateways (no ``loop_tick_socket`` flag in the payload) wrote the -# file on-loop, so their staleness remains proof and the old single-witness -# contract is unchanged. +# Legacy payloads (no ``loop_tick_socket`` flag) wrote on-loop, so staleness alone remains proof. GATEWAY_LOOP_ALIVE = "alive" GATEWAY_LOOP_WEDGED = "wedged" GATEWAY_LOOP_UNKNOWN = "unknown" -# Heartbeat cadence is 30s (gateway.shutdown_watchdog.DEFAULT_HEARTBEAT_INTERVAL_S). -# Three missed beats is decisive without false-positiving on one slow write. +# 3 missed 30s beats (gateway.shutdown_watchdog.DEFAULT_HEARTBEAT_INTERVAL_S): decisive, not one slow write. DEFAULT_LOOP_LIVENESS_STALE_AFTER_S = 90.0 # Sentinel for "the producer never wrote the witness flag" (legacy payload). _LOOP_TICK_ABSENT = object() -def _probe_loop_tick_socket( - pid: int, - home: Path | None, - timeout: float = 1.0, -) -> bool | None: - """Ping the loop-scheduling witness socket for ``pid``. - - Returns: - True — the loop answered: it is dispatching right now. - False — a socket node exists for this PID but did not answer (the loop - is not scheduling, or the node is a leftover from a dead - listener). - None — no socket node for this PID (legacy producer), or the path - could not be resolved. Not evidence either way. - """ +def _probe_loop_tick_socket(pid: int, home: Path | None, timeout: float = 1.0) -> bool | None: + """Ping the loop-tick witness socket: True answered, False node present but silent, None no node (not evidence).""" try: from gateway.shutdown_watchdog import get_loop_tick_socket_path @@ -463,62 +334,35 @@ def _probe_loop_tick_socket( return None except Exception: return None + return _ping_loop_tick_witness(socket.AF_UNIX, str(path), timeout) + + +def _ping_loop_tick_witness(family: int, address, timeout: float) -> bool: + """Connect to a loop-tick witness and expect one byte ``"1"``; False on refusal/timeout/any error.""" sock = None try: - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock = socket.socket(family, socket.SOCK_STREAM) sock.settimeout(max(float(timeout), 0.0)) - sock.connect(str(path)) + sock.connect(address) return sock.recv(1) == b"1" except Exception: - # ECONNREFUSED (node with no listener), timeout (loop not answering), - # transient errors: the witness exists but is silent. return False finally: if sock is not None: - try: + with contextlib.suppress(Exception): sock.close() - except Exception: - pass -def _probe_loop_tick_tcp( - port: int, - timeout: float = 1.0, -) -> bool | None: - """Ping the loop-scheduling witness via TCP loopback (Windows). - - Same protocol and semantics as the Unix socket variant: connect to - 127.0.0.1: and expect one byte "1" as proof the loop is - dispatching. Used on Windows / non-POSIX systems where AF_UNIX is not - available in asyncio. - - Returns: - True — the loop answered. - False — the port was reachable but did not answer, or refused. - None — invalid port / could not connect for unrelated reasons. - """ +def _probe_loop_tick_tcp(port: int, timeout: float = 1.0) -> bool | None: + """TCP-loopback variant of the tick probe for Windows (no AF_UNIX in asyncio); same semantics, None + on invalid port.""" try: port_num = int(port) if port_num <= 0 or port_num > 65535: return None except (TypeError, ValueError): return None - sock = None - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(max(float(timeout), 0.0)) - sock.connect(("127.0.0.1", port_num)) - return sock.recv(1) == b"1" - except Exception: - # Connection refused, timeout, transient errors: witness exists - # but is silent (or the process is dead and the port is closed). - return False - finally: - if sock is not None: - try: - sock.close() - except Exception: - pass + return _ping_loop_tick_witness(socket.AF_INET, ("127.0.0.1", port_num), timeout) def _probe_loop_tick_socket_sustained( @@ -530,23 +374,11 @@ def _probe_loop_tick_socket_sustained( gap_s: float = 0.2, tcp_port: int | None = None, ) -> bool | None: - """Probe the tick socket until a reply or the sustained-miss budget. + """Probe the tick socket up to ``strikes`` times, ``gap_s`` apart, until a reply. - A single silent probe is NOT destructive evidence: the loop may be in a - short transient synchronous stall (a reconnect storm, a heavy synchronous - callback, scheduler delay) that outlasts one recv timeout. Killing a - gateway on that would be a false wedge — the exact class of false - positive #90502 exists to prevent. Destructive authority therefore - requires the loop to fail to answer across a bounded window of - ``strikes`` consecutive misses, ``gap_s`` apart; any answer inside the - window proves the loop is dispatching and returns ``True``. - - Returns: - True — some attempt got an answer: the loop is dispatching. - False — every attempt observed a socket node that stayed silent: the - loop did not schedule for the whole window. - None — a probe found no socket node (witness vanished mid-window, or - legacy producer): not evidence either way. + One silent probe is not destructive evidence (a transient synchronous stall can outlast one + recv timeout). True: some attempt answered. False: a node stayed silent the whole window. + None: no socket node on some attempt (vanished / legacy producer) — not evidence. """ total = max(int(strikes), 0) for attempt in range(total): @@ -557,9 +389,7 @@ def _probe_loop_tick_socket_sustained( if result is True: return True if result is None: - # No socket node this attempt — either the witness never - # existed (legacy producer) or it vanished mid-window. Both - # are ambiguity, never a wedge: absence is not a miss. + # No node: ambiguity, never a wedge — absence is not a miss. return None if attempt < total - 1 and gap_s > 0: time.sleep(gap_s) @@ -576,34 +406,11 @@ def probe_gateway_loop_liveness( tick_strikes: int = 3, tick_gap_s: float = 0.2, ) -> str: - """Classify a gateway PID's event loop as alive / wedged / unknown. + """Classify a gateway PID's event loop as alive / wedged / unknown (see block comment above). - Two witnesses: - - - the loop-tick socket (``state/gateway.loop-tick..sock``): answered - by the gateway loop itself, so a reply is direct proof that the loop is - dispatching. It is never refreshed by the heartbeat executor thread and - never stalled by a filesystem that is slow to fsync. - - the heartbeat file (``state/gateway.heartbeat``): rewritten every 30s - on a thread since #90502, so freshness alone is no longer proof of loop - schedulability — a stalled write (measured at 112.6s max on the - incident box) or a saturated executor can age the file while the loop - runs, and a write can land after the loop froze. - - A stale file classifies as ``wedged`` only when the producer declared the - tick socket armed (``loop_tick_socket: true`` in the payload) AND the - socket stays silent across a sustained window — ``tick_strikes`` - consecutive misses (default 3). One silent probe is never destructive - authority: a short transient synchronous stall can outlast a single recv - timeout, so a single miss returns ``unknown`` and keeps the graceful - drain path. Any answer inside the window proves the loop is dispatching - and returns ``alive``. Any conflict or ambiguity returns ``unknown`` so - callers keep the safe graceful-drain path. Legacy producers (payload - without the flag) wrote the file on-loop, so their staleness remains - proof and the old contract is unchanged. - - Never raises; any ambiguity (missing file, unreadable JSON, PID mismatch) - returns ``GATEWAY_LOOP_UNKNOWN``. + A stale heartbeat is ``wedged`` only when the payload declares the tick socket armed AND the + socket stays silent across ``tick_strikes`` consecutive misses; any answer is ``alive``; any + conflict or ambiguity is ``unknown`` so callers keep the graceful-drain path. """ try: stale_budget = max(float(stale_after), 0.0) @@ -619,13 +426,10 @@ def probe_gateway_loop_liveness( except Exception: return GATEWAY_LOOP_UNKNOWN if heartbeat_pid <= 0 or int(pid) <= 0 or heartbeat_pid != int(pid): - # No heartbeat for THIS process — old gateway version, still starting - # up, or a stale file from a previous PID. Not evidence of a wedge. + # Heartbeat is not this process's (old version, starting up, stale file): not evidence. return GATEWAY_LOOP_UNKNOWN - # Pick the right witness probe: TCP loopback (Windows / non-POSIX) - # takes priority if the producer published a port, otherwise fall back - # to the AF_UNIX socket (POSIX / legacy). + # TCP loopback witness (Windows) takes priority when published; else the AF_UNIX socket. tcp_port = payload.get("loop_tick_tcp_port") try: tcp_port_int = int(tcp_port) if tcp_port is not None else None @@ -639,37 +443,24 @@ def probe_gateway_loop_liveness( witness = _probe_loop_tick_socket(pid, home, timeout=tick_timeout) tick_armed = payload.get("loop_tick_socket", _LOOP_TICK_ABSENT) if witness is True: - # The loop answered a ping — it is dispatching right now. A stale - # heartbeat file is a stalled write or a saturated executor, not a - # wedge (#90502). + # Loop answered: a stale file is a stalled write, not a wedge. return GATEWAY_LOOP_ALIVE age = time.time() - mtime if age <= stale_budget: if witness is False: - # File fresh but the loop did not answer: an off-loop write can - # land after the loop froze, so a fresh file is not a liveness - # proof while the loop itself is silent. + # Fresh file but silent loop: an off-loop write can land after the loop froze. return GATEWAY_LOOP_UNKNOWN return GATEWAY_LOOP_ALIVE - # File is stale past the budget. The verdict now depends on what the - # producer promised about its witness: + # Stale past the budget; the verdict depends on what the producer promised about its witness. if tick_armed is _LOOP_TICK_ABSENT: - # Legacy producer: the write ran on-loop, so staleness really does - # prove the loop stopped scheduling — old contract, unchanged. + # Legacy on-loop writer: staleness proves the loop stopped scheduling. return GATEWAY_LOOP_WEDGED if tick_armed is not True: - # New producer whose witness could not be armed (bind failed): the - # write is off-loop, so staleness is NOT proof. Never escalate - # without a witness. + # Witness could not be armed (bind failed); off-loop write means staleness is not proof. return GATEWAY_LOOP_UNKNOWN if witness is False: - # First miss. One silent probe is NOT destructive authority: a - # short transient synchronous stall can outlast a single recv - # timeout, and killing a live gateway on it would be the exact false - # wedge #90502 exists to prevent. Require the loop to stay silent - # across the whole bounded window — the first probe above is miss - # #1, so ``tick_strikes - 1`` more attempts follow. + # First miss. The probe above is miss #1, so ``tick_strikes - 1`` more attempts follow. sustained = _probe_loop_tick_socket_sustained( pid, home, @@ -679,39 +470,21 @@ def probe_gateway_loop_liveness( tcp_port=tcp_port_int, ) if sustained is False: - # Both witnesses agree, sustained: the loop did not schedule for - # the entire window. return GATEWAY_LOOP_WEDGED if sustained is True: - # The loop answered on a later attempt: it was a transient - # stall, not a wedge — the stale file is a stalled write. + # Transient stall, not a wedge. return GATEWAY_LOOP_ALIVE - # Witness vanished mid-window: ambiguity — never kill on it. The - # graceful drain path remains the backstop. + # Witness vanished mid-window: ambiguity — never kill on it. return GATEWAY_LOOP_UNKNOWN - # Armed producer but the socket is unreachable: ambiguity — never kill on - # it. The graceful drain path remains the backstop. + # Armed but unreachable socket: ambiguity — never kill on it. return GATEWAY_LOOP_UNKNOWN -def _escalate_wedged_gateway( - pid: int, - *, - term_grace: float = 5.0, - kill_wait: float = 5.0, -) -> bool: - """Bounded stop for a gateway whose loop is provably dead (#81642). +def _escalate_wedged_gateway(pid: int, *, term_grace: float = 5.0, kill_wait: float = 5.0) -> bool: + """Bounded stop (SIGTERM, ``term_grace``, SIGKILL, ``kill_wait``) for a provably dead loop. - SIGTERM first (the process may still have a live signal handler thread - even with a dead loop), a short grace, then SIGKILL. Total worst case is - ``term_grace + kill_wait`` (~10s by default) — never the 180s drain - budget, which only makes sense when a loop exists to run the drain. - - Callers MUST have classified the gateway as ``GATEWAY_LOOP_WEDGED`` - before calling this: escalating a merely busy gateway would bypass the - in-flight cron drain floor (#86684) and SIGKILL live work. - - Returns True once the PID has left the process table. + Callers MUST have classified the gateway ``GATEWAY_LOOP_WEDGED`` first: escalating a merely + busy gateway bypasses the cron drain floor and SIGKILLs live work. True once the PID is gone. """ from gateway.status import get_process_start_time @@ -731,16 +504,9 @@ def _escalate_wedged_gateway( def _get_ancestor_pids() -> set[int]: - """Return the set of PIDs in the current process's ancestor chain. - - Walks from the current PID up to PID 1 (init) so that process-table scans - never match the calling CLI process or any of its parents. This prevents - ``hermes gateway status`` from falsely counting the ``hermes`` CLI that - invoked it as a running gateway instance (see #13242). - """ + """PIDs of this process and its ancestors, so scans never count the invoking ``hermes`` CLI as a gateway.""" ancestors: set[int] = set() pid = os.getpid() - # Cap iterations to avoid infinite loops on exotic platforms. for _ in range(64): ancestors.add(pid) parent = _get_parent_pid(pid) @@ -750,9 +516,7 @@ def _get_ancestor_pids() -> set[int]: return ancestors -def _append_unique_pid( - pids: list[int], pid: int | None, exclude_pids: set[int] -) -> None: +def _append_unique_pid(pids: list[int], pid: int | None, exclude_pids: set[int]) -> None: if pid is None or pid <= 0: return if pid == os.getpid() or pid in exclude_pids or pid in pids: @@ -765,34 +529,20 @@ def _scan_gateway_pids( all_profiles: bool = False, include_restart_managers: bool = False, ) -> list[int]: - """Best-effort process-table scan for gateway PIDs. - - This supplements the profile-scoped PID file so status views can still spot - a live gateway when the PID file is stale/missing, and ``--all`` sweeps can - discover gateways outside the current profile. - """ - # Exclude the entire ancestor chain so the CLI process that invoked this - # scan (e.g. ``hermes gateway status``) is never mistaken for a running - # gateway. See #13242. + """Best-effort process-table scan for gateway PIDs (backs up a stale/missing PID file; ``--all`` sweeps).""" exclude_pids = exclude_pids | _get_ancestor_pids() pids: list[int] = [] - # Strict command-line matcher shared with gateway.status: requires the - # actual ``gateway run`` subcommand (or the dedicated entrypoints), so this - # scan no longer false-matches ``gateway status``/``dashboard`` siblings or - # unrelated processes like ``python -m tui_gateway``. Lazy import mirrors the - # circular-import avoidance used elsewhere in this module. + # Strict matcher shared with gateway.status: requires a real ``gateway run`` argv, so + # ``gateway status``/``dashboard`` siblings and ``python -m tui_gateway`` don't match. from gateway.status import ( looks_like_gateway_command_line, looks_like_gateway_runtime_command_line, ) current_home = str(get_hermes_home().resolve()) - # Forward slashes on both sides of the HERMES_HOME= match — see - # gateway.status._command_line_belongs_to_profile, which this mirrors. + # Forward slashes on both sides of the HERMES_HOME= match (mirrors gateway.status). current_home_lc = current_home.lower().replace("\\", "/") current_profile_arg = _profile_arg(current_home) - current_profile_name = ( - current_profile_arg.split()[-1] if current_profile_arg else "" - ) + current_profile_name = (current_profile_arg.split()[-1] if current_profile_arg else "") current_profile_name_lc = current_profile_name.lower() def _matches_current_profile(command: str) -> bool: @@ -804,100 +554,35 @@ def _scan_gateway_pids( or f"hermes_home={current_home_lc}" in command_lc ) - # Default-profile case: no profile flag in argv. Accept as long as - # the command doesn't advertise *some other* profile. HERMES_HOME - # may be passed via env (not visible in wmic/CIM command line) so - # its absence is NOT disqualifying — only a non-matching explicit - # HERMES_HOME= in argv is. + # Default profile: accept unless argv advertises another profile. HERMES_HOME may come via + # env (invisible to wmic/CIM), so only a non-matching explicit HERMES_HOME= disqualifies. if "--profile " in command_lc or " -p " in command_lc: return False - if ( + return not ( "hermes_home=" in command_lc and f"hermes_home={current_home_lc}" not in command_lc - ): - return False - return True + ) def _matches_gateway_runtime(command: str) -> bool: if looks_like_gateway_command_line(command): return True return include_restart_managers and looks_like_gateway_runtime_command_line(command) + def _consider(pid: int, command: str) -> None: + if _matches_gateway_runtime(command) and ( + all_profiles or _matches_current_profile(command) + ): + _append_unique_pid(pids, pid, exclude_pids) + try: if is_windows(): - # Prefer wmic when present (fast, stable output format). On - # modern Windows 11 / Win 10 late builds, wmic has been - # removed as part of the WMIC deprecation — fall back to - # PowerShell's Get-CimInstance. A spawn failure or timeout - # (result is None) trips the fallback. - # The scans go through ``bounded_probe_run`` — NOT plain - # ``subprocess.run(timeout=...)`` — because on Windows ``run()``'s - # post-timeout cleanup joins the pipe reader threads unbounded; a - # descendant (conhost.exe) holding duplicated pipe handles then - # wedges the caller forever. ``hermes update`` hung exactly there - # on slow-WMI machines where the full Win32_Process scan exceeds - # its budget (#87134). - # bounded_probe_run also hides the console window: this scan runs - # inside the windowless pythonw.exe gateway/desktop backend, so a - # bare wmic/powershell spawn would flash a conhost window on every - # watchdog probe. - from hermes_cli._subprocess_compat import bounded_probe_run - - wmic_path = shutil.which("wmic") - result = None - if wmic_path is not None: - result = bounded_probe_run( - [ - wmic_path, - "process", - "get", - "ProcessId,CommandLine", - "/FORMAT:LIST", - ], - timeout=10, - errors="ignore", - ) - if result is None or result.returncode != 0 or not (result.stdout or ""): - # Fallback: PowerShell Get-CimInstance, emit LIST-style output - # so the downstream parser below doesn't need to branch. - powershell = shutil.which("powershell") or shutil.which("pwsh") - if powershell is None: - return [] - ps_cmd = ( - "Get-CimInstance Win32_Process | " - "ForEach-Object { " - " 'CommandLine=' + ($_.CommandLine -replace \"`r`n\",' ' -replace \"`n\",' '); " - " 'ProcessId=' + $_.ProcessId; " - " '' " - "}" - ) - result = bounded_probe_run( - [powershell, "-NoProfile", "-Command", ps_cmd], - timeout=15, - errors="ignore", - ) - if result is None: - return [] - if result.returncode != 0 or result.stdout is None: + listing = _windows_process_listing() + if listing is None: return [] - current_cmd = "" - for line in result.stdout.split("\n"): - line = line.strip() - if line.startswith("CommandLine="): - current_cmd = line[len("CommandLine=") :] - elif line.startswith("ProcessId="): - pid_str = line[len("ProcessId=") :] - if _matches_gateway_runtime(current_cmd) and ( - all_profiles or _matches_current_profile(current_cmd) - ): - try: - _append_unique_pid(pids, int(pid_str), exclude_pids) - except ValueError: - pass - current_cmd = "" + for pid, command in _iter_windows_list_processes(listing): + _consider(pid, command) else: - # Try /proc first (works in Docker without procps installed), - # fall back to `ps -Aww` (BSD-safe; see below). + # /proc first (Docker without procps), then `ps -Aww`. _found_via_proc = False if os.path.isdir("/proc"): try: @@ -911,11 +596,7 @@ def _scan_gateway_pids( try: with open(f"/proc/{pid}/cmdline", "rb") as _f: cmdline = _f.read().decode("utf-8", errors="replace") - cmdline = cmdline.replace("\x00", " ") - if _matches_gateway_runtime(cmdline) and ( - all_profiles or _matches_current_profile(cmdline) - ): - _append_unique_pid(pids, pid, exclude_pids) + _consider(pid, cmdline.replace("\x00", " ")) except (OSError, PermissionError): continue _found_via_proc = True @@ -923,82 +604,103 @@ def _scan_gateway_pids( pass if not _found_via_proc: - result = subprocess.run( - # ``-Aww`` (not ``-A eww``): the BSD ``e`` flag (show - # environment) is illegal on macOS/BSD ps and makes the - # whole command fail with rc 1, silently returning [] on - # every macOS machine (#73626). The matcher only needs - # argv, not env vars, so ``e`` is unnecessary. ``-ww`` - # keeps unlimited-width output on both BSD and procps ps. - ["ps", "-Aww", "-o", "pid=,command="], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=10, - ) + # ``-Aww`` not ``-A eww``: BSD/macOS ps rejects ``e``; ``-ww`` = unlimited width. + result = subprocess.run(["ps", "-Aww", "-o", "pid=,command="], timeout=10, **_CAPTURE_TEXT) if result.returncode != 0: return [] for line in result.stdout.split("\n"): - stripped = line.strip() - if not stripped or "grep" in stripped: - continue - - pid = None - command = "" - - parts = stripped.split(None, 1) - if len(parts) == 2: - try: - pid = int(parts[0]) - command = parts[1] - except ValueError: - pid = None - - if pid is None: - aux_parts = stripped.split() - if len(aux_parts) > 10 and aux_parts[1].isdigit(): - pid = int(aux_parts[1]) - command = " ".join(aux_parts[10:]) - - if pid is None: - continue - if _matches_gateway_runtime(command) and ( - all_profiles or _matches_current_profile(command) - ): - _append_unique_pid(pids, pid, exclude_pids) + parsed = _parse_ps_line(line) + if parsed is not None: + _consider(*parsed) except (OSError, subprocess.TimeoutExpired): return [] - # Windows-specific: collapse venv launcher stubs. A venv-built - # ``pythonw.exe`` in ``/Scripts/`` is a ~100 KB launcher exe - # that spawns the base Python (e.g. ``C:\Program Files\Python311\ - # pythonw.exe``) with the same command line, preserving the venv's - # ``pyvenv.cfg`` context. This is standard Windows CPython venv - # behaviour — BUT it means every gateway run produces two pythonw - # PIDs with identical command lines (one launcher stub, one actual - # interpreter) which is confusing in ``gateway status`` output. - # Filter the stub: if a PID in our result is the PARENT of another - # PID in our result, and both are pythonw.exe, the parent is the - # launcher stub — drop it, keep the child. + # Windows: a venv ``pythonw.exe`` is a launcher stub that spawns the base Python with the same + # command line, so each gateway yields two matched PIDs. Drop a matched PID that parents another. if is_windows() and len(pids) > 1: pids = _filter_venv_launcher_stubs(pids) return pids -def _filter_venv_launcher_stubs(pids: list[int]) -> list[int]: - """Drop venv-launcher ``pythonw.exe`` stubs that are parents of the real - interpreter process. See comment at the tail of ``_scan_gateway_pids``. +def _parse_ps_line(line: str) -> tuple[int, str] | None: + """``(pid, command)`` from one ``ps -o pid=,command=`` line; also accepts ``ps aux`` rows.""" + stripped = line.strip() + if not stripped or "grep" in stripped: + return None + parts = stripped.split(None, 1) + if len(parts) == 2: + with contextlib.suppress(ValueError): + return int(parts[0]), parts[1] + aux_parts = stripped.split() + if len(aux_parts) > 10 and aux_parts[1].isdigit(): + return int(aux_parts[1]), " ".join(aux_parts[10:]) + return None - Uses ``psutil`` (core dependency). Safe on any platform; only invoked - on Windows by the caller because the stub pattern is Windows-specific. + +def _iter_windows_list_processes(listing: str): + """Yield ``(pid, command_line)`` from wmic/CIM ``/FORMAT:LIST`` output.""" + current_cmd = "" + for line in listing.split("\n"): + line = line.strip() + if line.startswith("CommandLine="): + current_cmd = line[len("CommandLine=") :] + elif line.startswith("ProcessId="): + with contextlib.suppress(ValueError): + yield int(line[len("ProcessId=") :]), current_cmd + current_cmd = "" + + +def _windows_process_listing() -> str | None: + """``CommandLine=``/``ProcessId=`` LIST output for every Windows process, or None. + + wmic when present, else Get-CimInstance emitting the same shape. ``bounded_probe_run``, NOT + ``subprocess.run(timeout=...)``: on Windows run()'s post-timeout cleanup joins pipe readers + unbounded and a conhost.exe holding duplicated handles wedges the caller forever; it also + hides the console window this windowless pythonw backend would otherwise flash. """ + from hermes_cli._subprocess_compat import bounded_probe_run + + wmic_path = shutil.which("wmic") + result = None + if wmic_path is not None: + result = bounded_probe_run( + [wmic_path, "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"], + timeout=10, + errors="ignore", + ) + if result is None or result.returncode != 0 or not (result.stdout or ""): + powershell = shutil.which("powershell") or shutil.which("pwsh") + if powershell is None: + return None + ps_cmd = ( + "Get-CimInstance Win32_Process | " + "ForEach-Object { " + " 'CommandLine=' + ($_.CommandLine -replace \"`r`n\",' ' -replace \"`n\",' '); " + " 'ProcessId=' + $_.ProcessId; " + " '' " + "}" + ) + result = bounded_probe_run( + [powershell, "-NoProfile", "-Command", ps_cmd], + timeout=15, + errors="ignore", + ) + if result is None: + return None + if result.returncode != 0 or result.stdout is None: + return None + return result.stdout + + +def _filter_venv_launcher_stubs(pids: list[int]) -> list[int]: + """Drop venv-launcher ``pythonw.exe`` stubs that parent another matched PID (see ``_scan_gateway_pids``).""" try: import psutil # type: ignore except ImportError: return pids pid_set = set(pids) - # Collect each PID's parent so we can flag "child of another matched PID". parent_of: dict[int, int | None] = {} for pid in pids: try: @@ -1006,7 +708,6 @@ def _filter_venv_launcher_stubs(pids: list[int]) -> list[int]: except (psutil.NoSuchProcess, psutil.AccessDenied): parent_of[pid] = None - # For each child whose parent is also in our set, drop the parent. drop: set[int] = set() for pid, ppid in parent_of.items(): if ppid is not None and ppid in pid_set: @@ -1015,20 +716,8 @@ def _filter_venv_launcher_stubs(pids: list[int]) -> list[int]: return [p for p in pids if p not in drop] -def find_gateway_pids( - exclude_pids: set | None = None, all_profiles: bool = False -) -> list: - """Find PIDs of running gateway processes. - - Args: - exclude_pids: PIDs to exclude from the result (e.g. service-managed - PIDs that should not be killed during a stale-process sweep). - all_profiles: When ``True``, return gateway PIDs across **all** - profiles (the pre-7923 global behaviour). ``hermes update`` - needs this because a code update affects every profile. - When ``False`` (default), only PIDs belonging to the current - Hermes profile are returned. - """ +def find_gateway_pids(exclude_pids: set | None = None, all_profiles: bool = False) -> list: + """Find running gateway PIDs for the current profile, or every profile with ``all_profiles`` (``hermes update``).""" _exclude = set(exclude_pids or set()) pids: list[int] = [] if not all_profiles: @@ -1087,20 +776,13 @@ def find_profile_gateway_processes( create_time = 0.0 except Exception as exc: if strict: - raise RuntimeError( - f"Could not inspect gateway PID for profile {profile.name}" - ) from exc + raise RuntimeError(f"Could not inspect gateway PID for profile {profile.name}") from exc continue if pid is None or pid <= 0 or pid in _exclude or pid in seen: continue seen.add(pid) processes.append( - ProfileGatewayProcess( - profile=profile.name, - path=profile.path, - pid=pid, - create_time=create_time, - ) + ProfileGatewayProcess(profile=profile.name, path=profile.path, pid=pid, create_time=create_time) ) return processes @@ -1112,12 +794,9 @@ def find_windows_gateway_services( ) -> list[WindowsGatewayService]: """Find profile gateways supervised by real Windows services. - Service-logon processes can deny the interactive Desktop access to their - command lines. The updater can still identify them without guessing: a - validated profile gateway PID comes from Hermes's own PID file, and its - parent chain terminates at a running SCM service PID. The complete service - subtree is returned so the Desktop preflight exempts only processes the CLI - updater will stop through the Service Control Manager. + Service-logon processes may hide their command lines, so identity comes from Hermes's own PID + file plus a parent chain ending at a running SCM service PID. The whole service subtree is + returned so the Desktop preflight exempts exactly what the updater stops through the SCM. """ if sys.platform != "win32": return [] @@ -1130,10 +809,7 @@ def find_windows_gateway_services( indeterminate_services_by_pid: dict[int, list[tuple[str, object]]] = {} for service in psutil_module.win_service_iter(): try: - if all( - callable(getattr(service, field, None)) - for field in ("name", "status", "pid") - ): + if all(callable(getattr(service, field, None)) for field in ("name", "status", "pid")): service_name = str(service.name() or "") service_status = service.status() service_pid = int(service.pid() or 0) @@ -1143,8 +819,7 @@ def find_windows_gateway_services( service_status = data.get("status") service_pid = int(data.get("pid") or 0) except FileNotFoundError: - # The service was deleted between enumeration and inspection; - # it cannot still supervise a live gateway tree. + # Deleted between enumeration and inspection. continue except Exception as exc: raise RuntimeError("SCM service inspection failed") from exc @@ -1159,9 +834,7 @@ def find_windows_gateway_services( ) continue if service_pid <= 0: - raise RuntimeError( - f"Running SCM service {service_name} has no valid process ID" - ) + raise RuntimeError(f"Running SCM service {service_name} has no valid process ID") service_names_by_pid.setdefault(service_pid, set()).add(service_name) except Exception as exc: raise RuntimeError("SCM service enumeration failed") from exc @@ -1195,11 +868,7 @@ def find_windows_gateway_services( + ", ".join(str(pid) for pid in shared_service_pids) ) service_pid = next( - ( - pid - for pid in ancestor_pids - if len(service_names_by_pid.get(pid, set())) == 1 - ), + (pid for pid in ancestor_pids if len(service_names_by_pid.get(pid, set())) == 1), None, ) if service_pid is None: @@ -1212,10 +881,7 @@ def find_windows_gateway_services( if int(profile_process.pid) not in descendants: continue descendant_identities = tuple( - sorted( - (int(child.pid), float(child.create_time())) - for child in descendant_processes - ) + sorted((int(child.pid), float(child.create_time())) for child in descendant_processes) ) found[service_name] = WindowsGatewayService( name=service_name, @@ -1246,16 +912,9 @@ def _gateway_run_args_for_profile(profile: str) -> list[str]: def _capture_gateway_argv(pid: int) -> list[str] | None: - """Return the live argv of a running gateway process, or ``None``. + """Live argv of a running gateway (snapshotted before update kills so unmapped gateways can respawn). - Used to respawn gateways that have no profile→PID-file mapping (e.g. a - Windows Scheduled Task running ``pythonw.exe -m hermes_cli.main gateway - run``). ``_pause_windows_gateways_for_update`` force-kills such gateways - before mutating the venv; without their original command line we cannot - bring them back, so we snapshot it here before the kill. - - Best-effort: returns ``None`` if psutil is unavailable, the process is - gone, access is denied, or the argv doesn't look like a gateway command. + ``None`` if psutil is unavailable, the process is gone/denied, or the argv isn't a gateway command. """ if pid <= 1: return None @@ -1271,9 +930,7 @@ def _capture_gateway_argv(pid: int) -> list[str] | None: return None if not argv: return None - # Guard against snapshotting an unrelated process whose PID happened to be - # reported by the scan: only respawn things that actually look like a - # gateway run command line. + # Never respawn an unrelated process the scan happened to report. try: from gateway.status import looks_like_gateway_command_line @@ -1287,20 +944,9 @@ def _capture_gateway_argv(pid: int) -> list[str] | None: def _prepare_profile_gateway_update_restart(profile: str, pid: int) -> str | None: """Choose who relaunches a profile gateway after ``hermes update``. - A gateway started with ``--external-supervisor`` must exit back to that - manager. Starting Hermes's detached watcher as well would escape the - manager and race its replacement process. Ordinary foreground gateways - retain the existing detached-watcher behavior. - - When the profile-derived relaunch cannot be armed -- typically because - ``_gateway_run_args_for_profile`` cannot rebuild a run argv for this - profile -- fall back to replaying the process's own captured command - line, which is what ``launch_detached_gateway_restart_by_cmdline`` - exists for and what the Windows post-update path already does for its - unmapped gateways. Without this the caller has no way to relaunch the - process and (before #88654) silently left it running pre-update modules - against post-update code on disk. ``argv`` is already captured above, - so the fallback costs nothing extra. + ``--external-supervisor`` gateways must exit back to their manager (a detached watcher would + race its replacement). Otherwise arm the profile-derived detached watcher, falling back to + replaying the captured command line. """ argv = _capture_gateway_argv(pid) if argv and "--external-supervisor" in argv: @@ -1312,17 +958,8 @@ def _prepare_profile_gateway_update_restart(profile: str, pid: int) -> str | Non return None -def launch_detached_gateway_restart_by_cmdline( - old_pid: int, run_argv: list[str] -) -> bool: - """Relaunch a gateway by replaying its captured command line after exit. - - Companion to ``launch_detached_profile_gateway_restart`` for gateways that - have no profile→PID-file mapping (Scheduled-Task / manually-launched - ``gateway run`` whose HERMES_HOME or argv doesn't match a known profile). - Uses the identical detached-watcher mechanism; only the respawn argv - differs (the process's own argv instead of a profile-derived one). - """ +def launch_detached_gateway_restart_by_cmdline(old_pid: int, run_argv: list[str]) -> bool: + """Relaunch a gateway with no profile→PID-file mapping by replaying its captured argv after exit.""" if old_pid <= 0 or not run_argv: return False return _spawn_gateway_restart_watcher(old_pid, list(run_argv)) @@ -1340,57 +977,31 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool: if old_pid <= 0 or not run_argv: return False - # The watcher is a tiny Python subprocess that polls the old PID and - # respawns the gateway once it's gone. Both legs of the chain need - # platform-appropriate detach semantics: - # - # POSIX — ``start_new_session=True`` (os.setsid in the child) detaches - # from the parent's process group so Ctrl+C in the CLI doesn't - # propagate and the watcher/gateway survive the CLI exiting. - # - # Windows — ``start_new_session`` is silently accepted but does NOT - # detach. The watcher stays attached to the CLI's console and dies - # when the user closes the terminal, leaving ``hermes update`` users - # with no running gateway until they re-invoke ``hermes gateway`` - # manually. The Win32 equivalent is the ``CREATE_NEW_PROCESS_GROUP | - # DETACHED_PROCESS | CREATE_NO_WINDOW`` creationflags bundle. - # - # ``windows_detach_popen_kwargs()`` returns the right kwargs for the - # host platform and is a no-op on POSIX (just ``start_new_session=True``). + # Both watcher and respawned gateway need platform-appropriate detach: POSIX + # ``start_new_session=True`` (setsid); on Windows that flag does NOT detach (the watcher would + # die with the CLI console), so ``windows_detach_popen_kwargs()`` supplies the creationflags. from hermes_cli._subprocess_compat import ( windows_detach_flags_without_breakaway, windows_detach_popen_kwargs, ) - # On Windows the incoming ``run_argv`` leads with the venv's console - # ``python.exe`` (from ``get_python_path()``). That's the interpreter we - # want: the watcher respawns it under CREATE_NO_WINDOW detach flags, so - # the gateway owns one hidden console that all descendants inherit — - # nothing flashes (#54220/#56747). The spec helper normalizes the - # interpreter and captures the stable cwd + env overlay (HERMES_HOME, - # VIRTUAL_ENV, PYTHONPATH) so the respawn doesn't depend on the watcher's - # transient working directory. No-op on POSIX. - # See gateway_windows.windowless_gateway_restart_spec. + # Windows: normalize the interpreter and capture a stable cwd + env overlay (HERMES_HOME, + # VIRTUAL_ENV, PYTHONPATH) so the respawn doesn't depend on the watcher's cwd. No-op on POSIX. respawn_cwd = "" respawn_env_overlay: dict[str, str] = {} if sys.platform == "win32": try: - from hermes_cli.gateway_windows import ( - windowless_gateway_restart_spec, - ) + from hermes_cli.gateway_windows import (windowless_gateway_restart_spec) run_argv, respawn_cwd, respawn_env_overlay = ( windowless_gateway_restart_spec(list(run_argv)) ) except Exception: - # Best-effort: if the rewrite fails for any reason, fall back to - # the original argv. A visible window is worse than nothing, but - # a failed respawn is worse still — keep the gateway coming back. + # Fall back to the original argv: a visible window beats a failed respawn. respawn_cwd = "" respawn_env_overlay = {} - # Serialized as JSON literals embedded in the watcher source so the - # inner respawn can apply cwd= / env= without extra argv plumbing. + # Embedded as JSON literals in the watcher source (no extra argv plumbing). respawn_cwd_literal = json.dumps(respawn_cwd) respawn_env_literal = json.dumps(respawn_env_overlay) @@ -1499,16 +1110,9 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool: respawn_env_literal=respawn_env_literal, ) - watcher_argv = [ - sys.executable, - "-c", - watcher, - str(old_pid), - *run_argv, - ] + watcher_argv = [sys.executable, "-c", watcher, str(old_pid), *run_argv] - # Same platform-aware detach for the watcher process itself — so - # closing the user's terminal doesn't kill the watcher. + # Same detach for the watcher itself, so closing the terminal doesn't kill it. try: subprocess.Popen( watcher_argv, @@ -1517,11 +1121,8 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool: **windows_detach_popen_kwargs(), ) except OSError: - # CREATE_BREAKAWAY_FROM_JOB rejected by the parent job object - # (Electron, Windows Terminal with restrictive job settings, …). - # Retry without it. POSIX never reaches this branch — there - # ``start_new_session=True`` cannot raise OSError — so the - # fallback is only meaningful on Windows. + # Parent job object rejected CREATE_BREAKAWAY_FROM_JOB; retry without it (Windows only — + # ``start_new_session=True`` cannot raise OSError on POSIX). try: fallback_kwargs: dict = ( {"creationflags": windows_detach_flags_without_breakaway()} @@ -1539,45 +1140,41 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool: return True +def _systemd_unit_is_active(system: bool) -> bool: + """``systemctl is-active`` == "active" for the installed unit in ``system`` scope, else False.""" + if not get_systemd_unit_path(system=system).exists(): + return False + try: + result = _run_systemctl(["is-active", get_service_name()], system=system, timeout=10, **_CAPTURE_TEXT) + except (RuntimeError, subprocess.TimeoutExpired): + return False + return result.stdout.strip() == "active" + + def _probe_systemd_service_running(system: bool = False) -> tuple[bool, bool]: selected_system = _select_systemd_scope(system) - unit_exists = get_systemd_unit_path(system=selected_system).exists() - if not unit_exists: - return selected_system, False - try: - result = _run_systemctl( - ["is-active", get_service_name()], - system=selected_system, - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=10, - ) - except (RuntimeError, subprocess.TimeoutExpired): - return selected_system, False - return selected_system, result.stdout.strip() == "active" + return selected_system, _systemd_unit_is_active(selected_system) def _read_systemd_unit_environment(system: bool = False) -> dict[str, str]: - """Parse the gateway unit's ``Environment=`` directives. + """Parse ``systemctl show -p Environment`` (one line of unquoted space-separated KEY=VALUE pairs).""" + body = _systemctl_show(("Environment",), system=system).get("Environment", "") + parsed: dict[str, str] = {} + for token in body.split(): + if "=" in token: + key, value = token.split("=", 1) + parsed[key] = value + return parsed - ``systemctl show -p Environment`` returns a single line of - space-separated ``KEY=VALUE`` pairs; values are not quoted in the output - even when the unit file quoted them. We split on whitespace and ``=``. - """ - selected_system = _select_systemd_scope(system) + +def _systemctl_show(properties: tuple[str, ...], *, system: bool) -> dict[str, str]: + """``systemctl show --property a,b`` for the gateway unit as ``{key: value}``; {} on failure.""" try: result = _run_systemctl( - [ - "show", - get_service_name(), - "--no-pager", - "--property", - "Environment", - ], - system=selected_system, - capture_output=True, - text=True, encoding='utf-8', errors='replace', + ["show", get_service_name(), "--no-pager", "--property", ",".join(properties)], + system=_select_systemd_scope(system), timeout=10, + **_CAPTURE_TEXT, ) except (RuntimeError, subprocess.TimeoutExpired, OSError): return {} @@ -1585,25 +1182,14 @@ def _read_systemd_unit_environment(system: bool = False) -> dict[str, str]: return {} parsed: dict[str, str] = {} for line in result.stdout.splitlines(): - if not line.startswith("Environment="): - continue - body = line[len("Environment=") :].strip() - for token in body.split(): - if "=" not in token: - continue - key, value = token.split("=", 1) - parsed[key] = value + if "=" in line: + key, value = line.split("=", 1) + parsed[key] = value.strip() return parsed def _hermes_home_from_systemd_unit_file(system: bool = False) -> str | None: - """Read ``HERMES_HOME`` from the on-disk unit file (not ``systemctl show``). - - Prefer the file when refreshing/comparing: under ``sudo``, ``systemctl`` - may be slow/unavailable in tests, and the on-disk unit is what - ``systemd_unit_is_current`` / ``refresh_systemd_unit_if_needed`` already - compare against. - """ + """``HERMES_HOME`` from the on-disk unit file — what refresh/compare already read, and reliable under ``sudo``.""" unit_path = get_systemd_unit_path(system=system) if not unit_path.exists(): return None @@ -1623,18 +1209,14 @@ def _hermes_home_from_systemd_unit_file(system: bool = False) -> str | None: def _sync_hermes_home_from_systemd_unit(system: bool) -> None: - """When acting on a system-scope unit, adopt its ``HERMES_HOME``. + """For a system-scope unit, adopt its ``HERMES_HOME``. - Under ``sudo``, ``HERMES_HOME`` is stripped and ``HOME=/root``, so - :func:`get_hermes_home` falls back to ``/root/.hermes`` — the wrong - profile. The unit file pins ``HERMES_HOME`` for the actual gateway - process, so we mirror that into our own environment to make - ``read_runtime_status`` / ``get_running_pid`` read the correct files. + Under ``sudo`` HERMES_HOME is stripped and HOME=/root, so get_hermes_home() would pick the + wrong profile; mirroring the unit's value makes runtime-status/PID reads hit the right files. """ if not system: return - # Prefer the on-disk unit (source of truth for refresh/compare). Fall - # back to ``systemctl show`` for units that only exist in the manager. + # On-disk unit first; ``systemctl show`` for units that only exist in the manager. unit_home = (_hermes_home_from_systemd_unit_file(system=True) or "").strip() if not unit_home: unit_home = _read_systemd_unit_environment(system=True).get("HERMES_HOME", "").strip() @@ -1648,43 +1230,10 @@ def _sync_hermes_home_from_systemd_unit(system: bool) -> None: def _read_systemd_unit_properties( system: bool = False, - properties: tuple[str, ...] = ( - "ActiveState", - "SubState", - "Result", - "ExecMainStatus", - "MainPID", - ), + properties: tuple[str, ...] = ("ActiveState", "SubState", "Result", "ExecMainStatus", "MainPID"), ) -> dict[str, str]: """Return selected ``systemctl show`` properties for the gateway unit.""" - selected_system = _select_systemd_scope(system) - try: - result = _run_systemctl( - [ - "show", - get_service_name(), - "--no-pager", - "--property", - ",".join(properties), - ], - system=selected_system, - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=10, - ) - except (RuntimeError, subprocess.TimeoutExpired, OSError): - return {} - - if result.returncode != 0: - return {} - - parsed: dict[str, str] = {} - for line in result.stdout.splitlines(): - if "=" not in line: - continue - key, value = line.split("=", 1) - parsed[key] = value.strip() - return parsed + return _systemctl_show(properties, system=system) def _systemd_main_pid_from_props(props: dict[str, str]) -> int | None: @@ -1769,27 +1318,24 @@ def _wait_for_systemd_service_restart( ): replacement_observed.append(True) - if active_state == "active": - if new_pid and (previous_pid is None or new_pid != previous_pid): - if runtime_pid != new_pid: - runtime_state = _gateway_runtime_status_for_pid(new_pid) - gateway_state = (runtime_state or {}).get("gateway_state") - if gateway_state == "running": - print(f"✓ {scope_label} service restarted (PID {new_pid})") - return True - if gateway_state == "startup_failed": - reason = (runtime_state or {}).get( - "exit_reason" - ) or "startup failed" - print( - f"⚠ {scope_label} service process restarted (PID {new_pid}), but gateway startup failed: {reason}" - ) - return False - if not printed_runtime_wait: - print( - f"⏳ {scope_label} service process started (PID {new_pid}); waiting for gateway runtime..." - ) - printed_runtime_wait = True + if active_state == "active" and new_pid and (previous_pid is None or new_pid != previous_pid): + if runtime_pid != new_pid: + runtime_state = _gateway_runtime_status_for_pid(new_pid) + gateway_state = (runtime_state or {}).get("gateway_state") + if gateway_state == "running": + print(f"✓ {scope_label} service restarted (PID {new_pid})") + return True + if gateway_state == "startup_failed": + reason = (runtime_state or {}).get("exit_reason") or "startup failed" + print( + f"⚠ {scope_label} service process restarted (PID {new_pid}), but gateway startup failed: {reason}" + ) + return False + if not printed_runtime_wait: + print( + f"⏳ {scope_label} service process started (PID {new_pid}); waiting for gateway runtime..." + ) + printed_runtime_wait = True if active_state == "activating" and sub_state == "auto-restart": time.sleep(1) @@ -1813,16 +1359,11 @@ def _systemd_restart_wait_timeout(system: bool = False) -> float: """Cover systemd's relaunch delays before applying the runtime wait floor.""" from gateway.shutdown_forensics import parse_systemd_duration_to_us - props = _read_systemd_unit_properties( - system=system, - properties=("RestartUSec", "TimeoutStartUSec"), - ) + props = _read_systemd_unit_properties(system=system, properties=("RestartUSec", "TimeoutStartUSec")) supervisor_budget = 0.0 for name in ("RestartUSec", "TimeoutStartUSec"): raw = props.get(name, "") - duration_us = ( - int(raw) if raw.isdigit() else parse_systemd_duration_to_us(raw) - ) + duration_us = (int(raw) if raw.isdigit() else parse_systemd_duration_to_us(raw)) if duration_us is not None: supervisor_budget += duration_us / 1_000_000 return 60.0 + supervisor_budget @@ -1870,9 +1411,7 @@ def _print_systemd_start_limit_wait(system: bool = False) -> None: print(f" Check logs: {journal_prefix}-u {svc} -l --since '5 min ago'") -def _recover_pending_systemd_restart( - system: bool = False, previous_pid: int | None = None -) -> bool: +def _recover_pending_systemd_restart(system: bool = False, previous_pid: int | None = None) -> bool: """Recover a planned service restart that is stuck in systemd state.""" props = _read_systemd_unit_properties(system=system) if not props: @@ -1894,10 +1433,7 @@ def _recover_pending_systemd_restart( if active_state == "activating" and sub_state == "auto-restart": print("⏳ Service restart already pending — waiting for systemd relaunch...") - return _wait_for_systemd_service_restart( - system=system, - previous_pid=previous_pid, - ) + return _wait_for_systemd_service_restart(system=system, previous_pid=previous_pid) if active_state == "failed" and ( exec_main_status == str(GATEWAY_SERVICE_RESTART_EXIT_CODE) @@ -1905,38 +1441,17 @@ def _recover_pending_systemd_restart( ): svc = get_service_name() scope_label = _service_scope_label(system).capitalize() - print( - f"↻ Clearing failed state for pending {scope_label.lower()} service restart..." - ) - _run_systemctl( - ["reset-failed", svc], - system=system, - check=False, - timeout=30, - ) - _run_systemctl( - ["start", svc], - system=system, - check=False, - timeout=90, - ) - return _wait_for_systemd_service_restart( - system=system, - previous_pid=previous_pid, - ) + print(f"↻ Clearing failed state for pending {scope_label.lower()} service restart...") + _run_systemctl(["reset-failed", svc], system=system, check=False, timeout=30) + _run_systemctl(["start", svc], system=system, check=False, timeout=90) + return _wait_for_systemd_service_restart(system=system, previous_pid=previous_pid) return False def _parse_launchd_pid_from_list_output(output: str) -> int | None: - """Extract the PID from ``launchctl list