diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 5ed758cf..c2c59528 100644 --- a/services/hermes/agent-deployment.yaml +++ b/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} diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 706a56b0..08f3e3ed 100644 --- a/services/hermes/kustomization.yaml +++ b/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 diff --git a/services/hermes/scripts/cli_lane_supervisor.py b/services/hermes/scripts/cli_lane_supervisor.py new file mode 100644 index 00000000..acd74cd9 --- /dev/null +++ b/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()) diff --git a/testing/tests/test_hermes_cli_process_supervisor.py b/testing/tests/test_hermes_cli_process_supervisor.py new file mode 100644 index 00000000..0b765ca8 --- /dev/null +++ b/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()