mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
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.
1146 lines
51 KiB
Python
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)
|