5a0225dfff
The detached watcher runs as `<python> -c <program>`, so `hermes_cli` resolved only because update_completion happened to spawn it with cwd=<checkout>. The program now puts the checkout on sys.path itself, and the bare-Python test runs it from an unrelated cwd (red without the sys.path line). Also drops the two unused re-export aliases in gateway.status (`_posix_is_zombie`, `_pid_exists_win32_ctypes`): nothing imports them and neither is in the old-updater compat surface.
2194 lines
104 KiB
Python
2194 lines
104 KiB
Python
"""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 asyncio
|
||
import contextlib
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import math
|
||
import os
|
||
import re
|
||
import shlex
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Callable, NamedTuple, Optional
|
||
|
||
from hermes_constants import _get_platform_default_hermes_home, get_hermes_home, get_process_hermes_home
|
||
from hermes_cli._subprocess_compat import pid_exists_stdlib
|
||
from utils import atomic_json_write
|
||
|
||
if sys.platform == "win32":
|
||
import msvcrt
|
||
else:
|
||
import fcntl
|
||
|
||
_GATEWAY_KIND = "hermes-gateway"
|
||
_RUNTIME_STATUS_FILE = "gateway_state.json"
|
||
_LOCKS_DIRNAME = "gateway-locks"
|
||
_IS_WINDOWS = sys.platform == "win32"
|
||
_UNSET = object()
|
||
_GATEWAY_LOCK_FILENAME = "gateway.lock"
|
||
_gateway_lock_handle = None
|
||
# Windows byte-range locks are mandatory for other readers: lock a byte well past
|
||
# the JSON payload so status/PID readers can read while another process holds it.
|
||
_WINDOWS_LOCK_OFFSET = 1024 * 1024
|
||
_GATEWAY_RUNNING_PID_CACHE_TTL_SECONDS = 1.0
|
||
_gateway_running_pid_cache_lock = threading.Lock()
|
||
# key: (pid_path, cleanup_stale, include_runtime_status) -> (cached_at, file signature, pid)
|
||
_gateway_running_pid_cache: dict[tuple[str, bool, bool], tuple[float, tuple, Optional[int]]] = {}
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class _RuntimeStatusWriter:
|
||
"""Persist the latest complete status snapshot on one daemon thread.
|
||
|
||
Runtime status is diagnostic state, not an event log. While one write is
|
||
blocked in filesystem I/O, newer submissions replace the single pending
|
||
snapshot. This bounds memory and keeps every status producer off asyncio.
|
||
"""
|
||
|
||
def __init__(self, write_fn: Optional[Callable[[Path, dict[str, Any]], None]] = None):
|
||
self._write_fn = write_fn
|
||
self._condition = threading.Condition()
|
||
self._pending: Optional[tuple[int, Path, dict[str, Any]]] = None
|
||
self._writing_generation = 0
|
||
self._submitted_generation = 0
|
||
self._completed_generation = 0
|
||
self._successful_generation = 0
|
||
self._last_error: Optional[BaseException] = None
|
||
self._failure_logged = False
|
||
self._thread: Optional[threading.Thread] = None
|
||
|
||
def submit(self, path: Path, payload: dict[str, Any]) -> int:
|
||
with self._condition:
|
||
self._submitted_generation += 1
|
||
generation = self._submitted_generation
|
||
self._pending = (generation, path, copy.deepcopy(payload))
|
||
if self._thread is None or not self._thread.is_alive():
|
||
self._thread = threading.Thread(
|
||
target=self._run, daemon=True, name="gateway-runtime-status-writer")
|
||
self._thread.start()
|
||
self._condition.notify_all()
|
||
return generation
|
||
|
||
def wait(self, generation: int, timeout: Optional[float] = None) -> bool:
|
||
"""Block until ``generation`` (or a later snapshot) is persisted; ``False`` on timeout/failure."""
|
||
deadline = None if timeout is None else time.monotonic() + max(timeout, 0.0)
|
||
with self._condition:
|
||
while (state := self.settled(generation)) is None:
|
||
if deadline is None:
|
||
self._condition.wait()
|
||
continue
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
return False
|
||
self._condition.wait(timeout=remaining)
|
||
if not state and timeout is None and self._last_error is not None:
|
||
raise self._last_error
|
||
return state
|
||
|
||
def flush(self, timeout: float = 2.0) -> bool:
|
||
with self._condition:
|
||
generation = self._submitted_generation
|
||
return generation == 0 or self.wait(generation, timeout=timeout)
|
||
|
||
def settled(self, generation: int) -> Optional[bool]:
|
||
"""``True`` once ``generation`` persisted, ``False`` once it can no longer, else ``None``."""
|
||
with self._condition:
|
||
if self._successful_generation >= generation:
|
||
return True
|
||
no_more_work = (
|
||
self._completed_generation >= generation
|
||
and self._writing_generation == 0
|
||
and self._pending is None
|
||
)
|
||
return False if no_more_work else None
|
||
|
||
def _run(self) -> None:
|
||
while True:
|
||
with self._condition:
|
||
while self._pending is None:
|
||
self._condition.wait()
|
||
generation, path, payload = self._pending
|
||
self._pending = None
|
||
self._writing_generation = generation
|
||
error: Optional[BaseException] = None
|
||
try:
|
||
(self._write_fn or _write_json_file)(path, _merge_over_on_disk(path, payload))
|
||
except BaseException as exc:
|
||
error = exc
|
||
with self._condition:
|
||
self._writing_generation = 0
|
||
self._completed_generation = max(self._completed_generation, generation)
|
||
if error is None:
|
||
self._successful_generation = max(self._successful_generation, generation)
|
||
self._last_error = None
|
||
else:
|
||
self._last_error = error
|
||
self._condition.notify_all()
|
||
if error is None:
|
||
if self._failure_logged:
|
||
logger.info("Gateway runtime-status persistence recovered")
|
||
self._failure_logged = False
|
||
elif not self._failure_logged:
|
||
logger.warning(
|
||
"Failed to persist gateway runtime status; later updates will retry: %s", error)
|
||
self._failure_logged = True
|
||
else:
|
||
logger.debug("Failed to persist gateway runtime status: %s", error)
|
||
|
||
|
||
_runtime_status_state_lock = threading.RLock()
|
||
_runtime_status_state_path: Optional[Path] = None
|
||
_runtime_status_state: Optional[dict[str, Any]] = None
|
||
|
||
|
||
def _merge_over_on_disk(path: Path, payload: dict[str, Any]) -> dict[str, Any]:
|
||
"""Lay the canonical snapshot over whatever is on disk right before writing. Out-of-process
|
||
writers (the migration's compensator clearing multiplex-owned status, container_boot
|
||
seeding ``desired_state``) stamp this file directly; the gateway's fields win, theirs survive."""
|
||
existing = _read_json_file(path)
|
||
return {**existing, **payload} if isinstance(existing, dict) else payload
|
||
|
||
|
||
_runtime_status_writer: Optional[_RuntimeStatusWriter] = None
|
||
|
||
|
||
def _get_runtime_status_writer() -> _RuntimeStatusWriter:
|
||
"""Lazily create the single writer; callers serialise on ``_runtime_status_state_lock``."""
|
||
global _runtime_status_writer
|
||
with _runtime_status_state_lock:
|
||
if _runtime_status_writer is None:
|
||
_runtime_status_writer = _RuntimeStatusWriter()
|
||
return _runtime_status_writer
|
||
|
||
|
||
def flush_runtime_status(timeout: float = 2.0) -> bool:
|
||
"""Wait boundedly for all runtime-status updates submitted so far."""
|
||
writer = _runtime_status_writer
|
||
return True if writer is None else writer.flush(timeout=timeout)
|
||
|
||
|
||
async def flush_runtime_status_async(timeout: float = 2.0) -> bool:
|
||
"""Await the current writer generation without blocking the event loop."""
|
||
writer = _runtime_status_writer
|
||
if writer is None:
|
||
return True
|
||
# ``flush()`` returns at its deadline, so the worker thread is bounded.
|
||
return await asyncio.to_thread(writer.flush, timeout=max(float(timeout), 0.0))
|
||
|
||
|
||
class StormInfo(NamedTuple):
|
||
"""Respawn-storm check result: start count, window, and backoff to sleep."""
|
||
|
||
count: int
|
||
window_s: float
|
||
backoff_s: float
|
||
|
||
|
||
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 is logged and swallowed, never fatal."""
|
||
try:
|
||
path = get_hermes_home() / "gateway-starts.log"
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
now = datetime.now(timezone.utc).timestamp()
|
||
existing: list[float] = []
|
||
if path.exists():
|
||
for line in path.read_text(encoding="utf-8-sig").splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
existing.append(float(line))
|
||
except ValueError:
|
||
pass # corrupt/truncated ledger line — skip, never fatal
|
||
existing.append(now)
|
||
recent = [ts for ts in existing if now - ts <= window_s]
|
||
# Ring-buffer the persisted file so it stays bounded.
|
||
to_write = existing[-max(max_starts * 4, 40):]
|
||
tmp = path.with_suffix(".tmp")
|
||
tmp.write_text("\n".join(repr(ts) for ts in to_write) + "\n", encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
if len(recent) <= max_starts:
|
||
return None
|
||
backoff = min(backoff_cap_s, 5.0 * (2 ** min(len(recent) - max_starts, 6)))
|
||
return StormInfo(count=len(recent), window_s=window_s, backoff_s=backoff)
|
||
except Exception as _e:
|
||
logger.debug("respawn-storm breaker bookkeeping failed (non-fatal): %s", _e)
|
||
return None
|
||
|
||
|
||
def _get_process_hermes_home() -> Path:
|
||
"""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."""
|
||
return get_process_hermes_home()
|
||
|
||
|
||
def _canonical_hermes_home(path: Path | str) -> Path:
|
||
"""Stable absolute HERMES_HOME path for persisted identity data."""
|
||
return Path(path).expanduser().resolve(strict=False)
|
||
|
||
|
||
def _same_hermes_home(left: Path | str, right: Path | str) -> bool:
|
||
"""Compare HERMES_HOME paths with the host platform's case semantics."""
|
||
left_c = os.path.normcase(str(_canonical_hermes_home(left)))
|
||
return left_c == os.path.normcase(str(_canonical_hermes_home(right)))
|
||
|
||
|
||
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:
|
||
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
|
||
try:
|
||
base = expected_home if expected_home is not None else _get_process_hermes_home()
|
||
return not _same_hermes_home(recorded_home, base)
|
||
except Exception:
|
||
return True
|
||
|
||
|
||
# Mirrors hermes_cli.profiles._PROFILE_ID_RE -- duplicated so gateway identity code
|
||
# stays import-light (hermes_constants + stdlib only).
|
||
_PROFILE_LABEL_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
|
||
|
||
|
||
def _profile_label_for_home(home: Path | str) -> Optional[str]:
|
||
"""Best-effort label: ``<root>/profiles/<name>`` -> name, root home -> "default", else None."""
|
||
try:
|
||
canonical = _canonical_hermes_home(home)
|
||
except Exception:
|
||
return None
|
||
if canonical.parent.name == "profiles" and _PROFILE_LABEL_RE.match(canonical.name):
|
||
return canonical.name
|
||
import hermes_constants
|
||
default_homes = (hermes_constants.get_default_hermes_root, _get_platform_default_hermes_home)
|
||
for default_home in default_homes:
|
||
with contextlib.suppress(Exception):
|
||
if _same_hermes_home(canonical, default_home()):
|
||
return "default"
|
||
return None
|
||
|
||
|
||
def scoped_lock_owner_label(record: Optional[dict[str, Any]]) -> Optional[str]:
|
||
"""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")
|
||
if isinstance(profile, str) and _PROFILE_LABEL_RE.match(profile.strip()):
|
||
return profile.strip()
|
||
home = record.get("hermes_home")
|
||
return _profile_label_for_home(home) if isinstance(home, str) and home.strip() else None
|
||
|
||
|
||
def _get_pid_path() -> Path:
|
||
return _get_process_hermes_home() / "gateway.pid"
|
||
|
||
|
||
def _get_gateway_lock_path(pid_path: Optional[Path] = None) -> Path:
|
||
return (pid_path or _get_pid_path()).with_name(_GATEWAY_LOCK_FILENAME)
|
||
|
||
|
||
def _get_runtime_status_path() -> Path:
|
||
return _get_process_hermes_home() / _RUNTIME_STATUS_FILE
|
||
|
||
|
||
def _get_lock_dir() -> Path:
|
||
"""Cross-profile rendezvous dir for machine-local locks; ``HERMES_GATEWAY_LOCK_DIR`` overrides.
|
||
|
||
Scope is the **OS user**, not the kernel host: separate users have separate ``$HOME``s,
|
||
separate ``~/.hermes`` profile roots and separate credentials, so "one gateway per host"
|
||
means "one per host per OS user". Holds the token-scoped locks (:func:`acquire_scoped_lock`)
|
||
and the host-role lock + rendezvous record (``gateway/host_rendezvous.py``); the per-home
|
||
``gateway.pid``/``gateway.lock`` above deliberately stay under each profile's HERMES_HOME.
|
||
"""
|
||
override = os.getenv("HERMES_GATEWAY_LOCK_DIR")
|
||
if override:
|
||
return Path(override)
|
||
# XDG spec: a relative $XDG_STATE_HOME is INVALID and must be ignored. Honouring one made the
|
||
# lock dir CWD-relative, so two serves started from different directories shared no singleton.
|
||
state_home_env = os.getenv("XDG_STATE_HOME") or ""
|
||
state_home = Path(state_home_env) if os.path.isabs(state_home_env) else Path.home() / ".local" / "state"
|
||
return state_home / "hermes" / _LOCKS_DIRNAME
|
||
|
||
|
||
def _utc_now_iso() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
# Epochs before 2000-01-01 are corrupt/hand-edited state (e.g. an accidental 0).
|
||
_EPOCH_MIN_PLAUSIBLE = 946684800.0 # 2000-01-01T00:00:00Z
|
||
|
||
|
||
def normalize_updated_at(value: Any) -> Optional[str]:
|
||
"""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.
|
||
if raw.endswith(("Z", "z")):
|
||
raw = raw[:-1] + "+00:00"
|
||
try:
|
||
parsed = datetime.fromisoformat(raw)
|
||
except ValueError:
|
||
return None
|
||
return (parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)).isoformat()
|
||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
seconds = float(value)
|
||
now = datetime.now(timezone.utc).timestamp()
|
||
if not math.isfinite(seconds) or seconds < _EPOCH_MIN_PLAUSIBLE or seconds > now + 86400:
|
||
return None
|
||
try:
|
||
return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat()
|
||
except (OverflowError, OSError, ValueError):
|
||
return None
|
||
return None
|
||
|
||
|
||
# ``exit_reason`` values the out-of-loop watchdogs (gateway/shutdown_watchdog.py) stamp together with
|
||
# ``gateway_state: degraded`` right before they hard-exit a wedged process (#113372).
|
||
WATCHDOG_EXIT_REASONS = frozenset({"loop_liveness_watchdog", "shutdown_watchdog"})
|
||
|
||
|
||
def retained_gateway_state(runtime: Any) -> str:
|
||
"""What a NOT-running gateway's retained ``gateway_state.json`` says about it now:
|
||
``"startup_failed"`` (or a watchdog-stamped ``"degraded"``) only while the operator still
|
||
wants it running, else ``"stopped"``.
|
||
|
||
``hermes gateway stop`` keeps the last ``startup_failed`` + ``exit_reason`` on disk for
|
||
diagnostics and records the durable stop intent as ``desired_state``; a profile the operator
|
||
stopped is "stopped", not a current failure. A watchdog exit (``degraded`` + an exit_reason in
|
||
``WATCHDOG_EXIT_REASONS``) is the same kind of current failure as ``startup_failed`` and is kept
|
||
under the same rule, so the dashboard agrees with ``hermes gateway status``. Any other retained
|
||
state of a dead process (``running``, ``starting``, missing) is just "stopped". Shared by
|
||
``/api/status`` and ``/api/messaging/platforms`` so the sidebar strip and the Channels page
|
||
cannot disagree."""
|
||
rt = runtime if isinstance(runtime, dict) else {}
|
||
if rt.get("desired_state") != "stopped":
|
||
if rt.get("gateway_state") == "startup_failed":
|
||
return "startup_failed"
|
||
if rt.get("gateway_state") == "degraded" and rt.get("exit_reason") in WATCHDOG_EXIT_REASONS:
|
||
return "degraded"
|
||
return "stopped"
|
||
|
||
|
||
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:
|
||
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.
|
||
|
||
On POSIX an expectation is optional, but when the caller provides one and it no longer matches the live
|
||
process, the kill is refused on every platform — a mismatched fingerprint always means the PID was
|
||
recycled. See #89614.
|
||
"""
|
||
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")
|
||
current_start_time = _get_process_start_time(pid)
|
||
if current_start_time is None:
|
||
raise OSError(f"refusing to force-kill PID {pid}; process start time is unavailable")
|
||
try:
|
||
if not _start_times_agree(current_start_time, expected_start_time):
|
||
raise OSError(f"refusing to force-kill PID {pid}; process identity changed")
|
||
except (TypeError, ValueError) as exc:
|
||
raise OSError(f"refusing to force-kill PID {pid}; malformed start time") from exc
|
||
if not (force and _IS_WINDOWS):
|
||
os.kill(pid, signal.SIGTERM if not force else getattr(signal, "SIGKILL", signal.SIGTERM))
|
||
return
|
||
# Hide flags: a bare taskkill spawn from windowless pythonw.exe would flash a conhost window.
|
||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||
|
||
try:
|
||
result = subprocess.run(
|
||
["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True,
|
||
encoding="utf-8", errors="replace", timeout=10, creationflags=windows_hide_flags(),
|
||
)
|
||
except FileNotFoundError:
|
||
os.kill(pid, signal.SIGTERM)
|
||
return
|
||
if result.returncode != 0:
|
||
details = (result.stderr or result.stdout or "").strip()
|
||
raise OSError(details or f"taskkill failed for PID {pid}")
|
||
|
||
|
||
def _start_times_agree(current: Any, *recorded: Any) -> bool:
|
||
"""Same process object: all fingerprints > 0 and within 1ms of ``current``; raises on junk."""
|
||
cur = float(current)
|
||
return cur > 0 and all(r > 0 and abs(r - cur) <= 0.001 for r in map(float, recorded))
|
||
|
||
|
||
# Same-host start-time readings can drift by ~1 s between the claim-time and a later liveness read
|
||
# (macOS ``kern.boottime`` adjustment, #117505). Both fingerprint scales are ×100 (Linux /proc ticks,
|
||
# psutil centiseconds), so 200 means 2 s on either platform — a recycled PID is essentially never
|
||
# that close to the original's start time.
|
||
START_TIME_DRIFT_TOLERANCE = 200
|
||
|
||
|
||
def start_time_fingerprints_match(recorded: Any, current: Any, tolerance: int = START_TIME_DRIFT_TOLERANCE) -> bool:
|
||
"""Liveness-reconciliation comparator for :func:`get_process_start_time` fingerprints: the
|
||
recorded owner and the current reading are the same incarnation when they agree within
|
||
``tolerance``. Raises on junk; callers decide what an unreadable (``None``) side means."""
|
||
return abs(int(current) - int(recorded)) <= tolerance
|
||
|
||
|
||
def _scope_hash(identity: str) -> str:
|
||
return hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
def _get_scope_lock_path(scope: str, identity: str) -> Path:
|
||
return _get_lock_dir() / f"{scope}-{_scope_hash(identity)}.lock"
|
||
|
||
|
||
def _get_process_start_time(pid: int) -> Optional[int]:
|
||
"""Return a stable per-process start-time fingerprint, or None.
|
||
|
||
Used as a PID-reuse guard: a ``(pid, start_time)`` pair uniquely identifies
|
||
a process, so a recycled PID (same number, different process) yields a
|
||
different value and is never mistaken for the original.
|
||
|
||
On Linux this is field 22 of ``/proc/<pid>/stat`` (start time in clock
|
||
ticks since boot, an int). On platforms without ``/proc`` (macOS, Windows)
|
||
we fall back to ``psutil.Process(pid).create_time()`` — a float epoch
|
||
timestamp — quantized to an int (centiseconds) for stable equality.
|
||
|
||
The two sources are never mixed on a single platform: ``/proc`` always
|
||
succeeds first on Linux, and always fails on macOS/Windows so psutil is
|
||
always used there. Because the guard only compares the value recorded at
|
||
spawn against the live value *on the same host*, the differing units across
|
||
platforms are irrelevant — only same-source equality matters.
|
||
"""
|
||
stat_path = Path(f"/proc/{pid}/stat")
|
||
try:
|
||
# Field 22 in /proc/<pid>/stat is process start time (clock ticks).
|
||
return int(stat_path.read_text(encoding="utf-8").split()[21]) # windows-footgun: ok (/proc is BOM-free)
|
||
except (FileNotFoundError, IndexError, PermissionError, ValueError, OSError):
|
||
pass
|
||
|
||
# No /proc (macOS / Windows): psutil is a hard dependency and exposes a
|
||
# cross-platform creation time. Quantize to centiseconds so repeated reads
|
||
# of the same process compare equal without float-precision fragility.
|
||
try:
|
||
import psutil # type: ignore
|
||
return int(round(psutil.Process(pid).create_time() * 100))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_process_start_time(pid: int) -> Optional[int]:
|
||
"""Public wrapper for retrieving a process start time when available."""
|
||
return _get_process_start_time(pid)
|
||
|
||
|
||
def _read_process_cmdline(pid: int) -> Optional[str]:
|
||
"""Process command line as one string: /proc, then psutil, then ``ps``.
|
||
|
||
Order is by cost, and this runs per live gateway on every roster/status poll. ``psutil`` reads
|
||
the process table in-process (a ``sysctl`` on macOS) where ``ps`` costs a fork+exec — measured
|
||
0.02ms against 4.2ms on macOS for the same string. It cannot always answer: on macOS it raises
|
||
``AccessDenied`` for a process owned by another user, which ``ps`` still reports, so ``ps``
|
||
stays as the fallback rather than being replaced."""
|
||
with contextlib.suppress(OSError):
|
||
raw = Path(f"/proc/{pid}/cmdline").read_bytes()
|
||
if raw:
|
||
return raw.replace(b"\x00", b" ").decode("utf-8", errors="ignore").strip()
|
||
with contextlib.suppress(Exception):
|
||
import psutil # type: ignore
|
||
cmdline_parts = psutil.Process(pid).cmdline()
|
||
if cmdline_parts:
|
||
return " ".join(cmdline_parts)
|
||
if not _IS_WINDOWS:
|
||
with contextlib.suppress(OSError, subprocess.TimeoutExpired):
|
||
result = subprocess.run(
|
||
["ps", "-p", str(pid), "-o", "command="],
|
||
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
return result.stdout.strip()
|
||
return None
|
||
|
||
|
||
|
||
def inline_source_flag_index(tokens: list[str]) -> int | None:
|
||
"""Index of the ``-c`` token when *tokens* is an interpreter running INLINE SOURCE, else None.
|
||
|
||
Everything after ``-c`` is data the inline program receives, not this process's own identity.
|
||
The detached gateway restart watcher (``gateway._spawn_gateway_restart_watcher``) is spawned as
|
||
``python -c <watcher source> <old_pid> <python> -m hermes_cli.main gateway run``: its trailing
|
||
argv is the command the watcher will LATER spawn, so every argv matcher used to read it as a
|
||
live gateway. See #107002 and the "never infer process identity from argv substrings" rule.
|
||
|
||
Only interpreter options may precede ``-c``; the first non-option token ends the option block
|
||
(``python -m hermes_cli.main …`` therefore never matches).
|
||
|
||
The walk is VALUE-AWARE: ``-X``/``-W``/``-Q`` and ``--check-hash-based-pycs``/``--jit`` take a
|
||
SEPARATE operand, so a naive "first non-option token ends the block" walk mistakes that operand
|
||
for the end of the block and never reaches the ``-c`` behind it (``python -X utf8 -c <src> …``
|
||
was still read as a live gateway). The operand sets are the canonical ones in
|
||
``hermes_state_holders``, not a second hand-rolled copy.
|
||
|
||
*tokens* must be CASE-PRESERVING: the operand-taking ``-Q``/``-W``/``-X`` differ from the
|
||
operand-less ``-q``/``-b``, so a lowercased argv would skip the token after a plain ``-q``.
|
||
"""
|
||
from hermes_state_holders import (
|
||
_PYTHON_LONG_OPTIONS_WITH_OPERANDS,
|
||
_PYTHON_SHORT_OPTIONS_WITH_OPERANDS,
|
||
)
|
||
|
||
index = 1
|
||
while index < len(tokens):
|
||
token = tokens[index]
|
||
if token == "--":
|
||
return None
|
||
if token in _PYTHON_LONG_OPTIONS_WITH_OPERANDS:
|
||
index += 2 # the next token is this option's operand, not the end of the option block
|
||
continue
|
||
if token.startswith("--"):
|
||
index += 1 # ``--opt=value`` and operand-less long options
|
||
continue
|
||
if not token.startswith("-") or token == "-":
|
||
return None
|
||
# Clustered short options (``-uc``, ``-IsB``). An operand-taking letter consumes the rest of
|
||
# the cluster as its attached value, or the following token when the cluster ends there --
|
||
# so ``-Xc`` is ``-X c``, NOT an inline-source ``-c``.
|
||
cluster = token[1:]
|
||
for position, letter in enumerate(cluster):
|
||
if letter == "c":
|
||
return index
|
||
if letter in _PYTHON_SHORT_OPTIONS_WITH_OPERANDS:
|
||
index += 1 if cluster[position + 1 :] else 2
|
||
break
|
||
else:
|
||
index += 1
|
||
return None
|
||
|
||
|
||
def command_line_runs_inline_source(tokens: list[str]) -> bool:
|
||
"""True when *tokens* is an interpreter running INLINE SOURCE (``python -c <src> [args]``)."""
|
||
return inline_source_flag_index(tokens) is not None
|
||
|
||
|
||
# Hermes' own inline bootstraps hand control to a Hermes entry point IN this process, so the argv
|
||
# they run with is this process's own identity; every other ``-c`` program keeps its trailing argv
|
||
# as data (#107002). Each pattern is one emitted source shape, anchored at both ends so a program
|
||
# merely CARRYING a bootstrap command line (the restart watcher's respawn argv) never matches.
|
||
_Q = r"""['"]?"""
|
||
_MAIN = rf"{_Q}__main__{_Q}"
|
||
_RUN_MODULE = rf"runpy\.run_module\(\s*{_Q}(?P<target>[\w.]+){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*,\s*alter_sys\s*=\s*True\s*\)"
|
||
_BOOTSTRAPS = (
|
||
# hermes_cli._launchers.runtime_command (store launcher, the Windows updater's relaunch)
|
||
("module", re.compile(rf"import os, sys, runpy;.*\b{_RUN_MODULE}", re.S)),
|
||
# hermes_cli.venv_sync.relaunch_command: argv is assigned inside the source
|
||
("module", re.compile(rf"import sys, runpy; sys\.path\.insert\(.*\b{_RUN_MODULE}", re.S)),
|
||
("path", re.compile(
|
||
rf"import sys, runpy; sys\.path\.insert\(.*\brunpy\.run_path\(\s*{_Q}(?P<target>[^'\"]+?){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*\)",
|
||
re.S)),
|
||
# hermes_cli._launchers._launcher_script (the published POSIX shell / Windows .cmd launcher)
|
||
("entry", re.compile(r"import os, re, sys\s.*\bfrom\s+(?P<target>[\w.]+)\s+import\s+(?P<func>\w+)\b.*\bsys\.exit\(\s*(?P=func)\(\)\s*\)", re.S)),
|
||
# hermes_cli._launchers._write_cmd_launcher: the launcher script, base64-encoded
|
||
("base64", re.compile(rf"import base64; exec\(base64\.b64decode\({_Q}(?P<target>[A-Za-z0-9+/=]+){_Q}\)\)")),
|
||
)
|
||
_ASSIGNED_ARGV = re.compile(r"\bsys\.argv\s*=\s*\[(.*?)\]\s*;")
|
||
|
||
|
||
def _bootstrap_entry(source: str, argv: list[str]) -> list[str] | None:
|
||
"""``[-m, <module>, *argv]`` (or ``[<path>, *argv]``) the inline *source* runs in-process, else None."""
|
||
source = source.strip()
|
||
kind, match = next(((k, m) for k, p in _BOOTSTRAPS if (m := p.fullmatch(source))), (None, None))
|
||
if match is None:
|
||
return None
|
||
target = match["target"]
|
||
if kind == "base64":
|
||
import base64
|
||
import binascii
|
||
try:
|
||
return _bootstrap_entry(base64.b64decode(target, validate=True).decode("utf-8"), argv)
|
||
except (binascii.Error, UnicodeDecodeError):
|
||
return None
|
||
if kind == "entry": # the launcher script's own ``--run-module <module>`` switch
|
||
return ["-m", argv[1], *argv[2:]] if argv[:1] == ["--run-module"] and len(argv) > 1 else ["-m", target, *argv]
|
||
if assigned := _ASSIGNED_ARGV.search(source):
|
||
argv = [item.strip().strip("'\"") for item in assigned.group(1).split(",")][1:]
|
||
return [target, *argv] if kind == "path" else ["-m", target, *argv]
|
||
|
||
|
||
def inline_bootstrap_argv(tokens: list[str]) -> list[str] | None:
|
||
"""*tokens* as the equivalent ``python -m <module> <argv…>`` when this interpreter's ``-c`` source
|
||
is a Hermes bootstrap running an entry point in-process; None for any other inline source.
|
||
|
||
Command lines usually arrive space-joined (``/proc``, psutil, ``ps``), which splits the source
|
||
across tokens; the shortest token run that ends in a recognised tail is the source, whatever
|
||
joined it, and the tokens after it are the entry point's argv.
|
||
"""
|
||
index = inline_source_flag_index(tokens)
|
||
if index is None:
|
||
return None
|
||
for end in range(index + 1, len(tokens)):
|
||
if tokens[end].rstrip().endswith(")"):
|
||
entry = _bootstrap_entry(" ".join(tokens[index + 1 : end + 1]), tokens[end + 1 :])
|
||
if entry is not None:
|
||
return [tokens[0], *entry]
|
||
return None
|
||
|
||
|
||
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`` / ``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:
|
||
raw_tokens = shlex.split(command, posix=False)
|
||
except ValueError:
|
||
raw_tokens = command.split()
|
||
# Strip surrounding quotes, normalize slashes + case per token.
|
||
cased_tokens = [t.strip("\"'").replace("\\", "/") for t in raw_tokens]
|
||
tokens = [t.lower() for t in cased_tokens]
|
||
if not tokens:
|
||
return None
|
||
basenames = [t.rsplit("/", 1)[-1] for t in tokens]
|
||
# ``python -c <src> … -m hermes_cli.main gateway run``: the trailing argv belongs to the program
|
||
# the inline source will spawn later, not to this process (#107002). Case-preserving tokens:
|
||
# the operand-taking ``-X``/``-W``/``-Q`` must not be conflated with ``-q``/``-b``.
|
||
if command_line_runs_inline_source(cased_tokens):
|
||
# …unless the source is a Hermes bootstrap running the entry point in THIS process (store
|
||
# launcher, launcher script, venv_sync re-entry): then its argv is this process's (#124318).
|
||
cased_tokens = inline_bootstrap_argv(cased_tokens)
|
||
if cased_tokens is None:
|
||
return None
|
||
tokens = [t.lower() for t in cased_tokens]
|
||
basenames = [t.rsplit("/", 1)[-1] for t in tokens]
|
||
# The launchd job's osascript wrapper (gateway_launchd.launchd_program_arguments) carries the gateway argv
|
||
# inside one JXA script string; the gateway itself is its child and is matched on its own command line.
|
||
if basenames[0] == "osascript":
|
||
return None
|
||
# Gateway-dedicated entrypoints carry no subcommand to inspect.
|
||
if any(t == "gateway/run.py" or t.endswith("/gateway/run.py") for t in tokens):
|
||
return "run"
|
||
# Atomic Hermes' bundled desktop runner shares HERMES_HOME with the CLI; without this,
|
||
# `gateway run --replace` does not recognise it as a running gateway, skips the
|
||
# terminate-and-scoped-lock-handoff path, and collides with its still-held scoped locks
|
||
# (e.g. the Discord bot-token lock). See #22418.
|
||
if any(b == "desktop-gateway.py" for b in basenames):
|
||
return "run"
|
||
if any(b in ("hermes-gateway", "hermes-gateway.exe") for b in basenames):
|
||
return "run"
|
||
joined = " ".join(tokens)
|
||
if "hermes_cli.main" not in joined and "hermes_cli/main.py" not in joined and not any(
|
||
b in ("hermes", "hermes.exe") for b in basenames
|
||
):
|
||
return None
|
||
# Drop --profile X / -p X / --profile=X / -p=X (consumes a VALUE of "gateway" too).
|
||
filtered: list[str] = []
|
||
skip_next = False
|
||
for token in tokens:
|
||
if skip_next:
|
||
skip_next = False
|
||
elif token in ("--profile", "-p"):
|
||
skip_next = True
|
||
elif not token.startswith(("--profile=", "-p=")):
|
||
filtered.append(token)
|
||
for i, token in enumerate(filtered):
|
||
if token == "gateway":
|
||
# Bare `hermes gateway` defaults to `run`.
|
||
return filtered[i + 1] if i + 1 < len(filtered) else "run"
|
||
return None
|
||
|
||
|
||
def gateway_spawn_intent_subcommand(command: str | None) -> str | None:
|
||
"""Gateway lifecycle subcommand a command line would EVENTUALLY launch, or None.
|
||
|
||
The identity matcher (``_gateway_command_subcommand``) deliberately refuses ``python -c <src>
|
||
…``: the trailing argv is the inline program's data, not that process's own identity (#107002).
|
||
Callers that inspect a command line as SPAWN INTENT — "if I launch this, does a gateway runtime
|
||
eventually appear?" — need the opposite answer, because
|
||
``gateway._spawn_gateway_restart_watcher`` hides a real ``… -m hermes_cli.main gateway run``
|
||
behind exactly that wrapper. ``tests/_fixtures/live_system_guard.py`` is the canonical caller.
|
||
|
||
Still no substring matching: the wrapper is peeled token-wise and each remaining suffix is
|
||
handed to the same canonical matcher.
|
||
"""
|
||
direct = _gateway_command_subcommand(command)
|
||
if direct is not None or not command:
|
||
return direct
|
||
try:
|
||
raw_tokens = shlex.split(command, posix=False)
|
||
except ValueError:
|
||
raw_tokens = command.split()
|
||
cased_tokens = [t.strip("\"'").replace("\\", "/") for t in raw_tokens]
|
||
flag_index = inline_source_flag_index(cased_tokens)
|
||
if flag_index is None:
|
||
return None
|
||
# Skip the interpreter, its options, ``-c`` and the source literal; then try every suffix —
|
||
# the embedded argv starts at an unknown offset (the watcher prefixes it with the old PID).
|
||
start = flag_index + 2
|
||
for i in range(start, len(cased_tokens)):
|
||
nested = _gateway_command_subcommand(" ".join(raw_tokens[i:]))
|
||
if nested is not None:
|
||
return nested
|
||
return None
|
||
|
||
|
||
def looks_like_gateway_command_line(command: str | None) -> bool:
|
||
"""True only for a real ``gateway run`` process command line."""
|
||
return _gateway_command_subcommand(command) == "run"
|
||
|
||
|
||
def looks_like_gateway_runtime_command_line(command: str | None) -> bool:
|
||
"""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"}
|
||
|
||
|
||
def _looks_like_gateway_process(pid: int) -> bool:
|
||
"""True when the live PID still looks like the Hermes gateway."""
|
||
cmdline = _read_process_cmdline(pid)
|
||
return bool(cmdline) and looks_like_gateway_command_line(cmdline)
|
||
|
||
|
||
def _record_looks_like_gateway(record: dict[str, Any]) -> bool:
|
||
"""Validate gateway identity from PID-file metadata when cmdline is unavailable."""
|
||
argv = record.get("argv")
|
||
if record.get("kind") != _GATEWAY_KIND or not isinstance(argv, list) or not argv:
|
||
return False
|
||
return looks_like_gateway_runtime_command_line(" ".join(str(part) for part in argv))
|
||
|
||
|
||
def _profile_name_for_home(profile_home: Path) -> Optional[str]:
|
||
"""Profile id for ``<root>/profiles/<name>``; None for the root/default home (bare gateway)."""
|
||
return profile_home.name if profile_home.parent.name == "profiles" else None
|
||
|
||
|
||
def profile_flag_value(command: str) -> Optional[str]:
|
||
"""The ``-p``/``--profile`` argument of a command line, or None. Token equality is the only safe
|
||
profile match: a substring test lets ``-p ops`` claim (and ``gateway stop`` SIGTERM) ``-p ops-2``."""
|
||
tokens = command.split()
|
||
for i, tok in enumerate(tokens):
|
||
if tok.startswith("--profile="):
|
||
return tok.partition("=")[2]
|
||
if tok in ("-p", "--profile") and i + 1 < len(tokens):
|
||
return tokens[i + 1]
|
||
return None
|
||
|
||
|
||
_HERMES_HOME_ASSIGNMENT_RE = re.compile(r"(?:^|\s)hermes_home=(?:\"([^\"]*)\"|'([^']*)'|(\S+))")
|
||
|
||
|
||
def hermes_home_assignments(command: str) -> list[str]:
|
||
"""Values of every ``HERMES_HOME=<value>`` assignment in ``command`` (the caller lowercases
|
||
and normalizes separators). Values are token-bounded, quotes stripped: the substring test
|
||
this replaces let ``HERMES_HOME=/root/profiles/ops`` claim a ``/root/profiles/ops2`` gateway.
|
||
The name is token-bounded too (``FOO=hermes_home=/x`` is not an assignment), and a trailing
|
||
separator on the value is stripped -- ``HERMES_HOME=/root/.hermes/`` (systemd ``Environment=``
|
||
or a shell wrapper spelling) is the same home as ``/root/.hermes``; callers strip the profile
|
||
home the same way."""
|
||
return [
|
||
next(g for g in m.groups() if g is not None).rstrip("/")
|
||
for m in _HERMES_HOME_ASSIGNMENT_RE.finditer(command)
|
||
]
|
||
|
||
|
||
def command_line_names_hermes_home(command_lc: str, home_lc: str) -> bool:
|
||
"""True when ``command_lc`` carries ``HERMES_HOME=<home_lc>`` (both lowercased, ``/``-separated,
|
||
no trailing separator). Argv reaches us space-joined, so an unquoted value with a space in it
|
||
(``HERMES_HOME=C:/Users/John Doe/.hermes``) is cut at the space by the token parser; a
|
||
token-bounded literal match of the whole home recovers that spelling."""
|
||
if home_lc in hermes_home_assignments(command_lc):
|
||
return True
|
||
return re.search(rf"(?:^|\s)hermes_home={re.escape(home_lc)}/?(?=\s|$)", command_lc) is not None
|
||
|
||
|
||
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 ANOTHER profile's live gateway. Named profiles carry ``-p``/``--profile <name>`` 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("\\", "/").rstrip("/")
|
||
if profile_name is not None and profile_name != "default":
|
||
if profile_flag_value(command_lc) == profile_name.lower():
|
||
return True
|
||
return command_line_names_hermes_home(command_lc, home_lc)
|
||
# Default profile: accept unless argv names another profile (any spelling the CLI pre-parser
|
||
# accepts, ``--profile=ops`` included -- a substring test let that gateway pass as the default's)
|
||
# or a conflicting explicit HERMES_HOME= (its absence is not disqualifying -- HERMES_HOME usually
|
||
# arrives via the env).
|
||
if profile_flag_value(command_lc) is not None:
|
||
return False
|
||
return not hermes_home_assignments(command_lc) or command_line_names_hermes_home(command_lc, home_lc)
|
||
|
||
|
||
def _host_gateway_serves_home(pid: int, profile_home: Path) -> bool:
|
||
"""Does the ONE host gateway — PID ``pid`` — serve ``profile_home``'s profile?
|
||
|
||
Argv cannot answer this: the host singleton runs ONE home's (usually bare/default) command line
|
||
while multiplexing every profile, so :func:`_command_line_belongs_to_profile` rejects every
|
||
secondary and the profile reads as "not running" while its messages are being served. The live
|
||
served set is the only proof; the argv rule stays as the fallback when no record exists.
|
||
"""
|
||
try:
|
||
from gateway.host_attach import host_gateway, profile_name_for_home
|
||
|
||
owner = host_gateway()
|
||
except Exception:
|
||
return False
|
||
return owner is not None and owner.pid == pid and owner.serves(profile_name_for_home(profile_home))
|
||
|
||
|
||
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. 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 — or serve it as the host multiplexer); unreadable cmdline
|
||
(Windows/EACCES) -> persisted record."""
|
||
live_cmdline = _read_process_cmdline(pid)
|
||
if not live_cmdline:
|
||
return _record_looks_like_gateway(record)
|
||
if not looks_like_gateway_runtime_command_line(live_cmdline):
|
||
return False
|
||
if expected_home is not None and _host_gateway_serves_home(pid, expected_home):
|
||
return True
|
||
return expected_home is None or _command_line_belongs_to_profile(live_cmdline, expected_home)
|
||
|
||
|
||
def _build_pid_record() -> dict:
|
||
return {
|
||
"pid": os.getpid(), "kind": _GATEWAY_KIND, "argv": list(sys.argv),
|
||
"start_time": _get_process_start_time(os.getpid()),
|
||
# Scoped locks are machine-global; the owner's home lets a cross-profile
|
||
# --replace place its takeover marker where the target will read it.
|
||
"hermes_home": str(_canonical_hermes_home(_get_process_hermes_home())),
|
||
}
|
||
|
||
|
||
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.
|
||
|
||
A gateway keeps serving the module versions it imported at startup, so stamping the identity into
|
||
``gateway_state.json`` lets `hermes update` (and the dashboard) prove whether a running gateway actually
|
||
picked up new code after the restart phase — instead of assuming it did (#88654, #69754). Never raises;
|
||
degrades to absent fields.
|
||
"""
|
||
try:
|
||
from hermes_cli.version_info import get_code_identity
|
||
|
||
identity = get_code_identity()
|
||
return {"code_sha": identity.get("sha"), "code_version": identity.get("version")}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
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);
|
||
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")
|
||
return not record_home or _same_hermes_home(record_home, _get_process_hermes_home())
|
||
|
||
|
||
def _build_runtime_status_record() -> dict[str, Any]:
|
||
return {
|
||
**_build_pid_record(), "gateway_state": "starting", "exit_reason": None,
|
||
"restart_requested": False, "active_agents": 0, "platforms": {},
|
||
"session_store": {"status": "unknown"}, "updated_at": _utc_now_iso(),
|
||
**_get_code_identity_fields(),
|
||
}
|
||
|
||
|
||
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}``."""
|
||
try:
|
||
raw = path.read_text(encoding="utf-8-sig").strip()
|
||
except (OSError, UnicodeDecodeError):
|
||
# OSError: file vanished or permission flipped between exists() and
|
||
# read. UnicodeDecodeError: file holds non-UTF-8 / binary garbage
|
||
# (a truncated or clobbered status file). Either way it's unusable.
|
||
return None
|
||
if not raw:
|
||
return None
|
||
try:
|
||
payload = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
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
|
||
|
||
|
||
def _write_json_file(path: Path, payload: dict[str, Any]) -> None:
|
||
atomic_json_write(path, payload, indent=None, separators=(",", ":"))
|
||
|
||
|
||
def _unlink_quietly(path: Path) -> None:
|
||
with contextlib.suppress(OSError):
|
||
path.unlink(missing_ok=True)
|
||
|
||
|
||
def _read_pid_record(pid_path: Optional[Path] = None) -> Optional[dict]:
|
||
return _read_json_file(pid_path or _get_pid_path(), bare_pid_ok=True)
|
||
|
||
|
||
def _read_gateway_lock_record(lock_path: Optional[Path] = None) -> Optional[dict[str, Any]]:
|
||
return _read_json_file(lock_path or _get_gateway_lock_path(), bare_pid_ok=True)
|
||
|
||
|
||
def _pid_from_record(record: Optional[dict[str, Any]], key: str = "pid") -> Optional[int]:
|
||
try:
|
||
return int(record[key])
|
||
except (KeyError, TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _start_times_conflict(recorded_start: Any, current_start: Any) -> bool:
|
||
"""PID-reuse guard: True only when BOTH start times are known and differ."""
|
||
return None not in (recorded_start, current_start) and current_start != recorded_start
|
||
|
||
|
||
def _live_pid_from_record(record: Optional[dict[str, Any]]) -> Optional[int]:
|
||
"""Record's PID when it is alive and passes the start-time PID-reuse guard, else None."""
|
||
pid = _pid_from_record(record)
|
||
if pid is None or not _pid_exists(pid):
|
||
return None
|
||
if _start_times_conflict(record.get("start_time"), _get_process_start_time(pid)):
|
||
return None
|
||
return pid
|
||
|
||
|
||
def _clear_running_pid_cache() -> None:
|
||
with _gateway_running_pid_cache_lock:
|
||
_gateway_running_pid_cache.clear()
|
||
|
||
|
||
def _file_cache_signature(path: Path) -> tuple[bool, Optional[int], Optional[int]]:
|
||
try:
|
||
st = path.stat()
|
||
except OSError:
|
||
return (False, None, None)
|
||
return (True, st.st_mtime_ns, st.st_size)
|
||
|
||
|
||
def _cleanup_invalid_pid_path(pid_path: Path, *, cleanup_stale: bool) -> None:
|
||
"""Force-unlink a stale PID file + sibling lock (lock confirmed inactive, so no pid check)."""
|
||
if not cleanup_stale:
|
||
return
|
||
_clear_running_pid_cache()
|
||
for path in (pid_path, _get_gateway_lock_path(pid_path)):
|
||
with contextlib.suppress(Exception):
|
||
path.unlink(missing_ok=True)
|
||
|
||
|
||
def _try_acquire_file_lock(handle) -> bool:
|
||
try:
|
||
if _IS_WINDOWS:
|
||
handle.seek(0, os.SEEK_END)
|
||
if handle.tell() == 0:
|
||
handle.write("\n")
|
||
handle.flush()
|
||
handle.seek(_WINDOWS_LOCK_OFFSET)
|
||
msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
|
||
else:
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
return True
|
||
except (BlockingIOError, OSError):
|
||
return False
|
||
|
||
|
||
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)`` 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
|
||
# Best-effort zombie check: status-read failures fall through to pid_exists().
|
||
# Windows has no POSIX zombies, and this probe costs ~7 ms per call — once per
|
||
# registry entry inside the session file lock (#115578). Skip it on Windows and
|
||
# let pid_exists() below (or the ctypes fallback) decide.
|
||
probe_zombie = os.name != "nt"
|
||
try:
|
||
# A zombie (defunct) process is still in the process table, so ``psutil.pid_exists()`` returns
|
||
# True for it — but it is already dead: SIGKILL has no effect and it cannot be a running
|
||
# gateway. Treating a zombie as alive makes ``--replace`` wait for the old PID to die (it never
|
||
# does, until its parent reaps it), then abort with exit 1 — a silent crash loop under systemd
|
||
# ``Restart=always``, which respawns the gateway before reaping the previous process (issue
|
||
# #42126). Report zombies as dead so the takeover proceeds. Best-effort: any failure to read
|
||
# status (partial/stub psutil, access denied, transient race) falls through to the authoritative
|
||
# ``pid_exists()`` below rather than raising.
|
||
if probe_zombie and psutil.Process(pid).status() == psutil.STATUS_ZOMBIE:
|
||
return False
|
||
except getattr(psutil, "NoSuchProcess", ()):
|
||
return False
|
||
except Exception:
|
||
pass
|
||
return bool(psutil.pid_exists(pid))
|
||
except ImportError:
|
||
pass # Fall through to stdlib fallback.
|
||
return pid_exists_stdlib(pid)
|
||
|
||
|
||
def _release_file_lock(handle) -> None:
|
||
with contextlib.suppress(OSError):
|
||
if _IS_WINDOWS:
|
||
handle.seek(_WINDOWS_LOCK_OFFSET)
|
||
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
||
else:
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||
|
||
|
||
def acquire_gateway_runtime_lock() -> bool:
|
||
"""Claim the cross-process runtime lock; the OS releases it if the process dies."""
|
||
global _gateway_lock_handle
|
||
if _gateway_lock_handle is not None:
|
||
return True
|
||
path = _get_gateway_lock_path()
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
handle = open(path, "a+", encoding="utf-8")
|
||
except PermissionError:
|
||
# 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")
|
||
except OSError:
|
||
return False
|
||
if not _try_acquire_file_lock(handle):
|
||
handle.close()
|
||
return False
|
||
handle.seek(0)
|
||
handle.truncate()
|
||
json.dump(_build_pid_record(), handle)
|
||
handle.flush()
|
||
with contextlib.suppress(OSError):
|
||
os.fsync(handle.fileno())
|
||
_gateway_lock_handle = handle
|
||
_clear_running_pid_cache()
|
||
return True
|
||
|
||
|
||
def release_gateway_runtime_lock() -> None:
|
||
"""Release the gateway runtime lock when owned by this process."""
|
||
global _gateway_lock_handle
|
||
handle, _gateway_lock_handle = _gateway_lock_handle, None
|
||
if handle is None:
|
||
return
|
||
_release_file_lock(handle)
|
||
with contextlib.suppress(OSError):
|
||
handle.close()
|
||
_clear_running_pid_cache()
|
||
|
||
|
||
def owns_gateway_runtime_lock() -> bool:
|
||
"""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 only the handle discriminates."""
|
||
return _gateway_lock_handle is not None
|
||
|
||
|
||
def _probe_lock_file(handle) -> bool:
|
||
"""True when another process holds the lock (a won probe is released); closes ``handle``."""
|
||
try:
|
||
held = not _try_acquire_file_lock(handle)
|
||
if not held:
|
||
_release_file_lock(handle)
|
||
return held
|
||
finally:
|
||
with contextlib.suppress(OSError):
|
||
handle.close()
|
||
|
||
|
||
def is_gateway_runtime_lock_active(lock_path: Optional[Path] = None) -> bool:
|
||
"""True when some process currently owns the gateway runtime lock."""
|
||
resolved_lock_path = lock_path or _get_gateway_lock_path()
|
||
if _gateway_lock_handle is not None and resolved_lock_path == _get_gateway_lock_path():
|
||
return True
|
||
if not resolved_lock_path.exists():
|
||
return False
|
||
try:
|
||
handle = open(resolved_lock_path, "a+", encoding="utf-8")
|
||
except PermissionError:
|
||
# Stale root-owned lock (see acquire_gateway_runtime_lock): report inactive.
|
||
_unlink_quietly(resolved_lock_path)
|
||
return False
|
||
return _probe_lock_file(handle)
|
||
|
||
|
||
def _strict_path_exists(path: Path, label: str) -> bool:
|
||
"""Like ``path.exists()`` but raises RuntimeError instead of False on EACCES-style errors."""
|
||
try:
|
||
path.stat()
|
||
return True
|
||
except FileNotFoundError:
|
||
return False
|
||
except OSError as exc:
|
||
raise RuntimeError(f"{label} metadata is not inspectable: {exc}") from exc
|
||
|
||
|
||
def _is_gateway_runtime_lock_active_strict(lock_path: Path) -> bool:
|
||
"""Probe ownership without treating access failures as absence."""
|
||
try:
|
||
handle = open(lock_path, "r+", encoding="utf-8")
|
||
except FileNotFoundError:
|
||
return False
|
||
except OSError as exc:
|
||
raise RuntimeError(f"gateway runtime lock is not inspectable: {exc}") from exc
|
||
try:
|
||
return _probe_lock_file(handle)
|
||
except OSError as exc:
|
||
raise RuntimeError(f"gateway runtime lock probe failed: {exc}") from exc
|
||
|
||
|
||
def write_pid_file() -> None:
|
||
"""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)
|
||
_write_json_excl(path, _build_pid_record())
|
||
_clear_running_pid_cache()
|
||
|
||
|
||
def _write_json_excl(path: Path, record: dict[str, Any]) -> None:
|
||
"""Create ``path`` with O_CREAT|O_EXCL and dump ``record``; unlinks on a failed write."""
|
||
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||
json.dump(record, handle)
|
||
except Exception:
|
||
_unlink_quietly(path)
|
||
raise
|
||
|
||
|
||
def _apply_set_fields(target: dict[str, Any], fields) -> None:
|
||
"""Assign each ``(key, value, coerce)`` whose value was explicitly passed (not ``_UNSET``)."""
|
||
for key, value, coerce in fields:
|
||
if value is not _UNSET:
|
||
target[key] = coerce(value) if coerce is not None else value
|
||
|
||
|
||
def _coerce_session_store(session_store: Any) -> dict[str, str]:
|
||
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 _prepare_runtime_status_update(
|
||
*, gateway_state: Any = _UNSET, exit_reason: Any = _UNSET, restart_requested: Any = _UNSET,
|
||
active_agents: Any = _UNSET, active_work: Any = _UNSET, platform: Any = _UNSET, platform_state: Any = _UNSET,
|
||
error_code: Any = _UNSET, error_message: Any = _UNSET, needs_attention: Any = _UNSET,
|
||
retrying_since: Any = _UNSET, served_profiles: Any = _UNSET, session_store: Any = _UNSET,
|
||
multiplex_standalone_reason: Any = _UNSET,
|
||
platform_metrics: Any = _UNSET,
|
||
ingress_url: Any = _UNSET, listener_base: Any = _UNSET, clear_profile_platforms: bool = False,
|
||
drop_profile_platforms: Optional[str] = None,
|
||
load_existing: bool = True, reload_existing: bool = False,
|
||
) -> tuple[Path, dict[str, Any], dict[str, Any]]:
|
||
"""Merge one update into the process-wide canonical status snapshot."""
|
||
global _runtime_status_state_path, _runtime_status_state
|
||
path = _get_runtime_status_path()
|
||
with _runtime_status_state_lock:
|
||
if reload_existing or _runtime_status_state_path != path or _runtime_status_state is None:
|
||
_runtime_status_state_path = path
|
||
_runtime_status_state = (
|
||
(_read_json_file(path) if load_existing else None) or _build_runtime_status_record())
|
||
# The module snapshot is only ever reassigned (never mutated in place) and
|
||
# submit() copies again, so the previous snapshot can be handed out as-is.
|
||
previous_payload = _runtime_status_state
|
||
payload = copy.deepcopy(previous_payload)
|
||
current_record = _build_pid_record()
|
||
payload.setdefault("platforms", {})
|
||
if not isinstance(payload["platforms"], dict):
|
||
payload["platforms"] = {}
|
||
if clear_profile_platforms or drop_profile_platforms:
|
||
drop_prefix = f"{drop_profile_platforms}:" if drop_profile_platforms else None
|
||
payload["platforms"] = {
|
||
k: v for k, v in payload["platforms"].items()
|
||
if not isinstance(k, str) or ":" not in k
|
||
or (drop_prefix is not None and not k.startswith(drop_prefix))
|
||
}
|
||
payload.update({key: current_record[key] for key in ("kind", "pid", "argv", "start_time")})
|
||
payload["updated_at"] = _utc_now_iso()
|
||
payload.update(_get_code_identity_fields())
|
||
_apply_set_fields(payload, (
|
||
("gateway_state", gateway_state, None), ("exit_reason", exit_reason, None),
|
||
("restart_requested", restart_requested, bool),
|
||
("active_agents", active_agents, parse_active_agents),
|
||
("active_work", active_work, lambda v: list(v) if v else None),
|
||
("served_profiles", served_profiles, lambda v: list(v or [])),
|
||
("multiplex_standalone_reason", multiplex_standalone_reason, lambda v: str(v) if v else None),
|
||
("session_store", session_store, _coerce_session_store),
|
||
))
|
||
if platform is not _UNSET:
|
||
platform_payload = copy.deepcopy(payload["platforms"].get(platform, {}))
|
||
if not isinstance(platform_payload, dict):
|
||
platform_payload = {}
|
||
if platform_state == "connected":
|
||
needs_attention = False if needs_attention is _UNSET else needs_attention
|
||
retrying_since = None if retrying_since is _UNSET else retrying_since
|
||
_apply_set_fields(platform_payload, (
|
||
("state", platform_state, None), ("error_code", error_code, None),
|
||
("error_message", error_message, None),
|
||
("needs_attention", needs_attention, bool),
|
||
("retrying_since", retrying_since, None),
|
||
("metrics", platform_metrics, None),
|
||
("ingress_url", ingress_url, None),
|
||
("listener_base", listener_base, None),
|
||
))
|
||
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
|
||
_runtime_status_state = payload
|
||
return path, payload, previous_payload
|
||
|
||
|
||
def _emit_runtime_status_transition(
|
||
previous_payload: dict[str, Any], payload: dict[str, Any]
|
||
) -> None:
|
||
with contextlib.suppress(Exception):
|
||
from agent.monitoring.gateway_health import emit_runtime_status_transition
|
||
emit_runtime_status_transition(previous_payload, payload)
|
||
|
||
def write_runtime_status(
|
||
*, reload_existing: bool = False, wait_timeout: Optional[float] = None, **fields: Any,
|
||
) -> bool:
|
||
"""Synchronously persist status for CLI callers and off-loop startup.
|
||
|
||
``wait_timeout`` bounds how long the caller waits for durable persistence.
|
||
A timed-out update remains queued for the single background writer.
|
||
Keyword ``fields`` are those of ``_prepare_runtime_status_update``.
|
||
"""
|
||
with _runtime_status_state_lock:
|
||
path, payload, previous_payload = _prepare_runtime_status_update(
|
||
reload_existing=reload_existing, **fields)
|
||
writer = _get_runtime_status_writer()
|
||
generation = writer.submit(path, payload)
|
||
# Report the transition once it is queued (matching ``publish_runtime_status``): a
|
||
# timed-out update is still written by the background writer, so its transition happened.
|
||
_emit_runtime_status_transition(previous_payload, payload)
|
||
return writer.wait(generation, timeout=wait_timeout)
|
||
|
||
|
||
def publish_runtime_status(**fields: Any) -> int:
|
||
"""Merge and enqueue status without waiting for filesystem persistence.
|
||
|
||
Keyword ``fields`` are those of ``_prepare_runtime_status_update``.
|
||
"""
|
||
with _runtime_status_state_lock:
|
||
path, payload, previous_payload = _prepare_runtime_status_update(
|
||
load_existing=False, **fields)
|
||
generation = _get_runtime_status_writer().submit(path, payload)
|
||
_emit_runtime_status_transition(previous_payload, payload)
|
||
return generation
|
||
|
||
|
||
def read_runtime_status(path: Optional[Path] = None) -> Optional[dict[str, Any]]:
|
||
"""Read ``gateway_state.json``; ``path`` lets callers inspect another profile's file."""
|
||
return _read_json_file(path or _get_runtime_status_path())
|
||
|
||
|
||
# Max age of a ``gateway_state.json`` snapshot before its liveness claim is suspect: an older record
|
||
# outlived an ungracefully-killed writer (taskkill /F, OOM, power loss) — or, with the PID alive, the
|
||
# housekeeping thread that re-stamps ``updated_at`` every tick has wedged (#113372). 2x the 60 s
|
||
# housekeeping interval.
|
||
_RUNTIME_STATUS_STALE_TTL_S = 120
|
||
|
||
|
||
def runtime_status_is_stale(
|
||
record: Optional[dict[str, Any]], ttl_s: int = _RUNTIME_STATUS_STALE_TTL_S
|
||
) -> bool:
|
||
"""True when the snapshot's ``updated_at`` is older than ``ttl_s`` (or missing/unparseable)."""
|
||
return not isinstance(record, dict) or _marker_is_stale(record.get("updated_at") or "", ttl_s)
|
||
|
||
|
||
def runtime_status_heartbeat_age_s(record: Optional[dict[str, Any]]) -> Optional[int]:
|
||
"""Whole seconds since the snapshot's ``updated_at``; None when missing/unparseable (an
|
||
unparseable stamp is a stale *file*, not a wedged heartbeat)."""
|
||
updated_at = normalize_updated_at(record.get("updated_at")) if isinstance(record, dict) else None
|
||
if not updated_at:
|
||
return None
|
||
return max(0, int((datetime.now(timezone.utc) - datetime.fromisoformat(updated_at)).total_seconds()))
|
||
|
||
|
||
def runtime_status_pid_is_live(record: Optional[dict[str, Any]]) -> bool:
|
||
"""True when the snapshot's PID is alive and passes the start-time PID-reuse guard."""
|
||
return _live_pid_from_record(record) is not None
|
||
|
||
|
||
def parse_active_agents(raw: Any) -> int:
|
||
"""Coerce ``active_agents`` to a non-negative int; shared by writer and both HTTP readers."""
|
||
try:
|
||
return max(0, int(raw))
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
|
||
# Live, serving states: a valid begin-drain target. ``degraded`` is a serving gateway with a parked
|
||
# platform (a dead watchdog-stamped ``degraded`` is already excluded by ``gateway_running=False``).
|
||
_DRAINABLE_GATEWAY_STATES = frozenset({"running", "degraded"})
|
||
|
||
|
||
def derive_gateway_busy(*, gateway_running: bool, gateway_state: Any, active_agents: Any) -> bool:
|
||
"""Busy iff live, serving (``running``/``degraded``), and ``active_agents > 0`` -- the contract NAS gates on. Liveness
|
||
keys off ``gateway_running``, NEVER ``updated_at`` (a stale heartbeat is a health warning, not death)."""
|
||
if not derive_gateway_drainable(gateway_running=gateway_running, gateway_state=gateway_state):
|
||
return False
|
||
return parse_active_agents(active_agents) > 0
|
||
|
||
|
||
def derive_gateway_drainable(*, gateway_running: bool, gateway_state: Any) -> bool:
|
||
"""Drainable iff live and serving; independent of ``active_agents`` (idle drains finish)."""
|
||
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"."""
|
||
|
||
running: bool
|
||
pid: Optional[int]
|
||
source: str
|
||
health_body: Optional[dict[str, Any]] = None
|
||
probe_error: bool = False
|
||
# The multiplexer's own ``gateway_state.json`` when the ``multiplexer`` rung answered: a served
|
||
# profile writes no runtime record of its own, so its platform states live there under
|
||
# ``<profile>:<platform>`` keys.
|
||
runtime: Optional[dict[str, Any]] = None
|
||
|
||
|
||
def profile_name_for_home(profile_home: Path) -> Optional[str]:
|
||
"""Profile id of any Hermes home: ``<root>/profiles/<name>`` → ``<name>``, the default root →
|
||
``"default"``, anything else → None. Multiplex-only makes ``default`` an ordinary served
|
||
profile, so reporting surfaces need a name for it too."""
|
||
home = Path(profile_home)
|
||
named = _profile_name_for_home(home)
|
||
if named:
|
||
return named
|
||
try:
|
||
from hermes_constants import get_default_hermes_root
|
||
if home.resolve() == Path(get_default_hermes_root()).resolve():
|
||
return "default"
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def multiplexer_liveness_for_profile(profile_dir: Path) -> Optional[tuple[int, dict[str, Any]]]:
|
||
"""``(pid, host gateway_state.json)`` when the ONE live host gateway serves the profile whose home
|
||
is ``profile_dir``; None for a home it does not serve or when no gateway owns the host role.
|
||
|
||
Multiplex-only: ``default`` is just another served profile, not the owner of a private topology —
|
||
resolving from the host rendezvous record (``gateway/host_topology.py``) is what lets it be
|
||
reported as SERVED rather than only as owner. A served profile owns no
|
||
``gateway.pid``/``gateway_state.json`` (#97120), so every PID-file rung of the dashboard ladder
|
||
reports it stopped while ``hermes -p X status`` says running — the two must agree.
|
||
"""
|
||
name = profile_name_for_home(Path(profile_dir))
|
||
if not name:
|
||
return None
|
||
from gateway.host_topology import host_gateway_topology
|
||
from hermes_cli.gateway import named_profile_served_by_running_multiplexer
|
||
from hermes_cli.gateway_multiplex_served import live_default_gateway_pid
|
||
from hermes_constants import get_default_hermes_root
|
||
# The roster is matched by NAME, and the multiplexer only serves ``<default root>/profiles/<name>``:
|
||
# a profile directory copied to another root (sandbox, restore-from-backup) keeps the name but is
|
||
# not the home being served, so it must not borrow the multiplexer's PID.
|
||
if name != "default" and not _same_hermes_home(profile_dir, get_default_hermes_root() / "profiles" / name):
|
||
return None
|
||
topology = host_gateway_topology()
|
||
if topology is not None and topology.serves(name):
|
||
pid: Optional[int] = topology.pid
|
||
elif name != "default" and named_profile_served_by_running_multiplexer(name):
|
||
# Config-derived fallback for a record that predates ``served_profiles``.
|
||
pid = live_default_gateway_pid()
|
||
else:
|
||
return None
|
||
if pid is None:
|
||
return None
|
||
return pid, read_runtime_status(get_default_hermes_root() / "gateway_state.json") or {}
|
||
|
||
|
||
def shared_listener_mirror_platforms(runtime: Optional[dict[str, Any]], profile: str) -> dict[str, Any]:
|
||
"""Entries for the api_server/webhook mirrors a served ``profile`` gets from the DEFAULT's
|
||
listener. The multiplexer never builds those adapters for a secondary (``gateway.run_adapters``
|
||
skips them: ``SHARED_LISTENER_MIRROR_PLATFORMS``), so the record has no ``<profile>:api_server``
|
||
entry and every reader fell through to ``pending_restart`` — "Restart needed" forever while
|
||
``/p/<profile>/v1/...`` answered. Only a live default entry is mirrored; its state is the profile's
|
||
state, plus the ``/p/<profile>`` URL the client must actually call.
|
||
"""
|
||
from gateway.config import SHARED_LISTENER_MIRROR_PATHS, SHARED_LISTENER_MIRROR_PLATFORMS
|
||
plats = (runtime or {}).get("platforms")
|
||
if not profile or profile == "default" or not isinstance(plats, dict):
|
||
return {}
|
||
mirrored: dict[str, Any] = {}
|
||
for name in sorted(SHARED_LISTENER_MIRROR_PLATFORMS):
|
||
entry = plats.get(name)
|
||
if not isinstance(entry, dict) or entry.get("state") not in {"connected", "connecting", "retrying"}:
|
||
continue
|
||
# api_server and webhook bind separate ports; each mirror hangs off its own listener. A record
|
||
# from an older gateway carries no ``listener_base``: connected, URL unknown.
|
||
base = entry.get("listener_base")
|
||
url = f"{base}/p/{profile}{SHARED_LISTENER_MIRROR_PATHS.get(name, '')}" if isinstance(base, str) and base else None
|
||
mirrored[name] = {k: v for k, v in entry.items() if k != "listener_base"}
|
||
mirrored[name].update(ingress_url=url, mirrored_from="default")
|
||
return mirrored
|
||
|
||
|
||
def profile_platforms_from_multiplexer(runtime: Optional[dict[str, Any]], profile: str) -> dict[str, Any]:
|
||
"""The ``<profile>:<platform>`` entries of a multiplexer record, re-keyed to bare platform names — the
|
||
same shape a standalone gateway for ``profile`` writes into its own ``gateway_state.json`` — plus the
|
||
default listener's api_server/webhook mirrors the profile is served through (``ingress_url`` set)."""
|
||
plats = (runtime or {}).get("platforms")
|
||
if not isinstance(plats, dict):
|
||
return {}
|
||
prefix = f"{profile}:"
|
||
own = {key[len(prefix):]: value for key, value in plats.items()
|
||
if isinstance(key, str) and key.startswith(prefix) and isinstance(value, dict)}
|
||
return {**shared_listener_mirror_platforms(runtime, profile), **own}
|
||
|
||
|
||
def resolve_gateway_liveness(
|
||
*, profile_dir: Optional[Path] = None, runtime: Any = _UNSET,
|
||
health_probe: Optional[Callable[[], tuple[bool, Optional[dict[str, Any]]]]] = None,
|
||
use_cache: bool = True, pid_probe: Optional[Callable[..., Optional[int]]] = None,
|
||
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) 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); (4) for a named ``profile_dir`` only, the live default multiplexer that records the
|
||
profile in ``served_profiles`` (a served profile writes no identity files of its own). ``*_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``.
|
||
|
||
Before this existed, ``/api/status`` and ``/api/messaging/platforms`` each open-coded their own ladder
|
||
and disagreed on the same page load — the sidebar read "running" while the Channels page rendered "The
|
||
gateway is not running." Three deployments hit it: a cross-container gateway (only ``/api/status`` ran
|
||
the HTTP health probe), a profile-scoped dashboard (only ``/api/status`` passed the profile's paths, so
|
||
messaging borrowed another profile's runtime state — issue #71211), and a launch-service-managed gateway
|
||
with no PID file (only some callers used the runtime-status fallback).
|
||
"""
|
||
_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
|
||
probe_error = False
|
||
scoped = profile_dir is not None
|
||
|
||
def guarded(fn, *args, fallback=None, **kwargs):
|
||
nonlocal probe_error
|
||
try:
|
||
return fn(*args, **kwargs)
|
||
except Exception:
|
||
probe_error = True
|
||
return fallback
|
||
|
||
# Zero-arg call when unscoped: callers monkeypatch with zero-arg lambdas and
|
||
# /api/status's cache signature is keyed on the call shape.
|
||
pid = guarded(_pid_probe, profile_dir / "gateway.pid") if scoped else guarded(_pid_probe)
|
||
if pid is not None:
|
||
return GatewayLiveness(running=True, pid=pid, source="pid")
|
||
health_body: Optional[dict[str, Any]] = None
|
||
if health_probe is not None:
|
||
alive, health_body = guarded(health_probe, fallback=(False, None))
|
||
if alive:
|
||
# Display-only PID: it belongs to the remote container.
|
||
remote_pid = health_body.get("pid") if health_body else None
|
||
return GatewayLiveness(
|
||
running=True, pid=remote_pid, source="health", health_body=health_body
|
||
)
|
||
if runtime is _UNSET:
|
||
reader_kwargs = {"path": profile_dir / "gateway_state.json"} if scoped else {}
|
||
runtime = guarded(_runtime_reader, **reader_kwargs)
|
||
# A scoped home with no ``gateway_state.json`` gets an EMPTY record, never ``None``: ``None`` makes
|
||
# the probe re-read the PROCESS home's record and lend that gateway's PID to a home it does not
|
||
# own (a profile directory copied out of another root).
|
||
probe_kwargs = {"expected_home": profile_dir} if scoped else {}
|
||
runtime_pid = guarded(_runtime_pid_probe, {} if (scoped and runtime is None) else runtime, **probe_kwargs)
|
||
if runtime_pid is not None:
|
||
return GatewayLiveness(
|
||
running=True, pid=runtime_pid, source="runtime_status", health_body=health_body
|
||
)
|
||
# (4) A named profile served by the live default multiplexer: no identity files of its own, but
|
||
# the multiplexer IS its gateway (mirrors `hermes -p X status` / `gateway list`). Unscoped, the
|
||
# question is about the process's OWN home — which is a named profile inside a pooled
|
||
# `hermes --profile X serve` (the Desktop's per-profile backend answers its REST without
|
||
# `?profile=`), so it takes the same rung instead of reporting the served profile stopped.
|
||
own_home = profile_dir if scoped else _get_process_hermes_home()
|
||
served = guarded(multiplexer_liveness_for_profile, own_home)
|
||
if served is not None:
|
||
mux_pid, mux_runtime = served
|
||
return GatewayLiveness(
|
||
running=True, pid=mux_pid, source="multiplexer", health_body=health_body, runtime=mux_runtime
|
||
)
|
||
return GatewayLiveness(
|
||
running=False, pid=None, source="none", health_body=health_body, probe_error=probe_error
|
||
)
|
||
|
||
|
||
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: 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
|
||
if payload.get("gateway_state") in {None, "stopped", "startup_failed"}:
|
||
return None
|
||
pid = _live_pid_from_record(payload)
|
||
if pid is None:
|
||
return None
|
||
# The record's hermes_home must match the home asked about (this process unscoped) so a stale
|
||
# or copied record cannot lend another home's gateway identity; legacy records without the
|
||
# stamp prove nothing either way and fall through to the live command-line check.
|
||
if expected_home is None and not _pid_record_belongs_to_current_profile(payload):
|
||
return None
|
||
if expected_home is not None and recorded_gateway_home_conflicts(payload, expected_home=expected_home):
|
||
return None
|
||
if not _record_matches_live_gateway_pid(payload, pid, expected_home=expected_home):
|
||
return None
|
||
return pid
|
||
|
||
|
||
def live_gateway_pid_for_home(home: Path) -> Optional[int]:
|
||
"""Verified PID of the gateway owned by ``home`` (pid file + runtime lock first, then the runtime
|
||
status record), or None. Every reader of another home's gateway identity goes through this so
|
||
they all prove the same thing: the PID passes the start-time reuse guard, its live command line is
|
||
a gateway's belonging to ``home``, and the record is not ``stopped``. Bare PID existence is not
|
||
identity -- a stale record whose PID was recycled by an unrelated process lent it ``served_profiles``
|
||
and put phantom gateways into the update inventory (#109680) -- while a launch-service gateway whose
|
||
``gateway.pid`` was unlinked is still live (#110166). Never unlinks ``home``'s identity files."""
|
||
home = Path(home)
|
||
# Cached: dashboard surfaces poll this for every served profile; the cache invalidates on any
|
||
# pid/lock file change, so a stopped or replaced gateway is seen at once.
|
||
pid = get_running_pid_cached(home / "gateway.pid", cleanup_stale=False)
|
||
if pid is not None:
|
||
return pid
|
||
return get_runtime_status_running_pid(read_runtime_status(home / "gateway_state.json"), expected_home=home)
|
||
|
||
|
||
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."""
|
||
with contextlib.suppress(Exception):
|
||
path = _get_pid_path()
|
||
file_pid = _pid_from_record(_read_json_file(path))
|
||
if file_pid is not None and file_pid != os.getpid():
|
||
return # Belongs to a different process — leave it alone.
|
||
path.unlink(missing_ok=True)
|
||
_clear_running_pid_cache()
|
||
|
||
|
||
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: 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")
|
||
current_start = _get_process_start_time(existing_pid)
|
||
if _start_times_conflict(recorded_start, current_start):
|
||
return True
|
||
if not _looks_like_gateway_process(existing_pid):
|
||
if _read_process_cmdline(existing_pid) is not None:
|
||
return True
|
||
if None in (recorded_start, current_start) and not _record_looks_like_gateway(existing):
|
||
return True
|
||
return _process_is_stopped(existing_pid)
|
||
|
||
|
||
def _process_is_stopped(pid: int) -> bool:
|
||
"""True for a stopped / tracing-stop state (T/t) in ``/proc/<pid>/status``."""
|
||
with contextlib.suppress(OSError):
|
||
for line in Path(f"/proc/{pid}/status").read_text(encoding="utf-8").splitlines():
|
||
if line.startswith("State:"):
|
||
return line.split()[1] in {"T", "t"}
|
||
return False
|
||
|
||
|
||
def acquire_scoped_lock(
|
||
scope: str, identity: str, metadata: Optional[dict[str, Any]] = None
|
||
) -> tuple[bool, Optional[dict[str, Any]]]:
|
||
"""Acquire a machine-local lock keyed by scope + identity (one Telegram token across homes)."""
|
||
lock_path = _get_scope_lock_path(scope, identity)
|
||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||
record = {
|
||
**_build_pid_record(), "scope": scope, "identity_hash": _scope_hash(identity),
|
||
"metadata": metadata or {}, "updated_at": _utc_now_iso(),
|
||
}
|
||
# Profile label for cross-profile conflict diagnostics ("token already in use (PID 559)" alone
|
||
# does not say WHICH profile). Omitted when not inferable; readers fall back to hermes_home.
|
||
profile = _profile_label_for_home(_get_process_hermes_home())
|
||
if profile:
|
||
record["profile"] = profile
|
||
existing = _read_json_file(lock_path)
|
||
if existing is None and lock_path.exists():
|
||
# Empty/invalid JSON: previous process died between O_EXCL create and json.dump().
|
||
_unlink_quietly(lock_path)
|
||
if existing:
|
||
existing_pid = _pid_from_record(existing)
|
||
# Our own PID: always self-reacquire. start_time guards reuse of OTHER PIDs; requiring
|
||
# equality here rejects reconnects when the on-disk record has start_time null.
|
||
# Same live PID as this process: always self-reacquire. ``start_time`` is a PID-reuse guard for
|
||
# *other* PIDs; it cannot distinguish two processes that share the caller's own PID (impossible
|
||
# while we are alive). Requiring start_time equality here falsely rejects reconnects when the
|
||
# on-disk record has ``start_time: null`` (older writers / psutil failure at first write) while the
|
||
# freshly built record has a real value — the gateway then reports itself as the foreign squatter of
|
||
# its own token (#81468).
|
||
if existing_pid == os.getpid():
|
||
_write_json_file(lock_path, record)
|
||
return True, existing
|
||
if not _scoped_lock_record_is_stale(existing, existing_pid):
|
||
return False, existing
|
||
# Rename to a tombstone instead of unlink(): with unlink()+O_EXCL two racing starters
|
||
# could both win. os.replace() lets exactly one claim it; a failed replace means another
|
||
# racer claimed it and O_EXCL below decides.
|
||
with contextlib.suppress(OSError):
|
||
tombstone = lock_path.with_name(lock_path.name + ".stale")
|
||
os.replace(lock_path, tombstone)
|
||
_unlink_quietly(tombstone)
|
||
try:
|
||
_write_json_excl(lock_path, record)
|
||
except FileExistsError:
|
||
return False, _read_json_file(lock_path)
|
||
return True, None
|
||
|
||
|
||
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."""
|
||
lock_path = _get_scope_lock_path(scope, identity)
|
||
if (_read_json_file(lock_path) or {}).get("pid") == os.getpid():
|
||
_unlink_quietly(lock_path)
|
||
|
||
|
||
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)."""
|
||
lock_dir = _get_lock_dir()
|
||
if not lock_dir.exists():
|
||
return 0
|
||
removed = 0
|
||
for lock_file in lock_dir.glob("*.lock"):
|
||
if owner_pid is not None:
|
||
record = _read_json_file(lock_file) or {}
|
||
if _pid_from_record(record) != owner_pid or (
|
||
owner_start_time is not None and record.get("start_time") != owner_start_time
|
||
):
|
||
continue
|
||
with contextlib.suppress(OSError):
|
||
lock_file.unlink(missing_ok=True)
|
||
removed += 1
|
||
return removed
|
||
|
||
|
||
# ── --replace takeover marker ─────────────────────────────────────────
|
||
# SIGTERM exits the gateway with code 1 so Restart=on-failure revives it after unexpected
|
||
# kills -- which would also revive a --replace target (flap loop against the replacer). The
|
||
# replacer therefore writes a short-lived marker naming the target PID + start_time BEFORE
|
||
# SIGTERM; the target's shutdown handler treats a matching marker as a planned takeover and
|
||
# exits 0. Unlinked once consumed, so a stale one can grief at most one future shutdown on
|
||
# the same PID, within _TAKEOVER_MARKER_TTL_S.
|
||
# When a new gateway starts with ``--replace``, it SIGTERMs the existing gateway so it can take over the bot
|
||
# token. ``hermes.service`` + ``hermes- gateway.service``). See #5646.
|
||
_TAKEOVER_MARKER_FILENAME = ".gateway-takeover.json"
|
||
_TAKEOVER_MARKER_TTL_S = 60 # Marker older than this is treated as stale
|
||
_PLANNED_STOP_MARKER_FILENAME = ".gateway-planned-stop.json"
|
||
_PLANNED_STOP_MARKER_TTL_S = 60
|
||
|
||
|
||
def _get_takeover_marker_path(hermes_home: Optional[Path] = None) -> Path:
|
||
"""Takeover marker path; ``hermes_home`` is given only for a verified cross-home handoff."""
|
||
home = _canonical_hermes_home(hermes_home or _get_process_hermes_home())
|
||
return home / _TAKEOVER_MARKER_FILENAME
|
||
|
||
|
||
def _get_planned_stop_marker_path() -> Path:
|
||
return _get_process_hermes_home() / _PLANNED_STOP_MARKER_FILENAME
|
||
|
||
|
||
def _marker_is_stale(written_at: str, ttl_s: int) -> bool:
|
||
try:
|
||
age = datetime.now(timezone.utc) - datetime.fromisoformat(written_at)
|
||
return age.total_seconds() > ttl_s
|
||
except (TypeError, ValueError):
|
||
return True
|
||
|
||
|
||
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 (must not wedge a new instance)."""
|
||
record = _read_json_file(path)
|
||
if not record:
|
||
return None
|
||
target_pid = _pid_from_record(record, "target_pid")
|
||
if target_pid is None or _marker_is_stale(record.get("written_at") or "", ttl_s):
|
||
_unlink_quietly(path)
|
||
return None
|
||
return record, target_pid, record.get("target_start_time")
|
||
|
||
|
||
def _pid_marker_names_self(target_pid: int, target_start_time: Any) -> bool:
|
||
"""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)
|
||
return None in (target_start_time, our_start_time) or target_start_time == our_start_time
|
||
|
||
|
||
def _consume_pid_marker_for_self(path: Path, *, ttl_s: int) -> bool:
|
||
parsed = _read_live_pid_marker(path, ttl_s)
|
||
if parsed is None:
|
||
return False
|
||
record, target_pid, target_start_time = parsed
|
||
# Cross-profile guard: new markers name the verified TARGET home, which permits a deliberate
|
||
# cross-HERMES_HOME --replace while ignoring a marker accidentally written into another
|
||
# profile's directory. Legacy markers have no target field: keep the same-replacer-home rule.
|
||
# See #29092.
|
||
our_home = _get_process_hermes_home()
|
||
target_home = record.get("target_hermes_home")
|
||
if target_home is not None:
|
||
if not isinstance(target_home, str) or not _same_hermes_home(target_home, our_home):
|
||
return False
|
||
else:
|
||
replacer_home = record.get("replacer_hermes_home")
|
||
if replacer_home is not None and not _same_hermes_home(replacer_home, our_home):
|
||
return False
|
||
matches = _pid_marker_names_self(target_pid, target_start_time)
|
||
_unlink_quietly(path)
|
||
return matches
|
||
|
||
|
||
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) + 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:
|
||
target_start_time = _get_process_start_time(target_pid)
|
||
return _write_marker(_get_takeover_marker_path(marker_home), {
|
||
"target_pid": target_pid, "target_start_time": target_start_time,
|
||
"target_hermes_home": str(marker_home), "replacer_pid": os.getpid(),
|
||
"replacer_hermes_home": str(_canonical_hermes_home(_get_process_hermes_home())),
|
||
"written_at": _utc_now_iso(),
|
||
})
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _write_marker(path: Path, record: dict[str, Any]) -> bool:
|
||
"""Atomically write a marker record; False (never raise) on OS failure."""
|
||
try:
|
||
_write_json_file(path, record)
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def consume_takeover_marker_for_self() -> bool:
|
||
"""Consume the takeover marker; True => planned takeover (exit 0); unlinked on match/stale."""
|
||
return _consume_pid_marker_for_self(_get_takeover_marker_path(), ttl_s=_TAKEOVER_MARKER_TTL_S)
|
||
|
||
|
||
def clear_takeover_marker(target_home: Optional[Path] = None) -> None:
|
||
"""Remove the takeover marker unconditionally. Safe to call repeatedly."""
|
||
_unlink_quietly(_get_takeover_marker_path(target_home))
|
||
|
||
|
||
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."""
|
||
if not isinstance(record, dict) or not _record_looks_like_gateway(record):
|
||
return None
|
||
owner_pid = _pid_from_record(record)
|
||
owner_start_time = record.get("start_time")
|
||
raw_home = record.get("hermes_home")
|
||
if (
|
||
owner_pid is None or owner_pid <= 0 or owner_pid == os.getpid()
|
||
or not isinstance(owner_start_time, int) or isinstance(owner_start_time, bool)
|
||
or not isinstance(raw_home, str) or not raw_home.strip()
|
||
or not Path(raw_home).expanduser().is_absolute()
|
||
):
|
||
return None
|
||
target_home = _canonical_hermes_home(raw_home)
|
||
if _scoped_lock_owner_state(owner_pid, owner_start_time) != "same":
|
||
return None
|
||
live_cmdline = _read_process_cmdline(owner_pid)
|
||
if live_cmdline is not None and not looks_like_gateway_runtime_command_line(live_cmdline):
|
||
return None
|
||
# The target home's own PID record must corroborate the claim.
|
||
pid_record = _read_json_file(target_home / "gateway.pid") or {}
|
||
pid_record_home = pid_record.get("hermes_home")
|
||
if (
|
||
not _record_looks_like_gateway(pid_record)
|
||
or _pid_from_record(pid_record) != owner_pid
|
||
or pid_record.get("start_time") != owner_start_time
|
||
or not isinstance(pid_record_home, str)
|
||
or not _same_hermes_home(pid_record_home, target_home)
|
||
):
|
||
return None
|
||
return owner_pid, owner_start_time, target_home
|
||
|
||
|
||
def _scoped_lock_owner_state(owner_pid: int, owner_start_time: int) -> str:
|
||
"""Return ``same``, ``exited``, or ``unknown`` for a validated owner."""
|
||
if not _pid_exists(owner_pid):
|
||
return "exited"
|
||
live_start_time = _get_process_start_time(owner_pid)
|
||
# A different start time means the PID was recycled; never signal the replacement.
|
||
if live_start_time is None:
|
||
return "unknown"
|
||
return "same" if live_start_time == owner_start_time else "exited"
|
||
|
||
|
||
def _wait_for_scoped_lock_owner_exit(
|
||
owner_pid: int, owner_start_time: int, *, attempts: int, delay: float
|
||
) -> tuple[bool, bool]:
|
||
"""Return ``(exited, safe_to_force)`` after bounded identity-aware waits."""
|
||
for _ in range(max(0, attempts)):
|
||
state = _scoped_lock_owner_state(owner_pid, owner_start_time)
|
||
if state == "exited":
|
||
return True, False
|
||
if state == "unknown":
|
||
return False, False
|
||
time.sleep(max(0.0, delay))
|
||
return False, _scoped_lock_owner_state(owner_pid, owner_start_time) == "same"
|
||
|
||
|
||
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)."""
|
||
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)
|
||
return []
|
||
|
||
|
||
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; 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:
|
||
if not child.is_running() or child.status() == psutil.STATUS_ZOMBIE:
|
||
continue
|
||
if child.ppid() == parent_pid:
|
||
logger.debug("Skipping child PID %d of old gateway %d: parent still appears "
|
||
"alive", child.pid, parent_pid)
|
||
continue
|
||
child.terminate()
|
||
live.append(child)
|
||
except psutil.NoSuchProcess:
|
||
continue
|
||
except Exception:
|
||
logger.debug("Could not terminate child PID %s of old gateway %d",
|
||
getattr(child, "pid", "?"), parent_pid, exc_info=True)
|
||
if not live:
|
||
return 0
|
||
gone, alive = psutil.wait_procs(live, timeout=max(0.0, timeout))
|
||
reaped = len(gone)
|
||
for child in alive:
|
||
try:
|
||
child.kill()
|
||
reaped += 1
|
||
except Exception:
|
||
logger.debug("Could not force-kill child PID %s of old gateway %d",
|
||
getattr(child, "pid", "?"), parent_pid, exc_info=True)
|
||
if reaped:
|
||
logger.info("Reaped %d orphaned child process(es) of replaced gateway PID %d.",
|
||
reaped, parent_pid)
|
||
except Exception:
|
||
logger.debug("Child reap for replaced gateway PID %d failed", parent_pid, exc_info=True)
|
||
return reaped
|
||
|
||
|
||
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 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)
|
||
if not write_takeover_marker(
|
||
owner_pid, target_home=target_home, target_start_time=owner_start_time
|
||
):
|
||
return None
|
||
try:
|
||
replaced = _terminate_verified_owner(
|
||
owner_pid, owner_start_time, graceful_attempts=graceful_attempts,
|
||
force_attempts=force_attempts,
|
||
)
|
||
finally:
|
||
# The target normally consumes the marker; clean up any remainder.
|
||
clear_takeover_marker(target_home)
|
||
if replaced is not None:
|
||
reap_gateway_children(owner_children, parent_pid=owner_pid)
|
||
return replaced
|
||
|
||
|
||
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; 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
|
||
if state != "same":
|
||
return None
|
||
for attempts, delay, kwargs in (
|
||
(graceful_attempts, 0.5, {"force": False}),
|
||
(force_attempts, 0.25, {"force": True, "expected_start_time": owner_start_time}),
|
||
):
|
||
try:
|
||
terminate_pid(owner_pid, **kwargs)
|
||
except ProcessLookupError:
|
||
return owner_pid
|
||
except OSError:
|
||
return None
|
||
exited, safe_to_force = _wait_for_scoped_lock_owner_exit(
|
||
owner_pid, owner_start_time, attempts=attempts, delay=delay
|
||
)
|
||
if exited:
|
||
return owner_pid
|
||
if not safe_to_force:
|
||
return None
|
||
return None
|
||
|
||
|
||
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 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),
|
||
"stopper_pid": os.getpid(), "written_at": _utc_now_iso(),
|
||
})
|
||
|
||
|
||
def consume_planned_stop_marker_for_self() -> bool:
|
||
"""Return True when the current process is being intentionally stopped."""
|
||
return _consume_pid_marker_for_self(
|
||
_get_planned_stop_marker_path(), ttl_s=_PLANNED_STOP_MARKER_TTL_S
|
||
)
|
||
|
||
|
||
def planned_stop_marker_targets_self() -> bool:
|
||
"""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])
|
||
|
||
|
||
def get_running_pid(
|
||
pid_path: Optional[Path] = None, *, cleanup_stale: bool = True
|
||
) -> Optional[int]:
|
||
"""PID of a running gateway (lock + PID file verified against the live process), or None.
|
||
An explicit ``pid_path`` is a scoped query into that home's identity files: records are
|
||
validated against the probed home (not the serve process's), and a live record is never
|
||
cleanup-unlinked, so polling another profile must not delete its gateway.pid/gateway.lock
|
||
(#106406). The unscoped path keeps main's poison-file housekeeping: a live record owned by
|
||
another home inside this home's gateway.pid is unlinked on refusal (#89315)."""
|
||
resolved_pid_path = pid_path or _get_pid_path()
|
||
resolved_lock_path = _get_gateway_lock_path(resolved_pid_path)
|
||
if is_gateway_runtime_lock_active(resolved_lock_path):
|
||
records = (
|
||
_read_pid_record(resolved_pid_path), _read_gateway_lock_record(resolved_lock_path),
|
||
)
|
||
expected_home = pid_path.parent if pid_path is not None else None
|
||
saw_live_pid = False
|
||
for record in records:
|
||
pid = _live_pid_from_record(record)
|
||
if pid is None:
|
||
continue
|
||
home_ok = (
|
||
_pid_record_belongs_to_current_profile(record) if expected_home is None
|
||
else not recorded_gateway_home_conflicts(record, expected_home=expected_home)
|
||
)
|
||
if home_ok and _record_matches_live_gateway_pid(
|
||
record, pid, expected_home=expected_home
|
||
):
|
||
return pid
|
||
# Scoped only: a live record we could not adopt may still be a real gateway;
|
||
# unlinking its identity files would break that home's double-run protection
|
||
# while the PID is alive. Unscoped keeps the #89315 poison-file cleanup.
|
||
saw_live_pid = True
|
||
if expected_home is None or not saw_live_pid:
|
||
_cleanup_invalid_pid_path(resolved_pid_path, cleanup_stale=cleanup_stale)
|
||
return get_runtime_status_running_pid() if pid_path is None else None
|
||
# Lock inactive: the runtime-status fallback runs BEFORE cleanup here.
|
||
runtime_pid = get_runtime_status_running_pid() if pid_path is None else None
|
||
if runtime_pid is None:
|
||
_cleanup_invalid_pid_path(resolved_pid_path, cleanup_stale=cleanup_stale)
|
||
return runtime_pid
|
||
|
||
|
||
def get_running_pid_identity_strict(pid_path: Path) -> Optional[tuple[int, float]]:
|
||
"""Return a verified process identity or fail on ambiguous runtime state."""
|
||
resolved_pid_path = Path(pid_path)
|
||
resolved_lock_path = _get_gateway_lock_path(resolved_pid_path)
|
||
pid_exists = _strict_path_exists(resolved_pid_path, "gateway PID")
|
||
# A stale PID file without a lock is not a live gateway; the lock probe is authoritative
|
||
# for absence.
|
||
if not _strict_path_exists(resolved_lock_path, "gateway lock"):
|
||
return None
|
||
if not _is_gateway_runtime_lock_active_strict(resolved_lock_path):
|
||
return None
|
||
if not pid_exists:
|
||
raise RuntimeError("active gateway lock has no PID metadata")
|
||
records = (_read_pid_record(resolved_pid_path), _read_gateway_lock_record(resolved_lock_path))
|
||
if not all(records):
|
||
raise RuntimeError("gateway PID or lock metadata is malformed")
|
||
pid = _pid_from_record(records[0])
|
||
if pid is None or pid <= 0 or _pid_from_record(records[1]) != pid:
|
||
raise RuntimeError("gateway PID and lock identities disagree")
|
||
if not _pid_exists(pid):
|
||
raise RuntimeError("gateway identity is not live")
|
||
current_start = _get_process_start_time(pid)
|
||
starts = tuple(record.get("start_time") for record in records)
|
||
if current_start is None or any(start is None for start in starts):
|
||
raise RuntimeError("gateway creation time is unavailable")
|
||
try:
|
||
if not _start_times_agree(current_start, *starts):
|
||
raise RuntimeError("gateway process identity changed")
|
||
except (TypeError, ValueError) as exc:
|
||
raise RuntimeError("gateway creation time is malformed") from exc
|
||
if not all(_record_matches_live_gateway_pid(record, pid) for record in records):
|
||
raise RuntimeError("runtime metadata does not identify a live gateway")
|
||
current = float(current_start)
|
||
if not _IS_WINDOWS:
|
||
return pid, current
|
||
# Windows persists a centisecond fingerprint; SCM checks need the exact psutil epoch.
|
||
# Re-read only after validation and prove it rounds to the same value.
|
||
try:
|
||
import psutil # type: ignore
|
||
|
||
exact_create_time = float(psutil.Process(pid).create_time())
|
||
except Exception as exc:
|
||
raise RuntimeError("exact gateway creation time is unavailable") from exc
|
||
if int(round(exact_create_time * 100)) != int(current):
|
||
raise RuntimeError("gateway process identity changed")
|
||
return pid, exact_create_time
|
||
|
||
|
||
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 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()
|
||
include_runtime_status = pid_path is None
|
||
# The signature covers the PID file, its sibling lock and (unscoped) the runtime status file.
|
||
watched = [resolved_pid_path, _get_gateway_lock_path(resolved_pid_path)]
|
||
if include_runtime_status:
|
||
watched.append(_get_runtime_status_path())
|
||
signature = tuple(_file_cache_signature(p) for p in watched)
|
||
key = (str(resolved_pid_path), bool(cleanup_stale), include_runtime_status)
|
||
now = time.monotonic()
|
||
with _gateway_running_pid_cache_lock:
|
||
cached = _gateway_running_pid_cache.get(key)
|
||
if cached is not None and now - cached[0] <= ttl_seconds and cached[1] == signature:
|
||
return cached[2]
|
||
pid = get_running_pid(pid_path, cleanup_stale=cleanup_stale)
|
||
refreshed_signature = tuple(_file_cache_signature(p) for p in watched)
|
||
with _gateway_running_pid_cache_lock:
|
||
_gateway_running_pid_cache[key] = (time.monotonic(), refreshed_signature, pid)
|
||
return pid
|
||
|
||
|
||
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
||
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
||
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
||
# The whole block is removed by reverting the commit that added it.
|
||
|
||
def clear_planned_stop_marker() -> None:
|
||
"""Remove the planned-stop marker unconditionally."""
|
||
try:
|
||
_get_planned_stop_marker_path().unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
def is_gateway_running(
|
||
pid_path: Optional[Path] = None,
|
||
*,
|
||
cleanup_stale: bool = True,
|
||
) -> bool:
|
||
"""Check if the gateway daemon is currently running."""
|
||
return get_running_pid(pid_path, cleanup_stale=cleanup_stale) is not None
|
||
# ---- END PLUGIN-COMPAT ----
|