hermes: reap orphaned CLI lane processes

services/hermes/agent-deployment.yaml

@@ -865,7 +865,8 @@ spec:
               set -a
               . /opt/data/.env
               set +a
-              exec /opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_runner.py
+              exec /opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_supervisor.py -- \
+                /opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_runner.py
           env:
             - {name: HERMES_HOME, value: /opt/data}
             - {name: HERMES_AUTH_FILE, value: /runtime-access/hermes-auth.json}

services/hermes/kustomization.yaml

@@ -77,6 +77,7 @@ configMapGenerator:
       - cli_lane_retention.py=scripts/cli_lane_retention.py
       - cli_lane_routing.py=scripts/cli_lane_routing.py
       - cli_lane_runner.py=scripts/cli_lane_runner.py
+      - cli_lane_supervisor.py=scripts/cli_lane_supervisor.py
       - codex=scripts/codex
       - configure_agent_clients.py=scripts/configure_agent_clients.py
       - codex_broker.py=scripts/codex_broker.py

services/hermes/scripts/cli_lane_supervisor.py

@@ -0,0 +1,387 @@
+#!/usr/bin/env python3
+"""Own and reap the complete Linux process tree for one CLI lane runner.
+
+Provider CLIs create new sessions, and terminal commands may create more
+process groups below them.  When one of those intermediate parents exits,
+Linux reparents the surviving helper to the nearest child subreaper (or PID
+1).  The lane runner is threaded and has its own ``Popen.wait`` owners, so it
+must not install a competing SIGCHLD handler or call ``waitpid(-1)``.
+
+This single-threaded outer boundary is the only direct parent it creates.  It
+becomes a subreaper before forking the runner, then waits only for its direct
+or adopted children.  Consequently it cannot steal a provider child while
+the runner still owns that provider's exit status.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import json
+import os
+import signal
+import sys
+import time
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+PR_SET_CHILD_SUBREAPER = 36
+CONTROL_SIGNALS = (signal.SIGTERM, signal.SIGINT, signal.SIGHUP)
+EXIT_USAGE = 64
+
+
+@dataclass(frozen=True)
+class ProcessRecord:
+    """Non-sensitive identity fields read from one Linux procfs entry."""
+
+    parent: int
+    group: int
+    started: int
+    state: str
+
+
+@dataclass
+class ChildCounts:
+    """Bounded-cardinality process lifecycle counters."""
+
+    active: int = 0
+    adopted: int = 0
+    orphaned_total: int = 0
+    reaped_total: int = 0
+    reaped_orphans_total: int = 0
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _process_record(pid: int) -> ProcessRecord | None:
+    """Read parent, group, start identity, and state without command data."""
+    try:
+        raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
+        fields = raw[raw.rfind(")") + 2 :].split()
+        return ProcessRecord(
+            parent=int(fields[1]),
+            group=int(fields[2]),
+            started=int(fields[19]),
+            state=fields[0],
+        )
+    except (IndexError, OSError, ValueError):
+        return None
+
+
+def _process_records() -> dict[int, ProcessRecord]:
+    """Snapshot process identities from procfs without arguments or output."""
+    records: dict[int, ProcessRecord] = {}
+    try:
+        entries = Path("/proc").iterdir()
+    except OSError:  # pragma: no cover - Linux container contract failure
+        return records
+    for entry in entries:
+        if not entry.name.isdigit():
+            continue
+        pid = int(entry.name)
+        record = _process_record(pid)
+        if record is not None:
+            records[pid] = record
+    return records
+
+
+def _descendants(
+    root_pid: int,
+    records: dict[int, ProcessRecord],
+) -> dict[int, ProcessRecord]:
+    """Return the transitive process family below ``root_pid``."""
+    family = {root_pid}
+    changed = True
+    while changed:
+        changed = False
+        for pid, record in records.items():
+            if pid not in family and record.parent in family:
+                family.add(pid)
+                changed = True
+    return {pid: records[pid] for pid in family if pid != root_pid}
+
+
+def _same_process(pid: int, started: int) -> bool:
+    record = _process_record(pid)
+    return record is not None and record.started == started
+
+
+def _enable_subreaper() -> None:
+    """Make this process the adoption boundary before any runner is forked."""
+    libc = ctypes.CDLL(None, use_errno=True)
+    if libc.prctl(PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) != 0:
+        error = ctypes.get_errno()
+        raise OSError(error, "could not establish CLI child subreaper")
+
+
+def _decode_wait_status(status: int) -> int:
+    """Preserve an exited or signalled runner's shell-compatible status."""
+    if os.WIFEXITED(status):
+        return os.WEXITSTATUS(status)
+    if os.WIFSIGNALED(status):
+        return 128 + os.WTERMSIG(status)
+    return 1
+
+
+def _atomic_state(path: Path, document: dict[str, object]) -> None:
+    """Replace the fixed-shape process state document without growing a log."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    temporary.write_text(
+        json.dumps(document, sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+    temporary.chmod(0o600)
+    os.replace(temporary, path)
+
+
+class ChildSupervisor:
+    """Single-owner wait loop for a runner and its adopted descendants."""
+
+    def __init__(
+        self,
+        command: list[str],
+        state_path: Path,
+        *,
+        grace_seconds: float = 10.0,
+        poll_seconds: float = 0.1,
+    ) -> None:
+        self.command = command
+        self.state_path = state_path
+        self.grace_seconds = grace_seconds
+        self.poll_seconds = poll_seconds
+        self.pid = os.getpid()
+        self.root_pid = 0
+        self.root_status: int | None = None
+        self.requested_signal = 0
+        self.phase = "starting"
+        self.escalated = False
+        self.counts = ChildCounts()
+        self._active_orphans: dict[int, int] = {}
+        self._last_document = ""
+        self._last_write = 0.0
+        self._write_error_reported = False
+
+    def _spawn(self) -> int:
+        pid = os.fork()
+        if pid:
+            return pid
+        try:  # pragma: no cover - exercised through the exec integration test
+            for signum in CONTROL_SIGNALS:
+                signal.signal(signum, signal.SIG_DFL)
+            os.setsid()
+            # The absolute executable is pinned by the pod manifest and is
+            # validated in main; argv is passed directly without a shell.
+            os.execve(  # nosemgrep: dangerous-os-exec-tainted-env-args
+                self.command[0], self.command, os.environ.copy()
+            )
+        except OSError:  # pragma: no cover - exercised in the forked child
+            os.write(2, b"cli-child-supervisor: runner exec failed\n")
+        os._exit(127)  # pragma: no cover - coverage cannot flush after _exit
+
+    def _handle_control_signal(self, signum: int, _frame: object) -> None:
+        if not self.requested_signal:
+            self.requested_signal = signum
+
+    def _observe(self, records: dict[int, ProcessRecord]) -> dict[int, ProcessRecord]:
+        tree = _descendants(self.pid, records)
+        direct_orphans = {
+            pid: record
+            for pid, record in tree.items()
+            if record.parent == self.pid and pid != self.root_pid
+        }
+        for pid, record in direct_orphans.items():
+            if self._active_orphans.get(pid) != record.started:
+                self.counts.orphaned_total += 1
+            self._active_orphans[pid] = record.started
+        for pid in self._active_orphans.keys() - direct_orphans.keys():
+            del self._active_orphans[pid]
+        self.counts.active = len(tree)
+        self.counts.adopted = len(direct_orphans)
+        return tree
+
+    def _reap(self) -> None:
+        while True:
+            try:
+                pid, status = os.waitpid(-1, os.WNOHANG)
+            except ChildProcessError:
+                return
+            if pid <= 0:
+                return
+            self.counts.reaped_total += 1
+            if pid == self.root_pid:
+                self.root_status = status
+                continue
+            if pid not in self._active_orphans:
+                self.counts.orphaned_total += 1
+            self._active_orphans.pop(pid, None)
+            self.counts.reaped_orphans_total += 1
+
+    def _signal_tree(
+        self,
+        tree: dict[int, ProcessRecord],
+        signum: signal.Signals | int,
+    ) -> None:
+        own_group = os.getpgrp()
+        groups = {
+            record.group
+            for record in tree.values()
+            if record.group > 1 and record.group != own_group
+        }
+        for group in groups:
+            with suppress(PermissionError, ProcessLookupError):
+                os.killpg(group, signum)
+        for pid, record in tree.items():
+            if not _same_process(pid, record.started):
+                continue
+            with suppress(PermissionError, ProcessLookupError):
+                os.kill(pid, signum)
+
+    def _document(self) -> dict[str, object]:
+        return {
+            "active_children": self.counts.active,
+            "adopted_children": self.counts.adopted,
+            "escalated": self.escalated,
+            "orphaned_children_total": self.counts.orphaned_total,
+            "phase": self.phase,
+            "reaped_children_total": self.counts.reaped_total,
+            "reaped_orphans_total": self.counts.reaped_orphans_total,
+            "runner_exit_code": (
+                _decode_wait_status(self.root_status)
+                if self.root_status is not None
+                else None
+            ),
+            "termination_signal": self.requested_signal or None,
+            "updated_at": _utc_now(),
+        }
+
+    def _publish(self, *, force: bool = False) -> None:
+        now = time.monotonic()
+        document = self._document()
+        comparable = json.dumps({**document, "updated_at": ""}, sort_keys=True)
+        if not force and comparable == self._last_document:
+            return
+        if not force and now - self._last_write < 1.0:
+            return
+        try:
+            _atomic_state(self.state_path, document)
+        except OSError:
+            if not self._write_error_reported:
+                print(
+                    "cli-child-supervisor: process state unavailable",
+                    file=sys.stderr,
+                    flush=True,
+                )
+                self._write_error_reported = True
+            return
+        self._last_document = comparable
+        self._last_write = now
+
+    def _log_boundary(self) -> None:
+        print(
+            "cli-child-supervisor "
+            f"phase={self.phase} active={self.counts.active} "
+            f"adopted={self.counts.adopted} "
+            f"orphaned_total={self.counts.orphaned_total} "
+            f"reaped_total={self.counts.reaped_total}",
+            file=sys.stderr,
+            flush=True,
+        )
+
+    def run(self) -> int:
+        """Run until the runner exits and every adopted process is reaped."""
+        _enable_subreaper()
+        previous = {
+            signum: signal.signal(signum, self._handle_control_signal)
+            for signum in CONTROL_SIGNALS
+        }
+        shutdown_deadline: float | None = None
+        try:
+            self.root_pid = self._spawn()
+            self.phase = "running"
+            self._observe(_process_records())
+            self._publish(force=True)
+            self._log_boundary()
+            while True:
+                records = _process_records()
+                tree = self._observe(records)
+                self._reap()
+                records = _process_records()
+                tree = self._observe(records)
+                now = time.monotonic()
+
+                should_stop = bool(self.requested_signal) or self.root_status is not None
+                if should_stop and tree and shutdown_deadline is None:
+                    self.phase = "stopping"
+                    shutdown_deadline = now + self.grace_seconds
+                    stop_signal = self.requested_signal or signal.SIGTERM
+                    self._signal_tree(tree, stop_signal)
+                    self._publish(force=True)
+                    self._log_boundary()
+                elif tree and shutdown_deadline is not None and now >= shutdown_deadline:
+                    self.phase = "killing"
+                    self.escalated = True
+                    self._signal_tree(tree, signal.SIGKILL)
+                    shutdown_deadline = now + max(1.0, self.poll_seconds * 2)
+                    self._publish(force=True)
+
+                if self.root_status is not None and not tree:
+                    break
+                self._publish()
+                time.sleep(self.poll_seconds)
+
+            self._reap()
+            self._observe(_process_records())
+            self.phase = "exited"
+            self._publish(force=True)
+            self._log_boundary()
+            return _decode_wait_status(self.root_status)
+        finally:
+            for signum, handler in previous.items():
+                signal.signal(signum, handler)
+
+
+def _bounded_float(name: str, default: float, minimum: float, maximum: float) -> float:
+    try:
+        value = float(os.environ.get(name, str(default)))
+    except ValueError:
+        value = default
+    return max(minimum, min(value, maximum))
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Validate the fixed entrypoint shape and start the supervisor."""
+    arguments = list(sys.argv[1:] if argv is None else argv)
+    if (
+        len(arguments) < 2
+        or arguments[0] != "--"
+        or not Path(arguments[1]).is_absolute()
+    ):
+        print("usage: cli_lane_supervisor.py -- RUNNER [ARG ...]", file=sys.stderr)
+        return EXIT_USAGE
+    data_root = Path(os.environ.get("HERMES_HOME", "/opt/data"))
+    state_path = Path(
+        os.environ.get(
+            "HERMES_CLI_PROCESS_STATE_PATH",
+            str(data_root / "cli-lanes/process-supervisor.json"),
+        )
+    )
+    supervisor = ChildSupervisor(
+        arguments[1:],
+        state_path,
+        grace_seconds=_bounded_float(
+            "HERMES_CLI_PROCESS_GRACE_SECONDS", 10.0, 0.05, 30.0
+        ),
+        poll_seconds=_bounded_float(
+            "HERMES_CLI_PROCESS_POLL_SECONDS", 0.1, 0.01, 1.0
+        ),
+    )
+    return supervisor.run()
+
+
+if __name__ == "__main__":  # pragma: no cover - deployed script entrypoint
+    raise SystemExit(main())

testing/tests/test_hermes_cli_process_supervisor.py

@@ -0,0 +1,389 @@
+"""Linux process-owner coverage for the Hermes CLI lane supervisor."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import shutil
+import signal
+import sys
+from pathlib import Path
+
+import pytest
+import yaml
+
+
+ROOT = Path(__file__).parents[2]
+HERMES = ROOT / "services/hermes"
+SCRIPT = HERMES / "scripts/cli_lane_supervisor.py"
+SPEC = importlib.util.spec_from_file_location("cli_lane_supervisor", SCRIPT)
+assert SPEC and SPEC.loader
+supervisor = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = supervisor
+SPEC.loader.exec_module(supervisor)
+
+
+def _run(command: list[str], state_path: Path, *, grace: float = 0.1) -> int:
+    owner = supervisor.ChildSupervisor(
+        command,
+        state_path,
+        grace_seconds=grace,
+        poll_seconds=0.01,
+    )
+    return owner.run()
+
+
+def _state(path: Path) -> dict:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _python(code: str) -> list[str]:
+    return [sys.executable, "-c", code]
+
+
+def test_manifest_installs_supervisor_as_pid_one_boundary():
+    deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
+    containers = {
+        item["name"]: item
+        for item in deployment["spec"]["template"]["spec"]["containers"]
+    }
+    startup = containers["cli-lane-runner"]["args"][0]
+    kustomization = (HERMES / "kustomization.yaml").read_text()
+    source = SCRIPT.read_text()
+
+    assert "exec /opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_supervisor.py --" in startup
+    assert "/opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_runner.py" in startup
+    assert "cli_lane_supervisor.py=scripts/cli_lane_supervisor.py" in kustomization
+    assert "signal.signal(signal.SIGCHLD" not in source
+    assert len(source.splitlines()) < 500
+
+
+@pytest.mark.parametrize("exit_code", [0, 7])
+def test_normal_exit_preserves_status_stream_and_bounded_state(
+    tmp_path: Path,
+    capfd,
+    exit_code: int,
+):
+    state_path = tmp_path / "process-state.json"
+
+    result = _run(
+        _python(f"import sys; print('provider-stream-ok'); sys.exit({exit_code})"),
+        state_path,
+    )
+
+    captured = capfd.readouterr()
+    document = _state(state_path)
+    assert result == exit_code
+    assert "provider-stream-ok" in captured.out
+    assert document == {
+        "active_children": 0,
+        "adopted_children": 0,
+        "escalated": False,
+        "orphaned_children_total": 0,
+        "phase": "exited",
+        "reaped_children_total": 1,
+        "reaped_orphans_total": 0,
+        "runner_exit_code": exit_code,
+        "termination_signal": None,
+        "updated_at": document["updated_at"],
+    }
+    assert state_path.stat().st_mode & 0o777 == 0o600
+    assert "provider-stream-ok" not in state_path.read_text()
+
+
+@pytest.mark.parametrize(
+    ("binary", "arguments"),
+    [("git", ["git", "--version"]), ("ssh", ["ssh", "-V"])],
+)
+def test_provider_session_orphans_are_adopted_and_reaped_while_runner_lives(
+    tmp_path: Path,
+    binary: str,
+    arguments: list[str],
+):
+    executable = shutil.which(binary)
+    assert executable
+    state_path = tmp_path / f"{binary}.json"
+    provider_code = (
+        "import os,time\n"
+        "child=os.fork()\n"
+        "if child == 0:\n"
+        " time.sleep(0.05)\n"
+        f" os.execv({executable!r}, {arguments!r})\n"
+        "os._exit(0)\n"
+    )
+    runner_code = (
+        "import subprocess,sys,time\n"
+        f"provider=subprocess.Popen([sys.executable,'-c',{provider_code!r}],"
+        "start_new_session=True)\n"
+        "provider.wait()\n"
+        "time.sleep(0.25)\n"
+    )
+
+    assert _run(_python(runner_code), state_path) == 0
+
+    document = _state(state_path)
+    assert document["active_children"] == 0
+    assert document["orphaned_children_total"] >= 1
+    assert document["reaped_orphans_total"] >= 1
+    assert document["runner_exit_code"] == 0
+
+
+def test_concurrent_provider_orphans_have_one_non_competing_wait_owner(
+    tmp_path: Path,
+):
+    state_path = tmp_path / "concurrent.json"
+    provider_code = (
+        "import os,time\n"
+        "child=os.fork()\n"
+        "if child == 0:\n"
+        " time.sleep(0.05)\n"
+        " os.execl('/bin/true','true')\n"
+        "os._exit(0)\n"
+    )
+    runner_code = (
+        "import subprocess,sys,time\n"
+        f"code={provider_code!r}\n"
+        "providers=[subprocess.Popen([sys.executable,'-c',code],"
+        "start_new_session=True) for _ in range(4)]\n"
+        "statuses=[item.wait() for item in providers]\n"
+        "assert statuses == [0,0,0,0]\n"
+        "time.sleep(0.25)\n"
+    )
+
+    assert _run(_python(runner_code), state_path) == 0
+
+    document = _state(state_path)
+    assert document["orphaned_children_total"] >= 4
+    assert document["reaped_orphans_total"] >= 4
+    assert document["reaped_children_total"] >= 5
+
+
+def test_runner_exit_terminates_and_reaps_detached_helper_before_restart(
+    tmp_path: Path,
+):
+    state_path = tmp_path / "restart.json"
+    helper_pid_path = tmp_path / "helper.pid"
+    runner_code = (
+        "import os,signal,time\n"
+        "child=os.fork()\n"
+        "if child == 0:\n"
+        " os.setsid()\n"
+        " signal.signal(signal.SIGTERM,signal.SIG_IGN)\n"
+        " while True: time.sleep(1)\n"
+        f"open({str(helper_pid_path)!r},'w').write(str(child))\n"
+    )
+
+    assert _run(_python(runner_code), state_path, grace=0.05) == 0
+
+    helper_pid = int(helper_pid_path.read_text())
+    document = _state(state_path)
+    assert not Path(f"/proc/{helper_pid}").exists()
+    assert document["escalated"] is True
+    assert document["orphaned_children_total"] >= 1
+    assert document["reaped_orphans_total"] >= 1
+    assert document["runner_exit_code"] == 0
+
+
+def test_sigterm_cancellation_escalates_to_sigkill_and_reaps_nested_sessions(
+    tmp_path: Path,
+):
+    state_path = tmp_path / "cancel.json"
+    helper_pid_path = tmp_path / "cancel-helper.pid"
+    runner_code = (
+        "import os,signal,time\n"
+        "signal.signal(signal.SIGTERM,signal.SIG_IGN)\n"
+        "child=os.fork()\n"
+        "if child == 0:\n"
+        " os.setsid()\n"
+        " signal.signal(signal.SIGTERM,signal.SIG_IGN)\n"
+        " while True: time.sleep(1)\n"
+        f"open({str(helper_pid_path)!r},'w').write(str(child))\n"
+        "time.sleep(0.1)\n"
+        "os.kill(os.getppid(),signal.SIGTERM)\n"
+        "while True: time.sleep(1)\n"
+    )
+
+    assert _run(_python(runner_code), state_path, grace=0.05) == 137
+
+    helper_pid = int(helper_pid_path.read_text())
+    document = _state(state_path)
+    assert not Path(f"/proc/{helper_pid}").exists()
+    assert document["termination_signal"] == signal.SIGTERM
+    assert document["escalated"] is True
+    assert document["runner_exit_code"] == 137
+    assert document["active_children"] == 0
+
+
+def test_provider_cancellation_leaves_no_unowned_helper(tmp_path: Path):
+    state_path = tmp_path / "provider-cancel.json"
+    helper_ready = tmp_path / "provider-helper.ready"
+    provider_code = (
+        "import os,signal,time\n"
+        "child=os.fork()\n"
+        "if child == 0:\n"
+        " os.setsid()\n"
+        " signal.signal(signal.SIGTERM,signal.SIG_IGN)\n"
+        f" open({str(helper_ready)!r},'w').write('ready')\n"
+        " while True: time.sleep(1)\n"
+        "while True: time.sleep(1)\n"
+    )
+    runner_code = (
+        "import os,pathlib,signal,subprocess,sys,time\n"
+        f"provider=subprocess.Popen([sys.executable,'-c',{provider_code!r}],"
+        "start_new_session=True)\n"
+        f"ready=pathlib.Path({str(helper_ready)!r})\n"
+        "while not ready.exists(): time.sleep(0.01)\n"
+        "os.killpg(provider.pid,signal.SIGTERM)\n"
+        "provider.wait()\n"
+    )
+
+    assert _run(_python(runner_code), state_path, grace=0.05) == 0
+
+    document = _state(state_path)
+    assert document["active_children"] == 0
+    assert document["orphaned_children_total"] >= 1
+    assert document["reaped_orphans_total"] >= 1
+
+
+def test_observability_failure_does_not_mask_runner_status(capfd):
+    result = _run(_python("raise SystemExit(3)"), Path("/proc/not-writable/state"))
+
+    assert result == 3
+    assert capfd.readouterr().err.count("process state unavailable") == 1
+
+
+def test_exec_failure_is_generic_and_preserves_127(tmp_path: Path, capfd):
+    result = _run(["/definitely/missing/runner"], tmp_path / "exec.json")
+
+    assert result == 127
+    assert "runner exec failed" in capfd.readouterr().err
+    assert _state(tmp_path / "exec.json")["runner_exit_code"] == 127
+
+
+def test_helpers_cover_invalid_input_bounds_and_process_identity(
+    monkeypatch,
+    capfd,
+):
+    monkeypatch.setenv("SUPERVISOR_FLOAT", "invalid")
+    assert supervisor._bounded_float("SUPERVISOR_FLOAT", 2.0, 1.0, 3.0) == 2.0
+    monkeypatch.setenv("SUPERVISOR_FLOAT", "99")
+    assert supervisor._bounded_float("SUPERVISOR_FLOAT", 2.0, 1.0, 3.0) == 3.0
+    monkeypatch.setenv("SUPERVISOR_FLOAT", "-1")
+    assert supervisor._bounded_float("SUPERVISOR_FLOAT", 2.0, 1.0, 3.0) == 1.0
+    assert supervisor.main([]) == supervisor.EXIT_USAGE
+    assert supervisor.main(["--", "relative-runner"]) == supervisor.EXIT_USAGE
+    assert "usage:" in capfd.readouterr().err
+
+    current = supervisor._process_record(os.getpid())
+    assert current and current.parent > 0 and current.started > 0
+    assert supervisor._same_process(os.getpid(), current.started)
+    assert not supervisor._same_process(os.getpid(), current.started + 1)
+    records = {
+        10: supervisor.ProcessRecord(1, 10, 1, "S"),
+        11: supervisor.ProcessRecord(10, 11, 2, "S"),
+        12: supervisor.ProcessRecord(11, 12, 3, "Z"),
+        20: supervisor.ProcessRecord(1, 20, 4, "S"),
+    }
+    assert set(supervisor._descendants(10, records)) == {11, 12}
+    assert supervisor._decode_wait_status((signal.SIGSTOP << 8) | 0x7F) == 1
+
+
+def test_main_builds_bounded_supervisor_without_exposing_configuration(
+    tmp_path: Path,
+    monkeypatch,
+):
+    observed = {}
+
+    class FakeSupervisor:
+        def __init__(self, command, state_path, **kwargs):
+            observed.update(command=command, state_path=state_path, kwargs=kwargs)
+
+        @staticmethod
+        def run():
+            return 9
+
+    state_path = tmp_path / "configured.json"
+    monkeypatch.setattr(supervisor, "ChildSupervisor", FakeSupervisor)
+    monkeypatch.setenv("HERMES_CLI_PROCESS_STATE_PATH", str(state_path))
+    monkeypatch.setenv("HERMES_CLI_PROCESS_GRACE_SECONDS", "999")
+    monkeypatch.setenv("HERMES_CLI_PROCESS_POLL_SECONDS", "0")
+
+    assert supervisor.main(["--", "/runner", "argument"]) == 9
+    assert observed == {
+        "command": ["/runner", "argument"],
+        "state_path": state_path,
+        "kwargs": {"grace_seconds": 30.0, "poll_seconds": 0.01},
+    }
+
+
+def test_signal_forwarding_ignores_stale_and_disappeared_processes(monkeypatch):
+    owner = supervisor.ChildSupervisor(["runner"], Path("state"))
+    owner.requested_signal = signal.SIGTERM
+    owner._handle_control_signal(signal.SIGINT, None)
+    assert owner.requested_signal == signal.SIGTERM
+
+    groups = []
+    processes = []
+    tree = {
+        41: supervisor.ProcessRecord(1, 41, 1, "S"),
+        42: supervisor.ProcessRecord(1, 42, 2, "S"),
+    }
+    monkeypatch.setattr(supervisor.os, "getpgrp", lambda: 42)
+
+    def missing_group(group, _signal):
+        groups.append(group)
+        raise ProcessLookupError
+
+    def denied_process(pid, _signal):
+        processes.append(pid)
+        raise PermissionError
+
+    monkeypatch.setattr(supervisor.os, "killpg", missing_group)
+    monkeypatch.setattr(
+        supervisor,
+        "_same_process",
+        lambda pid, _started: pid == 42,
+    )
+    monkeypatch.setattr(supervisor.os, "kill", denied_process)
+
+    owner._signal_tree(tree, signal.SIGKILL)
+
+    assert groups == [41]
+    assert processes == [42]
+
+
+def test_orphan_identity_memory_is_bounded_by_current_adoptions(monkeypatch):
+    owner = supervisor.ChildSupervisor(["runner"], Path("state"))
+    owner.pid = 10
+    owner.root_pid = 11
+    root = supervisor.ProcessRecord(10, 11, 1, "S")
+    orphan = supervisor.ProcessRecord(10, 12, 2, "Z")
+
+    owner._observe({11: root, 12: orphan})
+    owner._observe({11: root, 12: orphan})
+    assert owner.counts.orphaned_total == 1
+    assert owner._active_orphans == {12: 2}
+
+    owner._observe({11: root})
+    assert owner._active_orphans == {}
+    waits = iter([(99, 0), (0, 0)])
+    monkeypatch.setattr(supervisor.os, "waitpid", lambda *_args: next(waits))
+    owner._reap()
+
+    assert owner.counts.orphaned_total == 2
+    assert owner.counts.reaped_orphans_total == 1
+    assert owner._active_orphans == {}
+
+
+def test_subreaper_setup_failure_is_explicit(monkeypatch):
+    class FailedPrctl:
+        @staticmethod
+        def prctl(*_args):
+            return -1
+
+    monkeypatch.setattr(supervisor.ctypes, "CDLL", lambda *_args, **_kwargs: FailedPrctl())
+    monkeypatch.setattr(supervisor.ctypes, "get_errno", lambda: 22)
+
+    with pytest.raises(OSError, match="subreaper"):
+        supervisor._enable_subreaper()