Files
kshitijk4poor a898c70a43 docs(process): the spawn_owned docs name release_process_group_once_empty for a stop
The layer-4 header and the spawn_owned docstring told callers to end a stop with
release_process_groups, which drops coverage at once, even for group members that survived the
stop. Every spawn_owned caller releases through release_process_group_once_empty (or
stop_owned_group_survivors), and release_process_groups is for handing a group back on purpose
(a finished terminal command, MCP after its own reap). The comments now say so.
2026-10-29 16:58:09 +05:30

1146 lines
51 KiB
Python

"""Process identity: spawn tags, the machine-wide spawn ledger, the Windows job-object self-attach,
and the POSIX owner-death supervisor.
Four layers make every long-lived Hermes process positively identifiable, so reapers (``hermes
update``, Desktop startup sweeps) never guess lineage from PPID archaeology or cmdline matching,
and make every long-lived CHILD die with the Hermes process that owns it:
1. spawn tags (``HERMES_SPAWN`` env stamped by the spawner); 2. a ``(pid, create_time)`` ledger;
3. Windows job-object self-attach with ``KILL_ON_JOB_CLOSE`` so the whole child tree dies with the
root — no launcher→worker chains left holding ``.pyd`` locks; 4. on POSIX, one supervisor per
Hermes process that reaps the owned children's process groups the moment Hermes dies.
"""
from __future__ import annotations
import functools
import hashlib
import json
import logging
import os
import platform
import signal
import subprocess
import sys
import threading
import time
from contextlib import ExitStack, contextmanager, suppress
from contextvars import ContextVar
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterator, Optional, Sequence
from utils import atomic_json_write
logger = logging.getLogger(__name__)
SPAWN_ENV_VAR = "HERMES_SPAWN"
_TAG_VERSION = "v1"
LEDGER_FILENAME = "spawn-ledger.json"
#: Purposes a reaper may treat as "safe to kill when the owner is gone".
#: Interactive processes (chat, REPLs) are deliberately NOT in this set.
REAPABLE_PURPOSES = frozenset({"serve", "dashboard", "gateway", "mcp-helper"})
#: Host-role owners a fresh backend may reap on a HELD_BY_OTHER conflict (#121964): the
#: serve/dashboard/gateway subset. The mcp-helper rung stays owned by reap_orphaned_mcp_helpers.
_BACKEND_OWNER_PURPOSES = REAPABLE_PURPOSES - {"mcp-helper"}
_IS_WINDOWS = platform.system() == "Windows"
# Module-global job handle: must live exactly as long as this process so the
# kernel closes it (and kills the job) when we die. Never close it manually.
_JOB_HANDLE = None
_LEDGER_LOCK = threading.Lock()
def install_id(project_root: Optional[Path] = None) -> str:
"""Stable 12-hex identifier for THIS install (derived from its path)."""
if project_root is None:
try:
from hermes_constants import PROJECT_ROOT as _root
project_root = Path(_root)
except Exception:
project_root = Path(__file__).resolve().parent.parent
try:
canonical = str(Path(project_root).resolve()).lower()
except OSError:
canonical = str(project_root).lower()
return hashlib.sha256(canonical.encode("utf-8", "replace")).hexdigest()[:12]
def process_birth(proc) -> float:
"""A managed worker's identity birth for the ``psutil.Process`` *proc*, the same in every process.
On macOS psutil shifts ``create_time()`` by ``abs(INIT_BOOT_TIME - boot_time())``
(``psutil._psosx.adjust_proc_create_time``), with a boot-time reference frozen when psutil
was imported in the CALLING process. After a clock step (NTP, sleep/wake) a long-lived
gateway owner and a freshly spawned worker read different births for the same process, and
the exact identity match refused every managed worker (#140580). The kernel's own start
(``monotonic=True``: the unadjusted ``kinfo_proc`` epoch start) is what every process sees.
Other platforms keep ``create_time()``: Linux and Windows apply no per-process reference.
Raises ``psutil.Error`` like ``create_time()``.
"""
import psutil
if getattr(psutil, "MACOS", False):
platform_proc = getattr(proc, "_proc", None)
if platform_proc is not None:
try:
return float(platform_proc.create_time(monotonic=True))
except TypeError: # a psutil without the monotonic read: fall back to the adjusted one
pass
return float(proc.create_time())
def worker_birth(proc) -> float:
"""The birth a managed worker's identity fences compare: its hello, its scope and the Kanban
owner receipt. :func:`process_birth`, except on Linux, where psutil's ``create_time()`` adds the
``/proc/stat`` btime it re-reads on every call. WSL2 re-anchors btime after host sleep, so the
owner read a live worker's birth shifted from the one it reported and refused it (KB-12). The
kernel's start ticks since boot (``monotonic=True``) never move. The spawn ledger keeps
:func:`process_birth`: its persisted epoch births must keep matching across an update.
Raises ``psutil.Error`` like ``create_time()``.
"""
import psutil
if getattr(psutil, "LINUX", False):
return float(proc._proc.create_time(monotonic=True))
return process_birth(proc)
def boot_start(proc) -> Optional[float]:
"""Seconds from boot to the start of the ``psutil.Process`` *proc* on Linux; ``None`` elsewhere.
The ledger's epoch ``create_time`` is :func:`process_birth`, which on Linux is the kernel's
start ticks plus the ``/proc/stat`` btime psutil re-reads on every call, so an NTP or VM-resume
clock step moves it by whole seconds and a live process stops matching its entry. Ledger
entries carry this alongside it (``boot_start`` / ``spawner_boot``) and compare it exactly.
Raises ``psutil.Error`` like ``create_time()``.
"""
import psutil
if getattr(psutil, "LINUX", False):
return float(proc._proc.create_time(monotonic=True))
return None
def _process_create_time(pid: Optional[int] = None) -> Optional[float]:
""":func:`process_birth` of ``pid`` (default: this process); ``None`` when psutil can't say."""
try:
import psutil
return process_birth(psutil.Process(os.getpid() if pid is None else pid))
except Exception:
return None
@functools.lru_cache(maxsize=1)
def _live_boot_id() -> Optional[str]:
"""The kernel's id for the running boot (Linux); ``None`` elsewhere or when unreadable."""
try:
return Path("/proc/sys/kernel/random/boot_id").read_text(encoding="utf-8").strip() or None
except OSError:
return None
# No live process has a negative boot_start, so an entry from an earlier boot never matches.
_EARLIER_BOOT = -1.0
def _entry_boot(entry: dict, key: str) -> Optional[float]:
"""The entry's boot-relative birth ``key`` (``boot_start``/``spawner_boot``) as exact proof.
Boot ticks restart at every boot while the ledger survives reboots (and early-boot services
get the same pid at the same tick), so they identify a process only within the boot that
recorded them: an entry from another boot is :data:`_EARLIER_BOOT`, matching nothing alive.
``None`` (epoch match) when either boot id is unknown.
"""
recorded, live = entry.get("boot_id"), _live_boot_id()
if not isinstance(recorded, str) or live is None:
return None
return entry.get(key) if recorded == live else _EARLIER_BOOT
def _verified_boot_start(pid: Optional[int], create_time: Optional[float]) -> Optional[float]:
""":func:`boot_start` of ``pid`` while it still matches ``create_time``; ``None`` when unproven
or when the boot is unidentifiable (ticks alone repeat across reboots)."""
if not isinstance(pid, int) or pid <= 0 or create_time is None or _live_boot_id() is None:
return None
try:
import psutil
proc = psutil.Process(pid)
return boot_start(proc) if _same_incarnation(proc, create_time) else None
except Exception: # health: allow BLE001 -- best-effort probe like _process_create_time: any failure is "unproven"
return None
# Layer 1 — spawn tags
@dataclass(frozen=True)
class SpawnTag:
install: str
purpose: str
spawner_pid: int
spawner_create: Optional[float]
def build_spawn_tag(purpose: str, *, project_root: Optional[Path] = None) -> str:
"""Value for the child's ``HERMES_SPAWN`` env var, stamped by the spawner."""
create = _process_create_time()
create_part = f"{create:.3f}" if create is not None else "-"
return ":".join((_TAG_VERSION, install_id(project_root), purpose, str(os.getpid()), create_part))
def spawn_env(purpose: str, *, project_root: Optional[Path] = None) -> dict[str, str]:
"""Env fragment a spawner merges into a child's environment."""
return {SPAWN_ENV_VAR: build_spawn_tag(purpose, project_root=project_root)}
def parse_spawn_tag(raw: object) -> Optional[SpawnTag]:
"""Parse a ``HERMES_SPAWN`` value; ``None`` for anything malformed."""
parts = raw.split(":") if isinstance(raw, str) else []
if len(parts) != 5 or parts[0] != _TAG_VERSION:
return None
_, install, purpose, pid_s, create_s = parts
if not install or not purpose:
return None
try:
pid = int(pid_s)
create = None if create_s == "-" else float(create_s)
except ValueError:
return None
return SpawnTag(install, purpose, pid, create) if pid > 0 else None
def spawned_by_this_process(pid: int) -> bool:
"""True when ``pid`` carries a spawn tag naming THIS process (pid and create time).
A shared daemon (one session or pid file every Hermes of this user can reach) is owned only by
the Hermes whose command actually started it: the CLI it ran was given ``spawn_env`` and the
daemon inherited it. A pid file proves what a process is, not who started it. Fails closed:
an unreadable environment (another user, a race) is not ours."""
import psutil
try:
tag = parse_spawn_tag(psutil.Process(int(pid)).environ().get(SPAWN_ENV_VAR))
except (psutil.Error, OSError):
return False
if tag is None or tag.spawner_pid != os.getpid():
return False
mine = _process_create_time()
return tag.spawner_create is None or mine is None or abs(tag.spawner_create - mine) < 2.0
# Layer 2 — spawn ledger
@dataclass
class LedgerEntry:
pid: int
create_time: Optional[float]
purpose: str
install: str
spawner_pid: Optional[int]
spawner_create: Optional[float]
registered_at: float
argv: str
# Structured launch identity a relauncher needs after an update, without parsing argv. Empty
# for purposes that don't supply it; readers must use .get() — older ledger files predate these.
host: str = ""
port: Optional[int] = None
profile: str = ""
hermes_home: str = ""
# `serve --isolated`: opted out of the host singleton (Desktop's SSH backend for another
# machine). Attach-first readers must never adopt it; argv is truncated, so this is canonical.
isolated: bool = False
# Linux boot-relative births (:func:`boot_start`) of the process and its spawner: exact
# identity across clock steps, within the boot named by ``boot_id`` (:func:`_entry_boot`).
# ``None`` elsewhere and in older entries (epoch match only).
boot_start: Optional[float] = None
spawner_boot: Optional[float] = None
boot_id: Optional[str] = None
def _ledger_path() -> Path:
"""Machine-root ledger path (shared by every profile of this install)."""
try:
from hermes_constants import get_default_hermes_root
return Path(get_default_hermes_root()) / LEDGER_FILENAME
except Exception:
from hermes_cli.config import get_hermes_home
return Path(get_hermes_home()) / LEDGER_FILENAME
def _read_ledger(path: Path) -> Optional[list[dict]]:
"""Entries list, ``[]`` for empty/missing, ``None`` for CORRUPT (never silently an empty roster).
Mirrors the #89298 contract: corrupt is a distinct state that must never be silently treated as an empty
roster.
"""
try:
text = path.read_text(encoding="utf-8-sig")
except FileNotFoundError:
return []
except (OSError, UnicodeError):
return None
if not text.strip():
return []
try:
parsed = json.loads(text)
except (ValueError, TypeError):
return None
return [e for e in parsed if isinstance(e, dict)] if isinstance(parsed, list) else None
def _read_ledger_or_quarantine(path: Path) -> Optional[list[dict]]:
"""Ledger entries; ``None`` after parking a corrupt file. Caller holds ``_LEDGER_LOCK``."""
entries = _read_ledger(path)
if entries is None:
parked = path.with_suffix(path.suffix + ".corrupt")
try:
os.replace(path, parked)
logger.warning("spawn ledger was unreadable; moved to %s", parked)
except OSError:
pass
return entries
def _same_incarnation(proc, create_time: Optional[float], boot: Optional[float] = None) -> bool:
"""Does the live ``proc`` match a recorded birth? A recorded ``boot`` (:func:`boot_start`)
compares exactly where the live one is readable; otherwise ``create_time`` (2 s tolerance;
``None`` matches)."""
if boot is not None and (live_boot := boot_start(proc)) is not None:
return live_boot == float(boot)
return create_time is None or abs(process_birth(proc) - float(create_time)) < 2.0
def _pid_alive_matches(
pid: int, create_time: Optional[float], *, strict: bool = False, boot: Optional[float] = None,
) -> Optional[bool]:
"""True/False when provable; ``None`` when psutil can't say."""
try:
import psutil
except Exception:
return None
try:
proc = psutil.Process(int(pid))
# ``strict`` is positive proof (an exact epoch birth); a clock step only withholds it.
if strict:
# Windows: skip the status() read entirely. psutil's Windows status path
# (NtQuerySystemInformation) does not release the GIL, so under load it freezes every
# other Python thread in the process for minutes (#142085) — and Windows has no POSIX
# zombies, so the STATUS_ZOMBIE comparison had nothing to contribute there anyway. The
# non-strict branch right below already makes this exact skip.
if os.name == "nt":
return (create_time is not None and process_birth(proc) == create_time
and proc.is_running())
return (create_time is not None and process_birth(proc) == create_time
and proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE)
# A zombie keeps its create_time until its parent reaps it, but it is already dead: the
# dashboard stop check (``gateway.status._pid_exists``) books it stopped, so the ledger must
# not report it as a live pre-update survivor. Windows has no zombies (and status() is slow there).
return _same_incarnation(proc, create_time, boot) and (
os.name == "nt" or proc.status() != getattr(psutil, "STATUS_ZOMBIE", "zombie"))
except psutil.NoSuchProcess:
return False
except Exception:
return None
def register_self(purpose: str, *, project_root: Optional[Path] = None, detail: Optional[dict] = None) -> bool:
"""Record this process in the machine spawn ledger. Best-effort.
Called at the top of every long-lived entry point; dead ``(pid, create_time)`` entries are
pruned on every write. ``detail`` may carry ``host``/``port``/``profile`` so the update
pipeline can relaunch a manually-started serve with its real bind address, and ``isolated``
so attach-first discovery skips a backend that opted out of the host singleton.
"""
from hermes_constants import hermes_home_key
tag = parse_spawn_tag(os.environ.get(SPAWN_ENV_VAR))
spawner_pid, spawner_create = (tag.spawner_pid, tag.spawner_create) if tag else _desktop_spawner_identity()
entry = _new_entry(os.getpid(), _process_create_time(), purpose, project_root, spawner_pid, spawner_create)
entry.spawner_boot = _verified_boot_start(spawner_pid, spawner_create)
entry.hermes_home = hermes_home_key()
if detail:
try:
entry.host = str(detail.get("host") or "")
entry.port = int(detail["port"]) if detail.get("port") is not None else None
entry.profile = str(detail.get("profile") or "")
entry.isolated = bool(detail.get("isolated"))
except (TypeError, ValueError):
pass
try:
import sys as _sys
# 10 tokens: enough for `hermes serve --host X --port N --profile P` while bounding
# pathological argv. Structured detail is canonical; argv is the human-readable fallback.
entry.argv = " ".join(_sys.argv[:10])
except Exception:
pass
return _append_entry(entry)
def is_desktop_owned_backend(argv: Optional[Sequence[str]] = None) -> bool:
"""Whether this process is the backend Desktop spawned and owns.
``HERMES_DESKTOP=1`` is inherited by every shell and agent child the app launches, so the
flag alone is not ownership proof (same class as #116107). Desktop hands its backend a
per-spawn credential the terminal pane never receives (and the terminal tool's env policy
strips from agent children): the local pool spawn mints ``HERMES_DASHBOARD_SESSION_TOKEN``,
the SSH spawn passes a 0600 token FILE on argv and deliberately sets no token env var.
``argv`` defaults to this process's own.
"""
if os.environ.get("HERMES_DESKTOP") != "1":
return False
if os.environ.get("HERMES_DASHBOARD_SESSION_TOKEN"):
return True
from hermes_cli._startup_fast import is_desktop_ssh_backend_argv
return is_desktop_ssh_backend_argv(list(sys.argv[1:] if argv is None else argv))
# Set while this process works for the Desktop app on its own machine: one request from, or one turn of
# a chat created by, a client ``local_desktop_client`` accepted. The gateway serves the Desktop, the CLI,
# the TUI and bots from one process, so this is a property of the activity, never of the process.
_SERVING_LOCAL_DESKTOP: ContextVar[bool] = ContextVar("hermes_serving_local_desktop", default=False)
@contextmanager
def serving_local_desktop(active: bool) -> Iterator[None]:
"""Mark the enclosed activity (and threads it hands its context to) as work for the local Desktop."""
token = _SERVING_LOCAL_DESKTOP.set(bool(active) or _SERVING_LOCAL_DESKTOP.get())
try:
yield
finally:
_SERVING_LOCAL_DESKTOP.reset(token)
def serves_local_desktop_now() -> bool:
return _SERVING_LOCAL_DESKTOP.get()
def local_desktop_client(identity: Optional[dict], declared: object) -> bool:
"""A connected client is the Desktop app on this machine when it says so (it knows its connection
mode) AND it redeemed a native ticket, which only this machine's private control socket mints, for
this OS user. Neither alone: the CLI and TUI hold native tickets too, and the claim is the client's
word. A remote Desktop reaches a backend through SSH, URL or Cloud credentials, never a native ticket."""
return (declared is True and isinstance(identity, dict) and identity.get("provider") == "local"
and identity.get("native_bootstrap") is True)
def _desktop_spawner_identity() -> tuple[Optional[int], Optional[float]]:
"""Spawner ``(pid, create_time)`` from the Electron app's HERMES_PARENT_PID (+ optional
``winms:<ms>`` start marker) parent-death watchdog vars, so ledger lineage works with every
Desktop version without a TS change. ``(None, None)`` when absent/malformed."""
try:
spawner_pid = int(os.environ.get("HERMES_PARENT_PID", ""))
except (TypeError, ValueError):
spawner_pid = 0
if spawner_pid <= 0:
return None, None
marker = os.environ.get("HERMES_PARENT_START_MARKER", "")
if not marker.startswith("winms:"):
return spawner_pid, None
try:
return spawner_pid, float(marker.split(":", 1)[1]) / 1000.0
except (ValueError, IndexError):
return spawner_pid, None
def _new_entry(
pid: int, create_time: Optional[float], purpose: str, project_root: Optional[Path],
spawner_pid: Optional[int], spawner_create: Optional[float],
) -> LedgerEntry:
return LedgerEntry(
pid, create_time, purpose, install_id(project_root), spawner_pid, spawner_create, time.time(), argv="",
boot_start=_verified_boot_start(pid, create_time), boot_id=_live_boot_id(),
)
# A ledger writer stopped mid-write (SIGSTOP, a debugger) must not stall every spawn_owned on the
# machine: past this the write goes ahead under the thread lock alone, as when no lock can be taken.
_LEDGER_LOCK_TIMEOUT_S = 2.0
_ledger_lock_warned = False
def _append_entry(entry: LedgerEntry) -> bool:
"""Prune dead entries and append ``entry`` — the ONLY ledger write path.
Serialized under ``_LEDGER_LOCK`` and, across processes, an exclusive lock on a sidecar file
(every ``spawn_owned`` writes here, so concurrent Hermes processes otherwise dropped each
other's rows), with an atomic tmp+replace; no writer touches the file outside this function.
See #91660.
"""
from hermes_cli.active_sessions import _FileLock
path = _ledger_path()
with _LEDGER_LOCK, ExitStack() as held:
try:
held.enter_context(_FileLock(path.with_name(path.name + ".lock"), timeout=_LEDGER_LOCK_TIMEOUT_S))
except (RuntimeError, OSError):
# Best-effort like the write itself: the thread lock is then the only guard, so a row another
# Hermes process writes meanwhile can be lost. Say so once where it shows (agent.log).
global _ledger_lock_warned
if not _ledger_lock_warned:
_ledger_lock_warned = True
logger.warning("spawn ledger lock unavailable (a stalled holder, or no file locks here); writing "
"without it, so concurrent Hermes processes may drop each other's rows")
logger.debug("spawn ledger lock unavailable", exc_info=True)
entries = _read_ledger_or_quarantine(path) or []
# Drop malformed entries, our own stale entry, and provably dead pids.
pruned = [
e for e in entries
if isinstance(e.get("pid"), int)
and e["pid"] != entry.pid
and entry_still_running(e)
]
pruned.append(asdict(entry))
try:
from hermes_constants import mkdir_under_hermes_home
mkdir_under_hermes_home(path.parent)
# argv may carry surrogate-escaped bytes (non-UTF-8 paths); ensure_ascii keeps the
# utf-8 text handle from raising UnicodeEncodeError (a ValueError, not an OSError).
atomic_json_write(path, pruned, mode=0o600, ensure_ascii=True)
return True
except OSError:
logger.debug("spawn ledger write failed", exc_info=True)
return False
def register_child(pid: int, purpose: str, *, project_root: Optional[Path] = None) -> bool:
"""Record a CHILD process this process just spawned. Best-effort.
Mirror of :func:`register_self` for children that cannot register themselves (stdio MCP
helpers: arbitrary ``npx``/binary servers never import Hermes code). Records the child's
``(pid, create_time)`` with THIS process as spawner, so a helper whose spawner is provably gone
is a reapable orphan and one whose spawner is alive is never reaped.
"""
try:
pid = int(pid)
except (TypeError, ValueError):
return False
child_create = _process_create_time(pid) if pid > 0 else None
if child_create is None:
return False
entry = _new_entry(pid, child_create, purpose, project_root, os.getpid(), _process_create_time())
entry.spawner_boot = _verified_boot_start(os.getpid(), entry.spawner_create)
try:
import psutil
entry.argv = " ".join(psutil.Process(pid).cmdline()[:10])
except Exception:
pass
return _append_entry(entry)
def ledger_entries(
*, project_root: Optional[Path] = None, all_installs: bool = False, verified_only: bool = False,
) -> list[dict]:
"""Ledger entries for this install, or all installs when explicitly requested.
``verified_only`` requires an exact live PID/create-time pair. The default preserves
unknown processes for reapers, which must not mistake missing proof for a dead process.
Entries whose ``(pid, create_time)`` no longer matches a live process are excluded (PID reuse reads as
dead, thanks to the create-time pair). A corrupt ledger is quarantined and read as empty — identical
philosophy to the backend-ownership fix (#89298): never let corruption erase or fake a roster; never let
it block the caller either.
"""
want_install = install_id(project_root)
with _LEDGER_LOCK:
entries = _read_ledger_or_quarantine(_ledger_path())
if entries is None:
return []
return [
e for e in entries
if (all_installs or e.get("install") == want_install)
and isinstance(e.get("pid"), int)
and (_pid_alive_matches(e["pid"], e.get("create_time"), strict=True) is True
if verified_only else _pid_alive_matches(e["pid"], e.get("create_time"), boot=_entry_boot(e, "boot_start")) is not False)
]
def entry_still_running(entry: dict) -> bool:
"""Is the ledger *entry*'s process still the recorded incarnation (``None`` from psutil
counts as running)? Exact boot ticks within their boot survive a clock step; an entry
without them keeps the 2 s epoch match, so a step reads it as a reused pid."""
pid = entry.get("pid")
if not isinstance(pid, int) or pid <= 0:
return False
return _pid_alive_matches(pid, entry.get("create_time"), boot=_entry_boot(entry, "boot_start")) is not False
def spawner_is_dead(entry: dict) -> Optional[bool]:
"""Is the recorded spawner of this entry provably gone? ``None`` when unrecorded/unprovable."""
spawner_pid = entry.get("spawner_pid")
if not isinstance(spawner_pid, int) or spawner_pid <= 0:
return None
# A spawner recorded in an earlier boot is dead (:func:`_entry_boot`), as the epoch match says.
boot = _entry_boot(entry, "spawner_boot")
alive = _pid_alive_matches(spawner_pid, entry.get("spawner_create"), boot=boot)
return None if alive is None else not alive
def _terminate_then_kill(pid: int, create_time: Optional[float], boot: Optional[float] = None) -> bool:
"""SIGTERM, 2 s grace, then SIGKILL. False when the incarnation moved on or is already gone."""
try:
import psutil
proc = psutil.Process(pid)
if not _same_incarnation(proc, create_time, boot):
return False # PID reused since registration — never signal a stranger
proc.terminate()
try:
proc.wait(timeout=2.0)
except psutil.TimeoutExpired:
proc.kill()
return True
except Exception:
logger.debug("orphan terminate failed for pid %s", pid, exc_info=True)
return False
def _reparented_orphan(pid: int) -> bool:
"""Null-spawner owner with no live supervisor: reparented to init, old, lock-unclaimed.
Mirrors ``_reap_orphaned_desktop_local_serves``: operator-managed remote backends legitimately
sit at ppid 1 (systemd, an exited sshd), so ppid alone is never proof — the SSH-lock claim
and the 180 s lock-write grace exclude live-supervised backends. Windows: the ppid probe is
unavailable there, so this rung never fires (desktop tree-kill reaps instead).
"""
try:
from hermes_cli.dashboard_procs import (
_REAP_MIN_AGE_SECONDS, _lock_owned_serve_pids, _process_ppid)
if _process_ppid(pid) not in (0, 1):
return False
if pid in (_lock_owned_serve_pids() or ()):
return False
import time as _time
import psutil
return max(0.0, _time.time() - psutil.Process(pid).create_time()) >= _REAP_MIN_AGE_SECONDS
except Exception:
return False # unprovable → never touch
def reap_orphaned_backend_owner(
pid: int, create_time: Optional[float], *, kill_fn=None) -> Optional[int]:
"""Kill the conflicting host owner when the ledger PROVES it is a dead session's orphan.
HELD_BY_OTHER rung for #121964: the owner PID is alive (re-parented to init), so
``record_is_stale()`` never fires and a fresh backend would loop observe-only against the
same orphan forever. Reaped only with a live ``serve``/``dashboard``/``gateway`` entry for
THIS install whose spawner is provably dead — or whose spawner was never recorded and which
is reparented to init, old, and claimed by no Desktop SSH lock. Live-spawned, lock-claimed,
young, interactive, self, or unprovable owners are never touched. Returns the reaped pid,
else ``None``; never raises.
# ponytail: a dead session's surviving backend.lock.json still shields its orphan (the lock
# has no session liveness); clearing stale locks is a Desktop-side lifecycle job, not this rung.
"""
try:
pid = int(pid)
except (TypeError, ValueError):
return None
if pid <= 0 or pid == os.getpid():
return None
try:
entry = next(
(e for e in ledger_entries()
if e.get("pid") == pid
and e.get("purpose") in _BACKEND_OWNER_PURPOSES
and (e.get("create_time") is None or create_time is None
or abs(float(e["create_time"]) - float(create_time)) < 2.0)),
None)
except Exception:
return None
if entry is None:
return None
if spawner_is_dead(entry) is not True and not (
entry.get("spawner_pid") is None and _reparented_orphan(pid)):
return None # live or unprovable spawner → never touch
try:
if kill_fn is not None:
kill_fn(pid)
elif not _terminate_then_kill(pid, entry.get("create_time"), _entry_boot(entry, "boot_start")):
return None
except Exception:
logger.debug("backend owner reap failed for %s", entry, exc_info=True)
return None
logger.info("reaped orphaned %s backend owner pid %s", entry.get("purpose"), pid)
return pid
def reap_orphaned_mcp_helpers(*, project_root: Optional[Path] = None, kill_fn=None) -> list[int]:
"""Kill ledger-registered stdio MCP helpers whose spawner is provably dead.
Ledger-driven startup-sweep rung (not cmdline-heuristic): a helper is reaped ONLY when it has a
live ``mcp-helper`` entry for THIS install AND ``spawner_is_dead`` is ``True`` — never
``None``/unprovable, never a live spawner.
"""
reaped: list[int] = []
try:
entries = ledger_entries(project_root=project_root)
except Exception:
return reaped
own_pid = os.getpid()
for entry in entries:
try:
pid = entry.get("pid")
if entry.get("purpose") != "mcp-helper" or not isinstance(pid, int) or pid <= 0 or pid == own_pid:
continue
if spawner_is_dead(entry) is not True:
continue # live or unprovable spawner → never touch
if kill_fn is not None:
kill_fn(pid)
else:
import psutil
proc = psutil.Process(pid)
if not _same_incarnation(proc, entry.get("create_time"), _entry_boot(entry, "boot_start")):
continue # PID reused since registration
# Windows: descendants (npx.cmd → node.exe) have no pgid to group-kill and
# reparent with ParentId=null when the direct child exits first (#61059), so
# reap the whole tree. On POSIX the killpg-based sweep already reaches them.
if _IS_WINDOWS:
_kill_process_tree_windows(proc)
else:
proc.terminate()
try:
proc.wait(timeout=2.0)
except psutil.TimeoutExpired:
proc.kill()
reaped.append(pid)
except Exception:
logger.debug("mcp-helper orphan reap failed for %s", entry, exc_info=True)
if reaped:
logger.info("reaped %d orphaned stdio MCP helper(s): %s", len(reaped), reaped)
return reaped
# Layer 3 — Windows job-object self-attach
def _kill_process_tree_windows(proc) -> None:
"""Terminate *proc* and every still-alive descendant (npx.cmd → node.exe), Windows-only
(#61059): without a pgid there is no group-kill, and grandchildren reparent to nothing
(ParentId=null) once the direct child exits, so they must be reached through the tree."""
import psutil
try:
descendants = proc.children(recursive=True)
except (psutil.NoSuchProcess, psutil.AccessDenied, OSError):
descendants = []
for child in descendants:
try:
child.terminate()
except Exception:
pass
try:
proc.terminate()
except Exception:
pass
try:
_, alive = psutil.wait_procs(descendants + [proc], timeout=2.0)
except Exception:
return
for survivor in alive:
try:
survivor.kill()
except Exception:
pass
def attach_self_to_kill_on_close_job() -> bool:
"""Place this process in a job that dies (whole tree) when we die. Windows-only, idempotent.
``BREAKAWAY_OK`` keeps children spawned with ``CREATE_BREAKAWAY_FROM_JOB`` (gateway relaunch
during update, detached watchers) escaping exactly as before.
"""
global _JOB_HANDLE
if not _IS_WINDOWS or _JOB_HANDLE is not None:
return _JOB_HANDLE is not None
try:
import ctypes
from ctypes import wintypes
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x2000
JOB_OBJECT_LIMIT_BREAKAWAY_OK = 0x0800
JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK = 0x1000
JobObjectExtendedLimitInformation = 9
class IO_COUNTERS(ctypes.Structure):
_fields_ = [(n, ctypes.c_ulonglong) for n in (
"ReadOperationCount", "WriteOperationCount", "OtherOperationCount",
"ReadTransferCount", "WriteTransferCount", "OtherTransferCount")]
class JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure):
_fields_ = [
("PerProcessUserTimeLimit", wintypes.LARGE_INTEGER), ("PerJobUserTimeLimit", wintypes.LARGE_INTEGER),
("LimitFlags", wintypes.DWORD), ("MinimumWorkingSetSize", ctypes.c_size_t),
("MaximumWorkingSetSize", ctypes.c_size_t), ("ActiveProcessLimit", wintypes.DWORD),
("Affinity", ctypes.POINTER(wintypes.ULONG)), ("PriorityClass", wintypes.DWORD),
("SchedulingClass", wintypes.DWORD),
]
class JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure):
_fields_ = [
("BasicLimitInformation", JOBOBJECT_BASIC_LIMIT_INFORMATION),
("IoInfo", IO_COUNTERS),
*((n, ctypes.c_size_t) for n in (
"ProcessMemoryLimit", "JobMemoryLimit", "PeakProcessMemoryUsed", "PeakJobMemoryUsed")),
]
job = kernel32.CreateJobObjectW(None, None)
if not job:
return False
info = JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
info.BasicLimitInformation.LimitFlags = (
JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE | JOB_OBJECT_LIMIT_BREAKAWAY_OK | JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK
)
ok = kernel32.SetInformationJobObject(job, JobObjectExtendedLimitInformation, ctypes.byref(info), ctypes.sizeof(info))
if not ok or not kernel32.AssignProcessToJobObject(job, kernel32.GetCurrentProcess()):
kernel32.CloseHandle(job)
return False
_JOB_HANDLE = job # keep alive for the life of the process — never close
logger.debug("attached to kill-on-close job object")
return True
except Exception:
logger.debug("job object self-attach failed", exc_info=True)
return False
# Layer 4 — POSIX owner-death supervisor
#
# If this process dies without running its cleanup path (SIGKILL, OOM killer, crash, force-quit),
# its long-lived children reparent to init or the nearest subreaper (systemd --user) and run
# forever; macOS has no PR_SET_PDEATHSIG, and on Linux it fires when the spawning THREAD exits,
# which kills children spawned from short-lived worker threads while the owner is alive. So ONE
# supervisor per Hermes process (``process_supervisor.sh``) holds the read end of a pipe whose only
# write end we hold: EOF means we died, by any cause, and it killpg()s every group still registered.
# A registered group that was never unregistered IS the orphan set, so a clean exit that stopped its
# children first ends with nothing to kill, and one that didn't takes them along.
#
# ``spawn_owned`` is THE way to start a long-lived child that must die with this process: the child
# leads its own process group, that group is handed to the supervisor, and the child is recorded in
# the spawn ledger. Children something else spawns (the MCP SDK, the browser-harness daemon) hand their
# group to ``own_process_groups`` where we learn the pgid; a descendant that leaves the group on
# purpose is registered by its own pgid the same way. Stopping an owned child yourself ends with
# ``release_process_group_once_empty`` (survivors stay covered); ``release_process_groups`` hands a
# group back on purpose. ``scripts/check_detached_spawns.py`` flags new detached spawns
# (start_new_session / process_group / setsid / setpgid; its docstring lists what it misses). POSIX-only: Windows children die
# through the layer-3 job object instead.
_supervisor = None # Optional[subprocess.Popen]; tests rebind it
_supervisor_lock = threading.Lock()
# Groups the supervisor is reaping on our behalf; replayed verbatim on respawn so a respawn never
# silently drops coverage for children that are still running.
_supervised_pgids: set = set()
_SUPERVISOR_SCRIPT = Path(__file__).with_name("process_supervisor.sh")
# Android (Termux) has no /bin/sh; its system shell is the same POSIX sh the script needs.
_POSIX_SH = "/bin/sh" if os.path.exists("/bin/sh") else "/system/bin/sh"
def _spawn_supervisor():
"""Start the per-process supervisor, or None if it cannot be started."""
try:
# start_new_session=True is load-bearing: shutdown paths killpg this process's own group,
# which would kill the supervisor before it could reap anything. close_fds keeps the write
# end of every OTHER pipe we hold out of it, so it never delays another child's EOF.
proc = subprocess.Popen(
[_POSIX_SH, str(_SUPERVISOR_SCRIPT), str(os.getpgid(0))],
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
start_new_session=True, close_fds=True, text=True, encoding="utf-8")
except Exception:
# Never let supervisor bookkeeping block a real child: graceful shutdown paths still reap
# normally; only the ungraceful-exit safety net is lost.
logger.debug("Could not start the owner-death supervisor", exc_info=True)
return None
_watch_supervisor(proc)
return proc
# A supervisor that dies sooner than this after starting may not be able to stay up (a broken install:
# sh exiting 2 or 127; a sandbox or OOM killer that keeps killing it): the waiter leaves it to the next
# registration change instead of restarting it in a loop. One killed by a signal gets one young restart
# (a new supervisor starts after every idle period, so a `pkill -f hermes` often meets a young one); if
# its replacement is killed young as well, that one waits for the next registration change too.
_SUPERVISOR_RESTART_MIN_UPTIME_S = 5.0
# Set (under _supervisor_lock) while the waiter starts the young restart, so that replacement's own
# waiter knows it already had one.
_starting_young_restart = False
def _watch_supervisor(proc) -> None:
"""One waiter thread per supervisor, blocked in the untimed ``wait()`` (waitpid, no polling). It
reaps the supervisor the moment it exits, so no zombie child is left for psutil walkers of this
process's children to trip on. A supervisor that died on its own while groups are still registered
(``pkill -f hermes``, the OOM killer) is restarted and replayed at once, so coverage does not wait
for the next registration change; a released one (EOF with an empty set) is only reaped."""
started = time.monotonic()
young_restart = _starting_young_restart
def wait_then_restart() -> None:
global _supervisor, _starting_young_restart
proc.wait()
with _supervisor_lock:
if _supervisor is not None and _supervisor is not proc:
return # already replaced: only reaped
# Still current, or dropped by a caller whose write hit its broken pipe: restart below if
# groups are still registered (a deliberate release leaves the set empty).
_supervisor = None
_prune_dead_supervised_pgids()
if not _supervised_pgids:
return
old_enough = time.monotonic() - started >= _SUPERVISOR_RESTART_MIN_UPTIME_S
young_signal = not old_enough and proc.returncode < 0 and not young_restart
restart = old_enough or young_signal
logger.warning(
"owner-death supervisor exited (status %s) with %d process group(s) registered; %s",
proc.returncode, len(_supervised_pgids),
"restarting it" if restart else "it could not stay up, retrying at the next registration")
if restart:
_starting_young_restart = young_signal
try:
_start_supervisor_locked()
finally:
_starting_young_restart = False
try:
# health: allow HX012 -- the thread waitpid()s one child and touches only process-global supervisor state
threading.Thread(target=wait_then_restart, name="owner-death-supervisor-waiter", daemon=True).start()
except RuntimeError: # "can't start new thread" (thread limit, interpreter shutdown)
# The supervisor still works; it is left a zombie until we exit and is not restarted if killed.
logger.debug("Could not start the owner-death supervisor's waiter", exc_info=True)
# An idle supervisor (nothing registered) exits this long after the set last emptied, unless a new
# registration arrives first.
_SUPERVISOR_IDLE_EXIT_S = 30.0
_idle_release: Optional[threading.Timer] = None
def _release_when_idle_locked(proc) -> None:
"""Close ``proc``'s pipe once the set has stayed empty for ``_SUPERVISOR_IDLE_EXIT_S``; the caller
holds ``_supervisor_lock``. Closing our write end is the same EOF our death sends; with an empty
set it exits, and its waiter thread reaps it. The next registration respawns and replays."""
global _idle_release
_cancel_idle_release_locked()
if _SUPERVISOR_IDLE_EXIT_S <= 0:
_close_idle_supervisor_locked(proc)
return
def release() -> None:
with _supervisor_lock:
_close_idle_supervisor_locked(proc)
try:
_idle_release = threading.Timer(_SUPERVISOR_IDLE_EXIT_S, release)
_idle_release.daemon = True
_idle_release.start()
except RuntimeError:
# No thread for the grace (thread limit, interpreter shutdown): release at once, as before the
# grace existed. Losing the grace must never fail the live child this call is about.
_idle_release = None
logger.debug("Could not start the supervisor idle timer; releasing it now", exc_info=True)
_close_idle_supervisor_locked(proc)
def _close_idle_supervisor_locked(proc) -> None:
global _supervisor
if _supervisor is not proc or _supervised_pgids:
return # replaced, or something registered meanwhile
with suppress(BrokenPipeError, ValueError, OSError):
proc.stdin.close()
_supervisor = None
def _cancel_idle_release_locked() -> None:
global _idle_release
if _idle_release is not None:
_idle_release.cancel()
_idle_release = None
def _start_supervisor_locked():
"""Start a supervisor and replay every live registration to it; the caller holds
``_supervisor_lock``. Returns it, or None when it cannot start or its pipe broke at once."""
global _supervisor
proc = _supervisor = _spawn_supervisor()
if proc is None:
return None
try:
proc.stdin.write("".join(f"register {pgid}\n" for pgid in _supervised_pgids))
proc.stdin.flush()
except (BrokenPipeError, ValueError, OSError):
_supervisor = None # its waiter (if it has one) reaps it; the next registration change retries
return None
return proc
def is_owner_death_supervisor(argv: Sequence[str]) -> bool:
"""True when a direct child's ``argv`` is this install's supervisor (``sh <script> <pgid>``).
Another family's child discovery (MCP's before/after child-pid delta) can catch it starting and
must never adopt, and later kill, it. Matched on the script path, not ``_supervisor.pid``: that
is assigned only after the spawn returns, so a scan racing the spawn would miss it."""
return list(argv[1:2]) == [str(_SUPERVISOR_SCRIPT)]
def _prune_dead_supervised_pgids() -> set:
"""Forget supervised groups with no members left; return what went. Caller holds
``_supervisor_lock``. Signal 0 is a pure existence probe (cannot terminate anything). It
narrows, but cannot close, the window where a dead group's pgid is recycled before we notice."""
stale = set()
for pgid in list(_supervised_pgids):
try:
os.killpg(pgid, 0) # windows-footgun: ok — POSIX-only, callers return early on nt
except ProcessLookupError:
stale.add(pgid)
except (PermissionError, OSError):
# Exists but not ours to signal, or the probe failed: keep it — dropping coverage on
# an ambiguous answer is the more expensive mistake.
pass
_supervised_pgids.difference_update(stale)
return stale
def own_process_groups(pgids) -> None:
"""Make these process groups die with this process, however it dies (see layer 4 above)."""
_update_supervised_pgids("register", pgids)
def release_process_groups(pgids) -> None:
"""Stop covering these groups: this process stopped them itself."""
_update_supervised_pgids("unregister", pgids)
def release_process_group_once_empty(pgid: int, *, proc=None, timeout: float = 1.0) -> bool:
"""Release ``pgid`` once no member is left; True if released. Polls ``proc`` (its leader) so a
dead-but-unreaped leader does not keep the group alive, and waits up to ``timeout`` for killed
members to be reaped by init. A member still there after that keeps the group covered, so it
still dies with this process. ``timeout=0`` probes once (async callers)."""
if os.name != "posix":
return False
deadline = time.monotonic() + timeout
while True:
if proc is not None:
try:
proc.poll()
except OSError:
pass
try:
os.killpg(pgid, 0) # windows-footgun: ok — POSIX-only, returned above; existence probe
except ProcessLookupError:
release_process_groups([pgid])
return True
except OSError:
pass # exists but not ours to probe: keep covering it
if time.monotonic() >= deadline:
return False
time.sleep(0.05)
def stop_owned_group_survivors(pgid: int, *, proc=None, grace: float = 3.0) -> bool:
"""Stop what is left of owned group ``pgid`` after its leader exited, then release it; True once
released. Members that outlived a crashed leader (a router's model children) keep the group, and
so its number, alive: TERM, wait ``grace``, KILL. Only groups this process registered are
signalled, so a number never registered here (or already released) is never touched."""
if os.name != "posix":
return False
with _supervisor_lock:
owned = pgid in _supervised_pgids
if owned:
for sig, wait in ((signal.SIGTERM, grace), (signal.SIGKILL, 1.0)): # windows-footgun: ok — POSIX-only, returned above
if release_process_group_once_empty(pgid, proc=proc, timeout=0):
return True
try:
os.killpg(pgid, sig) # windows-footgun: ok — POSIX-only, returned above
except ProcessLookupError:
pass
except OSError:
return False # not ours to signal: keep covering it
if release_process_group_once_empty(pgid, proc=proc, timeout=wait):
return True
return False
return release_process_group_once_empty(pgid, proc=proc, timeout=0)
def spawn_owned(argv, *, purpose: str, **popen_kwargs) -> subprocess.Popen:
"""Start a long-lived child that dies with this process: the one way to own a child.
POSIX: the child leads a new session and process group, and that group goes to the owner-death
supervisor, so the child and every descendant that stays in its group die the moment this
process dies by any cause (SIGKILL, OOM, crash). The child is recorded in the spawn ledger under
``purpose``. After stopping it yourself, ``release_process_group_once_empty(proc.pid, proc=proc)``.
Windows: a plain ``Popen`` with the caller's kwargs (the layer-3 job object covers ownership).
"""
if os.name != "posix":
return subprocess.Popen(argv, **popen_kwargs)
proc = subprocess.Popen(argv, start_new_session=True, **popen_kwargs)
own_process_groups([proc.pid])
register_child(proc.pid, purpose)
return proc
def _update_supervised_pgids(verb: str, pgids) -> None:
"""Send ``register``/``unregister`` for ``pgids`` to this process's supervisor. Failures are
swallowed: losing the safety net must never fail a live child."""
if os.name != "posix":
return
wanted = {int(pgid) for pgid in pgids}
if not wanted:
return
global _supervisor
with _supervisor_lock:
if verb == "register":
_supervised_pgids.update(wanted)
else:
_supervised_pgids.difference_update(wanted)
# A registration outlives its child only while some member survives (e.g. an orphaned
# grandchild teardown failed to kill, deliberately kept registered). Once that group is
# empty its pgid can be recycled by a stranger, so prune here too.
stale = _prune_dead_supervised_pgids()
proc = _supervisor
if proc is None or proc.poll() is not None:
if not _supervised_pgids:
# Nothing left to cover: nothing to tell and nothing to respawn for. Keyed on the
# SET, not the verb: after a broken-pipe write dropped the supervisor with groups
# still registered, an unregister must still rebuild coverage for the survivors.
return
# A fresh supervisor knows nothing: replay live coverage (already reflects this call's
# mutation and the prune, so pruned groups never reach the replacement).
_start_supervisor_locked()
return
if _supervised_pgids:
_cancel_idle_release_locked()
payload = "".join(f"{verb} {pgid}\n" for pgid in wanted)
payload += "".join(f"unregister {pgid}\n" for pgid in stale)
try:
proc.stdin.write(payload)
proc.stdin.flush()
except (BrokenPipeError, ValueError, OSError):
# It exited between poll() and write(). Drop it so the next call respawns and replays
# from ``_supervised_pgids`` (the set, not the pipe, is the record of what to reap). Its
# waiter thread reaps it.
_supervisor = None
return
if not _supervised_pgids:
# Nothing left to reap: release the supervisor after an idle grace rather than keep a
# process and a pipe resident for the life of a gateway. Not at once: a foreground
# command owns and releases its group every time, and a supervisor per command costs a
# /bin/sh spawn (~11 ms) under this lock.
_release_when_idle_locked(proc)