From e8d734565ca1ee8d839f8b95d957267a63f3f4fb Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:35:26 -0700 Subject: [PATCH] =?UTF-8?q?refactor(gateway):=20status.py=20=E2=80=94=20co?= =?UTF-8?q?mpact=20docstrings=20by=20hand,=20fold=20small=20guards?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gateway/session.py | 3 +- gateway/status.py | 395 ++++++++++++++++----------------------------- 2 files changed, 138 insertions(+), 260 deletions(-) diff --git a/gateway/session.py b/gateway/session.py index e7ffd9bdb8..39d85029e8 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -590,8 +590,7 @@ class SessionEntry: return cls( session_key=session_key, session_id=session_id, created_at=datetime.fromisoformat(data["created_at"]), - updated_at=datetime.fromisoformat(data["updated_at"]), - origin=origin, + updated_at=datetime.fromisoformat(data["updated_at"]), origin=origin, display_name=data.get("display_name"), platform=platform, chat_type=data.get("chat_type", "dm"), metadata=dict(data.get("metadata") or {}), last_resume_marked_at=_parse_iso(data.get("last_resume_marked_at")), diff --git a/gateway/status.py b/gateway/status.py index dae7b18fe7..e834428e52 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -1,9 +1,5 @@ -"""Gateway runtime status helpers. - -PID-file based detection of whether the gateway daemon is running (used by -send_message's check_fn to gate CLI availability). The PID file lives at -``{HERMES_HOME}/gateway.pid``, so separate homes/profiles get separate files. -""" +"""Gateway runtime status helpers: PID/lock/marker files under ``{HERMES_HOME}`` (one set per +home/profile) that tell whether the gateway daemon is running.""" import contextlib import copy @@ -62,10 +58,7 @@ def record_start_and_check_storm( max_starts: int = 5, window_s: float = 120.0, *, backoff_cap_s: float = 300.0 ) -> Optional[StormInfo]: """Record this start; :class:`StormInfo` when > ``max_starts`` landed in ``window_s``. - - Best-effort: a broken ``gateway-starts.log`` ledger (distinct from ``restart_loop.json``) - is logged and swallowed so it can never crash gateway startup. - """ + Best-effort: a broken ``gateway-starts.log`` ledger is logged and swallowed, never fatal.""" try: path = get_hermes_home() / "gateway-starts.log" path.parent.mkdir(parents=True, exist_ok=True) @@ -92,11 +85,9 @@ def record_start_and_check_storm( def _get_process_hermes_home() -> Path: - """Process-level HERMES_HOME for identity files (PID, lock, status, markers). - - ``get_hermes_home()`` honors the per-session ``_HERMES_HOME_OVERRIDE`` contextvar - and would misroute them; identity files must live in the launch home. - """ + """Launch-home HERMES_HOME for identity files (PID, lock, status, markers): + ``get_hermes_home()`` honors the per-session ``_HERMES_HOME_OVERRIDE`` and would misroute + them.""" val = os.environ.get("HERMES_HOME", "").strip() return Path(val) if val else _get_platform_default_hermes_home() @@ -115,14 +106,10 @@ def _same_hermes_home(left: Path | str, right: Path | str) -> bool: def recorded_gateway_home_conflicts( record: Optional[dict[str, Any]], *, expected_home: Optional[Path | str] = None ) -> bool: - """True when a persisted gateway record names a DIFFERENT HERMES_HOME. - - Cross-profile kill guard: a contaminated PID record can truthfully name another profile's - live gateway; destructive callers must refuse or profile B's stop SIGTERMs profile A and both - supervisors restart-loop. ``expected_home`` overrides the comparison base (``profile delete``). - Legacy records without ``hermes_home`` return False (they prove nothing; callers pair this with - PID + start-time guards). A comparison failure returns True: unprovable ownership fails closed. - """ + """True when a persisted gateway record names a DIFFERENT HERMES_HOME (cross-profile kill guard: + profile B's stop must never SIGTERM profile A). ``expected_home`` overrides the comparison base. + Legacy records without ``hermes_home`` prove nothing -> False; a comparison failure fails + closed -> True.""" recorded_home = record.get("hermes_home") if isinstance(record, dict) else None if not isinstance(recorded_home, str) or not recorded_home.strip(): return False @@ -156,11 +143,8 @@ def _profile_label_for_home(home: Path | str) -> Optional[str]: def scoped_lock_owner_label(record: Optional[dict[str, Any]]) -> Optional[str]: - """Profile label for the gateway owning a (machine-global) scoped lock (None: PID-only wording). - - Prefers the ``profile`` field stamped by :func:`acquire_scoped_lock` (validated: it flows into - log lines and a suggested CLI command), then infers from ``hermes_home`` for older locks. - """ + """Profile label of a scoped-lock owner (None: PID-only wording): the validated ``profile`` + field stamped by :func:`acquire_scoped_lock`, else inferred from ``hermes_home`` (old locks).""" if not isinstance(record, dict): return None profile = record.get("profile") @@ -175,9 +159,7 @@ def _get_pid_path() -> Path: def _get_gateway_lock_path(pid_path: Optional[Path] = None) -> Path: - if pid_path is not None: - return pid_path.with_name(_GATEWAY_LOCK_FILENAME) - return _get_process_hermes_home() / _GATEWAY_LOCK_FILENAME + return (pid_path or _get_pid_path()).with_name(_GATEWAY_LOCK_FILENAME) def _get_runtime_status_path() -> Path: @@ -202,13 +184,10 @@ _EPOCH_MIN_PLAUSIBLE = 946684800.0 # 2000-01-01T00:00:00Z def normalize_updated_at(value: Any) -> Optional[str]: - """Coerce a persisted ``updated_at`` value to an RFC3339 string or ``None``. - - ``/api/status`` and ``/health/detailed`` promise ``string | null``, but the file may hold - legacy epoch floats, hand edits, or corruption. ``str``: iff fromisoformat parses (trailing - ``Z`` tolerated; naive -> UTC). ``int``/``float``: epoch seconds; before 2000-01-01, > 1 day - ahead, or non-finite -> None. ``bool`` / other -> None. - """ + """Coerce a persisted ``updated_at`` (ISO string, legacy epoch, hand edit, garbage) to the + RFC3339 ``string | null`` that ``/api/status`` promises. ``str``: iff fromisoformat parses + (trailing ``Z`` tolerated; naive -> UTC). Epoch: before 2000-01-01, > 1 day ahead or + non-finite -> None. ``bool``/other -> None.""" if isinstance(value, str): raw = value.strip() # Python < 3.11 fromisoformat rejects a trailing 'Z'; tolerate it. @@ -234,12 +213,9 @@ def normalize_updated_at(value: Any) -> Optional[str]: def terminate_pid( pid: int, *, force: bool = False, expected_start_time: Optional[float] = None ) -> None: - """Terminate a PID; POSIX SIGTERM/SIGKILL, Windows taskkill /T /F for force. - - Identity guard: on Windows ``force=True`` REQUIRES a matching ``expected_start_time`` - (taskkill /T /F on a recycled PID has killed svchost.exe). On POSIX it is optional, but - a provided, mismatched fingerprint refuses the kill everywhere -- the PID was recycled. - """ + """Terminate a PID; POSIX SIGTERM/SIGKILL, Windows taskkill /T /F for force. Identity guard: + Windows ``force`` REQUIRES a matching ``expected_start_time`` (taskkill on a recycled PID has + killed svchost.exe); POSIX optional, but a provided mismatch refuses the kill everywhere.""" if force and (_IS_WINDOWS or expected_start_time is not None): if expected_start_time is None: raise OSError(f"refusing to force-kill PID {pid} without a process start-time guard") @@ -286,12 +262,9 @@ def _get_scope_lock_path(scope: str, identity: str) -> Path: def _get_process_start_time(pid: int) -> Optional[int]: - """Stable per-process start-time fingerprint (PID-reuse guard), or None. - - Linux: field 22 of ``/proc//stat`` (clock ticks since boot). Without ``/proc`` - (macOS/Windows): psutil ``create_time()`` quantized to centiseconds for stable - equality. Units differ per platform; the guard only compares same-host values. - """ + """Per-process start-time fingerprint (PID-reuse guard), or None: ``/proc//stat`` field 22 + on Linux, else psutil ``create_time()`` in centiseconds. Units differ per platform; the guard + only compares same-host values.""" with contextlib.suppress(IndexError, ValueError, OSError): return int(Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[21]) try: @@ -331,15 +304,11 @@ def _read_process_cmdline(pid: int) -> Optional[str]: def _gateway_command_subcommand(command: str | None) -> str | None: - """Hermes gateway lifecycle subcommand from a command line, or None. - - No loose substring matches: ``"gateway" in cmdline`` also matched ``gateway status`` and - ``python -m tui_gateway`` (restart() raced a draining process; false-positive status). - Requires a Hermes entrypoint plus the ``gateway`` subcommand, or a gateway-dedicated - entrypoint. Tokenizes quote-aware (Windows paths with spaces) and strips ``--profile``/ - ``-p`` selectors anywhere in argv -- ``_apply_profile_override`` removes them before - argparse, so they (and a profile literally named ``gateway``) can appear on either side. - """ + """Hermes gateway lifecycle subcommand from a command line, or None. No loose substring matches + (``"gateway" in cmdline`` also matched ``gateway status`` / ``python -m tui_gateway``): needs a + Hermes entrypoint plus the ``gateway`` subcommand, or a gateway-dedicated entrypoint. Tokenizes + quote-aware (Windows paths with spaces); ``--profile``/``-p`` selectors are stripped anywhere in + argv since ``_apply_profile_override`` removes them before argparse.""" if not command: return None try: @@ -384,12 +353,9 @@ def looks_like_gateway_command_line(command: str | None) -> bool: def looks_like_gateway_runtime_command_line(command: str | None) -> bool: - """True for command lines that can host the gateway runtime (``run`` or ``restart``). - - Without a service manager the manual restart fallback runs ``run_gateway()`` in-process, so - argv stays ``gateway restart`` while it owns the runtime. Use only for validating Hermes-owned - records / cleanup scans; ``looks_like_gateway_command_line()`` stays strict. - """ + """True for command lines that can host the runtime (``run`` or ``restart``: without a service + manager the manual restart fallback runs ``run_gateway()`` in-process). For validating + Hermes-owned records / cleanup scans only; ``looks_like_gateway_command_line`` stays strict.""" return _gateway_command_subcommand(command) in {"run", "restart"} @@ -413,23 +379,18 @@ def _profile_name_for_home(profile_home: Path) -> Optional[str]: def _command_line_belongs_to_profile(command: str, profile_home: Path) -> bool: - """True when a gateway command line belongs to ``profile_home``. - - Mirrors ``hermes_cli.gateway._matches_current_profile``: a stale state file can record a PID - recycled onto a DIFFERENT profile's live gateway (still a gateway, so the dead profile would - read running). Named profiles carry ``-p``/``--profile `` or ``HERMES_HOME=`` on argv; - the default gateway runs bare. Separators are normalized (Windows backslashes vs argv slashes). - """ + """True when a gateway command line belongs to ``profile_home`` (mirrors + ``hermes_cli.gateway._matches_current_profile``): a stale state file can record a PID recycled + onto ANOTHER profile's live gateway. Named profiles carry ``-p``/``--profile `` or + ``HERMES_HOME=`` on argv; the default gateway runs bare. Separators normalized.""" command_lc = command.lower().replace("\\", "/") profile_name = _profile_name_for_home(profile_home) home_lc = str(profile_home).lower().replace("\\", "/") if profile_name is not None and profile_name != "default": profile_lc = profile_name.lower() - return ( - f"--profile {profile_lc}" in command_lc - or f"-p {profile_lc}" in command_lc - or f"hermes_home={home_lc}" in command_lc - ) + return any(needle in command_lc for needle in ( + f"--profile {profile_lc}", f"-p {profile_lc}", f"hermes_home={home_lc}" + )) # Default profile: accept unless argv names another profile or a conflicting explicit # HERMES_HOME= (its absence is not disqualifying -- HERMES_HOME usually arrives via the env). if "--profile " in command_lc or " -p " in command_lc: @@ -440,12 +401,9 @@ def _command_line_belongs_to_profile(command: str, profile_home: Path) -> bool: def _record_matches_live_gateway_pid( record: dict[str, Any], pid: int, *, expected_home: Optional[Path] = None ) -> bool: - """True when a live PID still identifies as this gateway record. - - Prefer the live command line: a stale record's argv must not make an unrelated process (PID - reuse) count as a gateway; with ``expected_home`` it must also belong to that profile. When the - command line is unreadable (Windows/permission), fall back to the persisted record. - """ + """True when a live PID still identifies as this gateway record. The live command line wins (a + stale record's argv must not make a recycled PID count as a gateway; with ``expected_home`` it + must also belong to that profile); unreadable cmdline (Windows/EACCES) -> persisted record.""" live_cmdline = _read_process_cmdline(pid) if not live_cmdline: return _record_looks_like_gateway(record) @@ -468,9 +426,7 @@ def _build_pid_record() -> dict: def _get_code_identity_fields() -> dict[str, Any]: """Code identity of THIS process for ``gateway_state.json`` (restart picked up new code?). - - Lazy import keeps ``gateway.status`` free of ``hermes_cli`` at import time. Never raises. - """ + Lazy import keeps ``gateway.status`` free of ``hermes_cli`` at import time. Never raises.""" try: from hermes_cli.build_info import get_code_identity identity = get_code_identity() @@ -480,11 +436,8 @@ def _get_code_identity_fields() -> dict[str, Any]: def _pid_record_belongs_to_current_profile(record: Optional[dict[str, Any]]) -> bool: - """True when the record's ``hermes_home`` matches the current process (legacy records: True). - - A record from a different HERMES_HOME belongs to another profile and must be ignored, or - the default gateway assumes that profile's identity. - """ + """True when the record's ``hermes_home`` matches the current process (legacy records: True); + another HERMES_HOME's record must be ignored or the default gateway assumes its identity.""" if not isinstance(record, dict): return False record_home = record.get("hermes_home") @@ -506,10 +459,8 @@ def _build_runtime_status_record() -> dict[str, Any]: def _read_json_file(path: Path, *, bare_pid_ok: bool = False) -> Optional[dict[str, Any]]: - """JSON object at ``path``, or None when absent/empty/unreadable/invalid. - - ``bare_pid_ok`` also accepts legacy bare-integer PID files as ``{"pid": N}``. - """ + """JSON object at ``path``, or None when absent/empty/unreadable/invalid. ``bare_pid_ok`` also + accepts legacy bare-integer PID files as ``{"pid": N}``.""" try: raw = path.read_text(encoding="utf-8").strip() if path.exists() else "" except (OSError, UnicodeDecodeError): # vanished, EACCES, non-UTF-8 garbage @@ -519,12 +470,10 @@ def _read_json_file(path: Path, *, bare_pid_ok: bool = False) -> Optional[dict[s try: payload = json.loads(raw) except json.JSONDecodeError: - if not bare_pid_ok: - return None - try: - return {"pid": int(raw)} - except ValueError: - return None + payload = None + if bare_pid_ok: + with contextlib.suppress(ValueError): + payload = int(raw) if bare_pid_ok and isinstance(payload, int): return {"pid": payload} return payload if isinstance(payload, dict) else None @@ -609,14 +558,10 @@ def _try_acquire_file_lock(handle) -> bool: def _pid_exists(pid: int) -> bool: - """Cross-platform "is this PID alive" check that does NOT kill the target. - - CRITICAL on Windows: ``os.kill(pid, 0)`` is NOT a no-op -- CPython maps ``sig=0`` to - ``CTRL_C_EVENT`` for the target's whole console group (bpo-14484). Prefer psutil; fall back - to ctypes ``OpenProcess``/``WaitForSingleObject`` on Windows and ``os.kill(pid, 0)`` on POSIX. - Zombies are reported dead: treating one as alive makes --replace wait forever under systemd - Restart=always, which respawns before reaping. - """ + """Cross-platform "is this PID alive" check that does NOT kill the target. CRITICAL on Windows: + ``os.kill(pid, 0)`` sends ``CTRL_C_EVENT`` to the whole console group (bpo-14484), so prefer + psutil, then ctypes ``OpenProcess`` (Windows) / ``os.kill(pid, 0)`` (POSIX). Zombies report + dead: treating one as alive makes --replace wait forever under systemd Restart=always.""" pid = int(pid) try: import psutil # type: ignore @@ -709,8 +654,8 @@ def acquire_gateway_runtime_lock() -> bool: try: handle = open(path, "a+", encoding="utf-8") except PermissionError: - # Stale root-owned lock from a launchd Background session that ran as root. - # The directory owner can unlink it; retry once with a fresh file. + # Stale root-owned lock (launchd session that ran as root): the directory owner can + # unlink it; retry once with a fresh file. try: path.unlink() handle = open(path, "a+", encoding="utf-8") @@ -743,11 +688,8 @@ def release_gateway_runtime_lock() -> None: def owns_gateway_runtime_lock() -> bool: - """True when THIS process holds the runtime lock. - - ``is_gateway_runtime_lock_active`` answers "does anyone hold it?"; re-probing our own - flock succeeds on POSIX, so the in-process handle is the only self-ownership discriminator. - """ + """True when THIS process holds the runtime lock. ``is_gateway_runtime_lock_active`` answers + "does anyone?"; re-probing our own flock succeeds on POSIX, so the handle is the discriminator.""" return _gateway_lock_handle is not None @@ -773,10 +715,8 @@ def is_gateway_runtime_lock_active(lock_path: Optional[Path] = None) -> bool: try: handle = open(resolved_lock_path, "a+", encoding="utf-8") except PermissionError: - # Stale root-owned lock (launchd session that ran as root): the directory - # owner can unlink it; report inactive so a fresh one is made. - with contextlib.suppress(OSError): - resolved_lock_path.unlink() + # Stale root-owned lock (see acquire_gateway_runtime_lock): report inactive. + _unlink_quietly(resolved_lock_path) return False return _probe_lock_file(handle) @@ -807,10 +747,10 @@ def _is_gateway_runtime_lock_active_strict(lock_path: Path) -> bool: def write_pid_file() -> None: - """Write this process's PID record via O_CREAT|O_EXCL; concurrent racers get FileExistsError.""" + """Write this process's PID record via O_CREAT|O_EXCL; a racing gateway's FileExistsError + propagates for the caller to decide.""" path = _get_pid_path() path.parent.mkdir(parents=True, exist_ok=True) - # FileExistsError propagates: another gateway is racing us; caller decides. _write_json_excl(path, _build_pid_record()) _clear_running_pid_cache() @@ -834,10 +774,8 @@ def _apply_set_fields(target: dict[str, Any], fields) -> None: def _coerce_session_store(session_store: Any) -> dict[str, str]: - state = "unknown" - if isinstance(session_store, dict): - state = str(session_store.get("status") or "unknown") - return {"status": state if state in {"ok", "unavailable", "retrying", "unknown"} else "unknown"} + state = str(session_store.get("status") or "") if isinstance(session_store, dict) else "" + return {"status": state if state in {"ok", "unavailable", "retrying"} else "unknown"} def write_runtime_status( @@ -888,10 +826,8 @@ def write_runtime_status( )) # Per-entry writer provenance: top-level pid/start_time only identify the most recent # writer; /api/status tells "live" from "preserved" by exact (pid, start_time) equality. - platform_payload.update( - updated_at=_utc_now_iso(), writer_pid=current_record["pid"], - writer_start_time=current_record["start_time"], - ) + platform_payload.update(updated_at=_utc_now_iso(), writer_pid=current_record["pid"], + writer_start_time=current_record["start_time"]) payload["platforms"][platform] = platform_payload _write_json_file(path, payload) with contextlib.suppress(Exception): @@ -934,11 +870,8 @@ _DRAINABLE_GATEWAY_STATES = frozenset({"running"}) def derive_gateway_busy(*, gateway_running: bool, gateway_state: Any, active_agents: Any) -> bool: - """Busy iff live, ``running``, and ``active_agents > 0`` -- the contract NAS gates on. - - Liveness keys off ``gateway_running``, NEVER ``updated_at`` -- a healthy idle gateway never - advances that timestamp. - """ + """Busy iff live, ``running``, and ``active_agents > 0`` -- the contract NAS gates on. Liveness + keys off ``gateway_running``, NEVER ``updated_at`` (an idle gateway never advances it).""" if not derive_gateway_drainable(gateway_running=gateway_running, gateway_state=gateway_state): return False try: @@ -948,20 +881,15 @@ def derive_gateway_busy(*, gateway_running: bool, gateway_state: Any, active_age def derive_gateway_drainable(*, gateway_running: bool, gateway_state: Any) -> bool: - """Drainable iff live and ``running``. - - Independent of ``active_agents``: an idle drain completes at once. - """ + """Drainable iff live and ``running`` (independent of ``active_agents``: idle drains complete).""" return bool(gateway_running) and gateway_state in _DRAINABLE_GATEWAY_STATES @dataclass(frozen=True) class GatewayLiveness: - """Resolved gateway liveness for one dashboard surface. - - ``source``: which ladder rung answered (logging/tests only -- never branch product behavior - on it). ``probe_error``: a rung raised; lets fail-open callers tell "down" from "unknown". - """ + """Resolved gateway liveness for one dashboard surface. ``source``: which ladder rung answered + (logging/tests only -- never branch product behavior on it). ``probe_error``: a rung raised; + lets fail-open callers tell "down" from "unknown".""" running: bool pid: Optional[int] @@ -977,17 +905,13 @@ def resolve_gateway_liveness( runtime_reader: Optional[Callable[..., Optional[dict[str, Any]]]] = None, runtime_pid_probe: Optional[Callable[..., Optional[int]]] = None, ) -> GatewayLiveness: - """Single source of truth for "is the gateway up?" across dashboard surfaces. - - Ladder, most to least authoritative: (1) PID file + runtime lock (scoped to ``profile_dir``; - cached by default so polling does not re-flock ``gateway.lock``); (2) caller-supplied HTTP - health probe (gateway in another container); (3) runtime status PID validated against the live - process table with ``expected_home`` so a recycled PID of another profile never counts. Rung 3 - only uses the LOCAL state record (the probe body's PID belongs to another host); pass - ``runtime`` if already read. The ``*_probe``/``runtime_reader`` kwargs are the dashboard's - injection/test seam. A rung that raises degrades to the next (never 500 a status endpoint) and - sets ``probe_error``. - """ + """Single source of truth for "is the gateway up?" across dashboard surfaces. Ladder, most to + least authoritative: (1) PID file + runtime lock (scoped to ``profile_dir``; cached by default + so polling does not re-flock ``gateway.lock``); (2) caller-supplied HTTP health probe (gateway + in another container); (3) LOCAL runtime status PID validated against the live process table + with ``expected_home`` (a recycled PID of another profile never counts; pass ``runtime`` if + already read). ``*_probe``/``runtime_reader`` are the dashboard's injection/test seam. A rung + that raises degrades to the next (never 500 a status endpoint) and sets ``probe_error``.""" _pid_probe = pid_probe or (get_running_pid_cached if use_cache else get_running_pid) _runtime_reader = runtime_reader or read_runtime_status _runtime_pid_probe = runtime_pid_probe or get_runtime_status_running_pid @@ -1033,13 +957,10 @@ def resolve_gateway_liveness( def get_runtime_status_running_pid( runtime: Optional[dict[str, Any]] = None, *, expected_home: Optional[Path] = None ) -> Optional[int]: - """Live gateway PID from the runtime status record, or None. - - Conservative fallback to ``get_running_pid()`` for launch-service-managed gateways with a - fresh ``gateway_state.json`` but no ``gateway.pid``. ``expected_home`` scopes the OS-identity - check to another profile's home so a PID recycled onto a different profile's gateway is not - reported running for the dead one. - """ + """Live gateway PID from the runtime status record, or None: the ``get_running_pid()`` fallback + for launch-service-managed gateways with a fresh ``gateway_state.json`` but no ``gateway.pid``. + ``expected_home`` scopes the OS-identity check to another profile's home so a PID recycled onto + a different profile's gateway is not reported running for the dead one.""" payload = runtime if runtime is not None else read_runtime_status() if not isinstance(payload, dict): return None @@ -1048,8 +969,8 @@ def get_runtime_status_running_pid( pid = _live_pid_from_record(payload) if pid is None: return None - # Active-profile context: the record's hermes_home must match this process - # so a stale record cannot lend another profile's identity. + # Active-profile context: the record's hermes_home must match this process so a stale record + # cannot lend another profile's identity. if expected_home is None and not _pid_record_belongs_to_current_profile(payload): return None if not _record_matches_live_gateway_pid(payload, pid, expected_home=expected_home): @@ -1058,11 +979,8 @@ def get_runtime_status_running_pid( def remove_pid_file() -> None: - """Remove the PID file only if it belongs to this process. - - During --replace the old process's atexit can fire AFTER the new process wrote its own - record; blind removal would leave the gateway invisible. - """ + """Remove the PID file only if it belongs to this process: during --replace the old process's + atexit can fire AFTER the new process wrote its own record.""" with contextlib.suppress(Exception): path = _get_pid_path() file_pid = _pid_from_record(_read_json_file(path)) @@ -1073,14 +991,12 @@ def remove_pid_file() -> None: def _scoped_lock_record_is_stale(existing: dict[str, Any], existing_pid: Optional[int]) -> bool: - """True when a foreign scoped-lock record no longer names a live gateway. - - Stale when the PID is missing/dead, its start time changed (PID reuse), or the live process - is not a gateway: a readable cmdline says so (also catches boot-time PID+start_time collisions - -- systemd spawns deterministically). Cmdline unreadable (Windows has no ps) AND start_time - unavailable on either side => consult the lock record's own argv, the only identity signal - left. Stopped processes (Ctrl+Z / SIGTSTP) look alive to _pid_exists; stale so --replace works. - """ + """True when a foreign scoped-lock record no longer names a live gateway: PID missing/dead, + start time changed (PID reuse), or the live process is not a gateway -- a readable cmdline says + so (also catches boot-time PID+start_time collisions; systemd spawns deterministically); + cmdline unreadable AND start_time unknown on either side => the lock record's own argv is the + only signal left. Stopped (SIGTSTP) processes look alive to _pid_exists; stale so --replace + works.""" if existing_pid is None or not _pid_exists(existing_pid): return True recorded_start = existing.get("start_time") @@ -1090,8 +1006,7 @@ def _scoped_lock_record_is_stale(existing: dict[str, Any], existing_pid: Optiona if not _looks_like_gateway_process(existing_pid): if _read_process_cmdline(existing_pid) is not None: return True - start_unknown = recorded_start is None or current_start is None - if start_unknown and not _record_looks_like_gateway(existing): + if None in (recorded_start, current_start) and not _record_looks_like_gateway(existing): return True return _process_is_stopped(existing_pid) @@ -1151,10 +1066,8 @@ def acquire_scoped_lock( def release_scoped_lock(scope: str, identity: str) -> None: - """Release a scope lock owned by this PID. - - No start_time equality check: on-disk null vs a live fingerprint would wedge reconnects. - """ + """Release a scope lock owned by this PID. No start_time equality check: on-disk null vs a live + fingerprint would wedge reconnects.""" lock_path = _get_scope_lock_path(scope, identity) if (_read_json_file(lock_path) or {}).get("pid") == os.getpid(): _unlink_quietly(lock_path) @@ -1163,11 +1076,8 @@ def release_scoped_lock(scope: str, identity: str) -> None: def release_all_scoped_locks( *, owner_pid: Optional[int] = None, owner_start_time: Optional[int] = None ) -> int: - """Remove scoped lock files (--replace cleanup); returns the count removed. - - With ``owner_pid`` only that gateway's records go (``owner_start_time`` narrows against - PID reuse); with no owner every lock file is removed. - """ + """Remove scoped lock files (--replace cleanup); returns the count removed. With ``owner_pid`` + only that gateway's records go (``owner_start_time`` narrows against PID reuse).""" lock_dir = _get_lock_dir() if not lock_dir.exists(): return 0 @@ -1217,11 +1127,8 @@ def _marker_is_stale(written_at: str, ttl_s: int) -> bool: def _read_live_pid_marker(path: Path, ttl_s: int) -> Optional[tuple[dict[str, Any], int, Any]]: - """``(record, target_pid, target_start_time)`` for a usable marker, else None. - - Malformed/expired markers can never match anyone, so they are unlinked here (a stale file - left by a previous instance must not wedge a new one). - """ + """``(record, target_pid, target_start_time)`` for a usable marker, else None. Malformed/expired + markers can never match anyone, so they are unlinked here (must not wedge a new instance).""" record = _read_json_file(path) if not record: return None @@ -1236,13 +1143,11 @@ def _read_live_pid_marker(path: Path, ttl_s: int) -> Optional[tuple[dict[str, An def _pid_marker_names_self(target_pid: int, target_start_time: Any) -> bool: - """PID match with an optional start-time PID-reuse guard (shared by watcher probe and consume). - - ``_get_process_start_time`` returns None without /proc (macOS, native Windows -- where the - planned-stop watcher matters most); requiring a match there would misclassify a legitimate - ``hermes gateway stop`` as an unexpected exit (revived by the service manager). So: both start - times known -> must match; either unknown -> PID equality decides (bounded by the marker TTL). - """ + """PID match with an optional start-time PID-reuse guard (watcher probe + consume). Both start + times known -> must match; either unknown -> PID equality decides (bounded by the marker TTL): + ``_get_process_start_time`` is None without /proc (macOS, native Windows -- where the + planned-stop watcher matters most) and requiring a match there would misclassify a legitimate + ``hermes gateway stop`` as an unexpected exit revived by the service manager.""" if target_pid != os.getpid(): return False our_start_time = _get_process_start_time(target_pid) @@ -1274,13 +1179,10 @@ def _consume_pid_marker_for_self(path: Path, *, ttl_s: int) -> bool: def write_takeover_marker( target_pid: int, *, target_home: Optional[Path] = None, target_start_time: Any = _UNSET ) -> bool: - """Record that ``target_pid`` is being replaced by this process; True on success. - - Captures the target's ``start_time`` (PID-reuse guard) and a timestamp for TTL checks. A - verified cross-home handoff passes ``target_home`` + validated ``target_start_time`` so the - marker lands in the target's home; such callers must fail closed on False (the target's - supervisor could otherwise revive it). - """ + """Record that ``target_pid`` is being replaced by this process; True on success. Captures the + target's ``start_time`` (PID-reuse guard) + a timestamp for TTL. A verified cross-home handoff + passes ``target_home`` + validated ``target_start_time`` so the marker lands in the target's + home; such callers must fail closed on False (the target's supervisor could revive it).""" try: marker_home = _canonical_hermes_home(target_home or _get_process_hermes_home()) if target_start_time is _UNSET: @@ -1317,11 +1219,9 @@ def clear_takeover_marker(target_home: Optional[Path] = None) -> None: def _validated_scoped_lock_gateway_owner(record: dict[str, Any]) -> Optional[tuple[int, int, Path]]: - """Resolve a live scoped-lock owner to a verified ``(pid, start_time, home)``. - - A lock file is only a claim: the record, the target home's PID record, and the live process - must agree on PID, start-time, gateway identity, and home. Missing legacy metadata fails closed. - """ + """Resolve a live scoped-lock owner to a verified ``(pid, start_time, home)``. A lock file is + only a claim: the record, the target home's PID record, and the live process must agree on + PID, start-time, gateway identity, and home. Missing legacy metadata fails closed.""" if not isinstance(record, dict) or not _record_looks_like_gateway(record): return None owner_pid = _pid_from_record(record) @@ -1380,16 +1280,13 @@ def _wait_for_scoped_lock_owner_exit( def _snapshot_gateway_children(pid: int) -> list: - """Best-effort snapshot of ``pid``'s live descendants (POSIX only; never raises). - - Take it while the parent is alive -- once it exits the children are reparented and - undiscoverable. ``[]`` on Windows (taskkill /T tree-kills). - """ + """Best-effort snapshot of ``pid``'s live descendants (POSIX only; never raises). Take it while + the parent is alive -- once it exits the children are reparented and undiscoverable. ``[]`` on + Windows (taskkill /T tree-kills).""" if _IS_WINDOWS: return [] try: import psutil # type: ignore - return psutil.Process(int(pid)).children(recursive=True) except Exception: logger.debug("Could not snapshot children of gateway PID %d", pid, exc_info=True) @@ -1397,20 +1294,17 @@ def _snapshot_gateway_children(pid: int) -> list: def reap_gateway_children(children: list, *, parent_pid: int, timeout: float = 5.0) -> int: - """Best-effort reap of a dead gateway's orphaned descendants (POSIX); returns count signalled. - - Surviving adapter subprocesses keep holding token locks. Call only AFTER the parent is confirmed - dead, with a snapshot from :func:`_snapshot_gateway_children`. ``is_running()`` is identity- - aware so a recycled child PID is never signalled; a child whose ppid still equals ``parent_pid`` - is skipped (parent alive => not an orphan). SIGTERM, bounded wait, SIGKILL survivors. Never - raises. - """ + """Best-effort reap of a dead gateway's orphaned descendants (POSIX; surviving adapter + subprocesses keep holding token locks); returns count signalled. Call only AFTER the parent is + confirmed dead, with a :func:`_snapshot_gateway_children` snapshot. ``is_running()`` is + identity-aware so a recycled child PID is never signalled; a child whose ppid still equals + ``parent_pid`` is skipped (parent alive => not an orphan). SIGTERM, bounded wait, SIGKILL + survivors. Never raises.""" if _IS_WINDOWS or not children: return 0 reaped = 0 try: import psutil # type: ignore - live = [] for child in children: try: @@ -1449,23 +1343,19 @@ def reap_gateway_children(children: list, *, parent_pid: int, timeout: float = 5 def take_over_scoped_lock_holder( record: dict[str, Any], *, graceful_attempts: int = 20, force_attempts: int = 20 ) -> Optional[int]: - """Terminate one verified scoped-lock holder for explicit ``--replace``. - - Returns the owner PID only after that exact PID/start-time identity exited; validation or - marker-write failure returns None without signalling. A cross-home handoff must place a - consumable marker in the target's home or its supervisor could revive it (flap loop). On POSIX - the owner's snapshotted children are then reaped. - """ + """Terminate one verified scoped-lock holder for explicit ``--replace``. Returns the owner PID + only after that exact PID/start-time identity exited; validation or marker-write failure returns + None without signalling (a cross-home handoff must place a consumable marker in the target's + home or its supervisor revives it: flap loop). On POSIX the snapshotted children are reaped.""" owner = _validated_scoped_lock_gateway_owner(record) if owner is None: return None owner_pid, owner_start_time, target_home = owner # Snapshot while the owner is alive; afterwards children are reparented. owner_children = _snapshot_gateway_children(owner_pid) - marker_ok = write_takeover_marker( + if not write_takeover_marker( owner_pid, target_home=target_home, target_start_time=owner_start_time - ) - if not marker_ok: + ): return None try: replaced = _terminate_verified_owner( @@ -1483,11 +1373,8 @@ def take_over_scoped_lock_holder( def _terminate_verified_owner( owner_pid: int, owner_start_time: int, *, graceful_attempts: int, force_attempts: int ) -> Optional[int]: - """Bounded identity-aware SIGTERM-then-SIGKILL of a verified owner. - - Returns the PID once it exited, else None. Each signal step: ``ProcessLookupError`` => - already gone; any other ``OSError`` => refuse without escalating. - """ + """Bounded identity-aware SIGTERM-then-SIGKILL of a verified owner; the PID once it exited, else + None. Per signal step: ``ProcessLookupError`` => already gone; other ``OSError`` => refuse.""" state = _scoped_lock_owner_state(owner_pid, owner_start_time) if state == "exited": return owner_pid @@ -1514,11 +1401,9 @@ def _terminate_verified_owner( def write_planned_stop_marker(target_pid: int) -> bool: - """Record that ``target_pid`` is being stopped intentionally. - - Unexpected SIGTERM exits non-zero so service managers revive the gateway; the CLI writes - this marker first so a deliberate stop exits cleanly. - """ + """Record that ``target_pid`` is being stopped intentionally: unexpected SIGTERM exits non-zero + so service managers revive the gateway; the CLI writes this first so a deliberate stop exits + cleanly.""" return _write_marker(_get_planned_stop_marker_path(), { "target_pid": target_pid, "target_start_time": _get_process_start_time(target_pid), @@ -1535,12 +1420,9 @@ def consume_planned_stop_marker_for_self() -> bool: def planned_stop_marker_targets_self() -> bool: - """Non-destructive probe for the watcher thread: True when a live planned-stop marker names us. - - Unlike :func:`consume_planned_stop_marker_for_self` it never unlinks a matching marker (the - shutdown handler does the authoritative consume). Malformed/expired markers are still cleaned - up; markers naming another PID are left alone and report False. - """ + """Non-destructive watcher probe: True when a live planned-stop marker names us. Never unlinks a + matching marker (the shutdown handler does the authoritative consume); malformed/expired ones + are still cleaned up; markers naming another PID are left alone.""" parsed = _read_live_pid_marker(_get_planned_stop_marker_path(), _PLANNED_STOP_MARKER_TTL_S) return parsed is not None and _pid_marker_names_self(parsed[1], parsed[2]) @@ -1623,11 +1505,8 @@ def get_running_pid_cached( pid_path: Optional[Path] = None, *, cleanup_stale: bool = True, ttl_seconds: float = _GATEWAY_RUNNING_PID_CACHE_TTL_SECONDS, ) -> Optional[int]: - """Cached ``get_running_pid()`` for high-frequency dashboard polling. - - Short TTL, invalidated on PID/lock/runtime-status file changes, so status endpoints do not - re-flock ``gateway.lock`` hundreds of times a minute. - """ + """Cached ``get_running_pid()`` for dashboard polling: short TTL, invalidated on PID/lock/ + runtime-status file changes, so status endpoints do not re-flock ``gateway.lock`` constantly.""" if ttl_seconds <= 0: return get_running_pid(pid_path, cleanup_stale=cleanup_stale) resolved_pid_path = pid_path or _get_pid_path()