From 7da67cfac1e9052db61db9ed6773b3c28e4ee0be Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:46:40 +0530 Subject: [PATCH] refactor(process): spawn_owned is the one way to own a child; own/release replace the verb string Names the owned-child API in process_identity layer 4: - spawn_owned(argv, purpose=..., **popen_kwargs): starts the child as a session/group leader, hands the group to the owner-death supervisor and records it in the spawn ledger. - own_process_groups(pgids) / release_process_groups(pgids): for groups something else spawned (MCP SDK servers, agent-browser's Chromium, harness daemons) and for children the owner stopped itself. The stringly update_supervised_pgids("register"/"unregister", ...) becomes the private protocol writer. llama-server's spawn_server(owned=True) now goes through spawn_owned instead of re-spelling it. Adds a contract test for spawn_owned (group leader, supervisor registration, ledger entry, release); it goes red when the registration is removed. --- hermes_cli/local_runtime/processes.py | 7 +- hermes_cli/local_runtime/supervisor.py | 4 +- hermes_cli/process_identity.py | 46 ++++++++++--- .../test_local_runtime_owned_router.py | 4 +- .../test_process_identity_supervisor.py | 68 ++++++++++++------- .../test_browser_use_cli_owned_daemon.py | 3 +- tools/browser_use_cli.py | 6 +- tools/mcp_tool_lifecycle.py | 4 +- tools/mcp_tool_transport.py | 6 +- 9 files changed, 98 insertions(+), 50 deletions(-) diff --git a/hermes_cli/local_runtime/processes.py b/hermes_cli/local_runtime/processes.py index b62fc3801a..f54cfff53f 100644 --- a/hermes_cli/local_runtime/processes.py +++ b/hermes_cli/local_runtime/processes.py @@ -122,12 +122,9 @@ def spawn_server(cmd, *, owned: bool = False, **kwargs) -> tuple[subprocess.Pope if sys.platform != "win32": if not owned: return subprocess.Popen(cmd, **kwargs), None - from hermes_cli.process_identity import register_child, update_supervised_pgids + from hermes_cli.process_identity import spawn_owned - proc = subprocess.Popen(cmd, start_new_session=True, **kwargs) - update_supervised_pgids("register", [proc.pid]) - register_child(proc.pid, "llama-server") - return proc, None + return spawn_owned(cmd, purpose="llama-server", **kwargs), None job = _WindowsJob() proc = None try: diff --git a/hermes_cli/local_runtime/supervisor.py b/hermes_cli/local_runtime/supervisor.py index 51344cc711..f82923d474 100644 --- a/hermes_cli/local_runtime/supervisor.py +++ b/hermes_cli/local_runtime/supervisor.py @@ -27,7 +27,7 @@ from pathlib import Path from hermes_cli.local_runtime.binaries import runtimes_root from hermes_cli.local_runtime.endpoint import child_query from hermes_cli.local_runtime.processes import server_child_env, spawn_server -from hermes_cli.process_identity import update_supervised_pgids +from hermes_cli.process_identity import release_process_groups logger = logging.getLogger(__name__) @@ -300,7 +300,7 @@ class LlamaServerSupervisor: if self.proc is not None: # Stopped by us: the owner-death supervisor has nothing left to cover. Only # after a terminate that returned, so a tree we failed to kill stays covered. - update_supervised_pgids("unregister", [self.proc.pid]) + release_process_groups([self.proc.pid]) # Retain state: deleting it could race a replacement publication. if self._log_handle: self._log_handle.close() diff --git a/hermes_cli/process_identity.py b/hermes_cli/process_identity.py index 0868a57ec7..1693c0506f 100644 --- a/hermes_cli/process_identity.py +++ b/hermes_cli/process_identity.py @@ -787,10 +787,14 @@ def attach_self_to_kill_on_close_job() -> bool: # 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. # -# Callers own process GROUPS: spawn the child as a group leader (``start_new_session=True``) and -# hand its pgid to ``update_supervised_pgids`` (stdio MCP servers, whose pgids the SDK spawns). A -# descendant that leaves the group on purpose must be registered by its own pgid where we learn it. -# POSIX-only: Windows children die through the layer-3 job object instead. +# ``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, a daemon's own Chromium) 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_groups``. ``scripts/check_detached_spawns.py`` keeps new detached spawns +# (start_new_session / setsid / setpgid) from bypassing this. POSIX-only: Windows children die +# through the layer-3 job object instead. _supervisor = None # Optional[subprocess.Popen]; tests rebind it _supervisor_lock = threading.Lock() @@ -837,10 +841,36 @@ def _prune_dead_supervised_pgids() -> set: return stale -def update_supervised_pgids(verb: str, pgids) -> None: - """Register or unregister process groups (``verb`` is ``"register"``/``"unregister"``) with - this process's supervisor. Failures are swallowed: losing the safety net must never fail a - live child.""" +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 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``. Call ``release_process_groups([proc.pid])`` after stopping it yourself. + 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} diff --git a/tests/hermes_cli/test_local_runtime_owned_router.py b/tests/hermes_cli/test_local_runtime_owned_router.py index ddaf2baa51..1ba46a072e 100644 --- a/tests/hermes_cli/test_local_runtime_owned_router.py +++ b/tests/hermes_cli/test_local_runtime_owned_router.py @@ -105,8 +105,8 @@ def test_a_graceful_stop_releases_the_routers_group(monkeypatch, tmp_path): released = [] monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path) - monkeypatch.setattr(supervisor, "update_supervised_pgids", - lambda verb, pgids: released.append((verb, list(pgids)))) + monkeypatch.setattr(supervisor, "release_process_groups", + lambda pgids: released.append(("unregister", list(pgids)))) sup = supervisor.LlamaServerSupervisor(tmp_path, tmp_path, port=59998) sup.proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"], stdin=subprocess.DEVNULL, start_new_session=True) diff --git a/tests/hermes_cli/test_process_identity_supervisor.py b/tests/hermes_cli/test_process_identity_supervisor.py index 106860f506..9ac01a0318 100644 --- a/tests/hermes_cli/test_process_identity_supervisor.py +++ b/tests/hermes_cli/test_process_identity_supervisor.py @@ -384,6 +384,26 @@ def all_groups_alive(monkeypatch): monkeypatch.setattr(process_identity.os, "killpg", lambda pgid, sig: None) +@pytest.mark.live_system_guard_bypass +def test_spawn_owned_starts_a_group_leader_registers_it_and_records_the_ledger(monkeypatch): + """The one way to own a child: own group, handed to the supervisor, recorded in the ledger.""" + fake = _FakeSupervisor() + ledger = [] + monkeypatch.setattr(process_identity, "_spawn_supervisor", lambda: fake) + monkeypatch.setattr(process_identity, "register_child", lambda pid, purpose: ledger.append((pid, purpose))) + proc = process_identity.spawn_owned(_VICTIM, purpose="unit-test", stdin=subprocess.DEVNULL) + try: + assert os.getpgid(proc.pid) == proc.pid, "an owned child must lead its own process group" + assert os.getpgid(proc.pid) != os.getpgid(0) + assert f"register {proc.pid}" in fake.lines(), "the owned child's group never reached the supervisor" + assert ledger == [(proc.pid, "unit-test")] + process_identity.release_process_groups([proc.pid]) + assert proc.pid not in process_identity._supervised_pgids + finally: + _kill(proc.pid) + proc.wait(timeout=10) + + def test_register_starts_the_supervisor_once_and_reuses_it(monkeypatch, all_groups_alive): spawned = [] @@ -394,8 +414,8 @@ def test_register_starts_the_supervisor_once_and_reuses_it(monkeypatch, all_grou monkeypatch.setattr(process_identity, "_spawn_supervisor", _spawn) - process_identity.update_supervised_pgids("register", [111]) - process_identity.update_supervised_pgids("register", [222]) + process_identity.own_process_groups([111]) + process_identity.own_process_groups([222]) assert len(spawned) == 1, "each register spawned its own supervisor" assert spawned[0].lines() == ["register 111", "register 222"] @@ -417,17 +437,17 @@ def test_supervisor_is_released_once_nothing_is_left_to_reap(monkeypatch, all_gr monkeypatch.setattr(process_identity, "_spawn_supervisor", _spawn) - process_identity.update_supervised_pgids("register", [111, 222]) - process_identity.update_supervised_pgids("unregister", [111]) + process_identity.own_process_groups([111, 222]) + process_identity.release_process_groups([111]) assert not spawned[0].closed, "released the supervisor while a group was still registered" - process_identity.update_supervised_pgids("unregister", [222]) + process_identity.release_process_groups([222]) assert spawned[0].closed, "supervisor kept resident with nothing left to reap" assert getattr(spawned[0], "waited", False), "released supervisor was never wait()ed -> zombie until the next Popen" assert spawned[0].lines()[-1] == "unregister 222", "release happened before the last unregister was sent" assert process_identity._supervisor is None - process_identity.update_supervised_pgids("register", [333]) + process_identity.own_process_groups([333]) assert len(spawned) == 2 and spawned[1].lines() == ["register 333"] @@ -435,10 +455,10 @@ def test_supervisor_survives_the_real_eof_release(): """End to end: closing the control pipe with nothing registered exits cleanly.""" child = subprocess.Popen(_VICTIM, start_new_session=True) try: - process_identity.update_supervised_pgids("register", [os.getpgid(child.pid)]) + process_identity.own_process_groups([os.getpgid(child.pid)]) proc = process_identity._supervisor assert proc is not None and proc.poll() is None - process_identity.update_supervised_pgids("unregister", [os.getpgid(child.pid)]) + process_identity.release_process_groups([os.getpgid(child.pid)]) assert process_identity._supervisor is None assert proc.wait(timeout=10) == 0, "supervisor did not exit on the release EOF" assert child.poll() is None, "release reaped a group that had been unregistered" @@ -455,7 +475,7 @@ def test_unregister_alone_does_not_start_a_supervisor(monkeypatch): lambda: spawned.append(1) or _FakeSupervisor(), ) - process_identity.update_supervised_pgids("unregister", [111]) + process_identity.release_process_groups([111]) assert spawned == [] @@ -466,8 +486,8 @@ def test_a_dead_supervisor_is_replaced_and_live_coverage_replayed(monkeypatch, a queue = [dead, replacement] monkeypatch.setattr(process_identity, "_spawn_supervisor", lambda: queue.pop(0)) - process_identity.update_supervised_pgids("register", [111]) - process_identity.update_supervised_pgids("register", [222]) + process_identity.own_process_groups([111]) + process_identity.own_process_groups([222]) # Losing the supervisor must not silently drop the child registered with # it -- the replacement has to be told about 111 as well as 222. @@ -480,9 +500,9 @@ def test_replay_does_not_resurrect_an_unregistered_group(monkeypatch, all_groups queue = [dead, replacement] monkeypatch.setattr(process_identity, "_spawn_supervisor", lambda: queue.pop(0)) - process_identity.update_supervised_pgids("register", [111]) - process_identity.update_supervised_pgids("register", [222]) - process_identity.update_supervised_pgids("unregister", [111]) + process_identity.own_process_groups([111]) + process_identity.own_process_groups([222]) + process_identity.release_process_groups([111]) assert process_identity._supervised_pgids == {222} # 111 was legitimately replayed to the replacement (it was live when the @@ -491,7 +511,7 @@ def test_replay_does_not_resurrect_an_unregistered_group(monkeypatch, all_groups lines = replacement.lines() assert lines.index("unregister 111") > lines.index("register 111") assert "register 111" not in lines[lines.index("unregister 111") :] - process_identity.update_supervised_pgids("register", [333]) # any later replay/append + process_identity.own_process_groups([333]) # any later replay/append assert "register 111" not in replacement.lines()[len(lines) :] @@ -511,7 +531,7 @@ def test_a_broken_pipe_never_propagates_into_a_live_child(monkeypatch, all_group monkeypatch.setattr(process_identity, "_spawn_supervisor", _BrokenPipe) - process_identity.update_supervised_pgids("register", [111]) # must not raise + process_identity.own_process_groups([111]) # must not raise # Dropped, so the next registration respawns instead of writing into a # pipe that is known to be dead. @@ -534,7 +554,7 @@ def test_unregister_after_a_broken_pipe_rebuilds_coverage_for_survivors(monkeypa return fake monkeypatch.setattr(process_identity, "_spawn_supervisor", _spawn) - process_identity.update_supervised_pgids("register", [111, 222]) + process_identity.own_process_groups([111, 222]) class _DeadStdin: def write(self, _payload): @@ -544,11 +564,11 @@ def test_unregister_after_a_broken_pipe_rebuilds_coverage_for_survivors(monkeypa pass spawned[0].stdin = _DeadStdin() - process_identity.update_supervised_pgids("register", [333]) # the write fails; supervisor dropped + process_identity.own_process_groups([333]) # the write fails; supervisor dropped assert process_identity._supervisor is None assert process_identity._supervised_pgids == {111, 222, 333} - process_identity.update_supervised_pgids("unregister", [222]) + process_identity.release_process_groups([222]) assert len(spawned) == 2, "unregister after a lost supervisor did not respawn one" assert sorted(spawned[1].lines()) == ["register 111", "register 333"], ( @@ -560,7 +580,7 @@ def test_unregister_after_a_broken_pipe_rebuilds_coverage_for_survivors(monkeypa def test_a_supervisor_that_cannot_start_is_not_fatal(monkeypatch, all_groups_alive): monkeypatch.setattr(process_identity, "_spawn_supervisor", lambda: None) - process_identity.update_supervised_pgids("register", [111]) # must not raise + process_identity.own_process_groups([111]) # must not raise assert process_identity._supervisor is None @@ -697,7 +717,7 @@ def test_scoped_teardown_of_one_owner_keeps_the_other_owner_supervised(monkeypat _mcp_lifecycle._stdio_pids[b.pid] = "profile-b" _mcp_lifecycle._stdio_pgids[a.pid] = pg_a _mcp_lifecycle._stdio_pgids[b.pid] = pg_b - process_identity.update_supervised_pgids("register", [pg_a, pg_b]) + process_identity.own_process_groups([pg_a, pg_b]) _mcp_lifecycle._kill_orphaned_mcp_children(include_active=True, server_name="profile-a") a.wait(timeout=10) @@ -737,7 +757,7 @@ def test_a_group_with_nothing_left_alive_is_forgotten_and_unregistered(monkeypat survivor = subprocess.Popen(_VICTIM, start_new_session=True) survivor_pgid = os.getpgid(survivor.pid) try: - process_identity.update_supervised_pgids("register", [doomed_pgid, survivor_pgid]) + process_identity.own_process_groups([doomed_pgid, survivor_pgid]) assert process_identity._supervised_pgids == {doomed_pgid, survivor_pgid} # Reap it fully so the group is genuinely empty, not a zombie. @@ -745,7 +765,7 @@ def test_a_group_with_nothing_left_alive_is_forgotten_and_unregistered(monkeypat doomed.wait(timeout=10) # Any later registration change is when we notice. - process_identity.update_supervised_pgids("register", [survivor_pgid]) + process_identity.own_process_groups([survivor_pgid]) assert doomed_pgid not in process_identity._supervised_pgids, ( "a group with no members left stayed registered, so a recycled " @@ -789,6 +809,6 @@ def test_no_pgids_is_a_no_op(monkeypatch): lambda: spawned.append(1) or _FakeSupervisor(), ) - process_identity.update_supervised_pgids("register", []) + process_identity.own_process_groups([]) assert spawned == [] diff --git a/tests/tools/test_browser_use_cli_owned_daemon.py b/tests/tools/test_browser_use_cli_owned_daemon.py index 4bb1516296..f726f34347 100644 --- a/tests/tools/test_browser_use_cli_owned_daemon.py +++ b/tests/tools/test_browser_use_cli_owned_daemon.py @@ -52,7 +52,8 @@ def fake_daemon(tmp_path): @pytest.fixture def registered(monkeypatch): calls = [] - monkeypatch.setattr(browser_use_cli, "update_supervised_pgids", lambda verb, pgids: calls.append((verb, list(pgids)))) + for verb, name in (("register", "own_process_groups"), ("unregister", "release_process_groups")): + monkeypatch.setattr(browser_use_cli, name, lambda pgids, verb=verb: calls.append((verb, list(pgids)))) monkeypatch.setattr(browser_use_cli, "register_child", lambda pid, purpose: None) monkeypatch.setattr(browser_use_cli, "_owned_daemon_groups", {}) return calls diff --git a/tools/browser_use_cli.py b/tools/browser_use_cli.py index 9774096db5..b280e9464e 100644 --- a/tools/browser_use_cli.py +++ b/tools/browser_use_cli.py @@ -23,7 +23,7 @@ import time from pathlib import Path from typing import Any, Callable, Dict, List, NamedTuple, Optional -from hermes_cli.process_identity import register_child, update_supervised_pgids +from hermes_cli.process_identity import own_process_groups, register_child, release_process_groups from hermes_constants import get_hermes_home, hermes_home_key from tools.browser_harness_dirs import (daemon_env, daemon_identity, fit_default_socket_dir, namespace_harness_dirs) @@ -642,7 +642,7 @@ def _reload_daemons(daemons) -> None: # each ``daemon_identity`` daemon, with i with _owned_daemon_groups_lock: pgid = _owned_daemon_groups.pop(daemon, None) if pgid is not None: - update_supervised_pgids("unregister", [pgid]) + release_process_groups([pgid]) # Daemon identity (``daemon_identity``: its dirs + BU_NAME) -> process group of the harness daemon this @@ -693,7 +693,7 @@ def _own_harness_daemon(env: dict) -> None: if _owned_daemon_groups.get(daemon) == pid: return _owned_daemon_groups[daemon] = pid - update_supervised_pgids("register", [pid]) + own_process_groups([pid]) register_child(pid, "browser-harness-daemon") diff --git a/tools/mcp_tool_lifecycle.py b/tools/mcp_tool_lifecycle.py index 4525a4fc43..0d406110cf 100644 --- a/tools/mcp_tool_lifecycle.py +++ b/tools/mcp_tool_lifecycle.py @@ -6,7 +6,7 @@ import asyncio import os import time from typing import Dict, Optional -from hermes_cli.process_identity import update_supervised_pgids +from hermes_cli.process_identity import release_process_groups from tools.mcp_tool_common import _core from tools import mcp_tool_loop as _loop @@ -386,7 +386,7 @@ def _kill_orphaned_mcp_children(include_active: bool = False, server_name: Optio logger.warning("Force-killed MCP process %d (%s) after SIGTERM timeout", pid, owner) # These groups are reaped. Release them last, so a crash partway through the SIGTERM/SIGKILL # dance still leaves the supervisor holding them. - update_supervised_pgids("unregister", pgids.values()) + release_process_groups(pgids.values()) def _stop_mcp_loop_if_idle() -> bool: diff --git a/tools/mcp_tool_transport.py b/tools/mcp_tool_transport.py index a65da16963..8aa7249f2b 100644 --- a/tools/mcp_tool_transport.py +++ b/tools/mcp_tool_transport.py @@ -14,7 +14,7 @@ from agent.proxy_bypass import is_loopback_host, should_bypass_proxy from agent import runtime_cwd as _runtime_cwd from tools.mcp_tool_errors import _MCP_HTTP_MAX_BODY_BYTES, NonMcpEndpointError, _apply_identity_header, _describe_http_failure, _handshake_answered_with_unsupported_version, _handshake_rejected_as_modern, _is_streamable_http_rejection, _make_http_rejection_recorder, _make_mcp_body_cap_transport, _make_redirect_header_stripper, _resolve_client_cert, _unwrap_exception_group from tools.mcp_tool_lifecycle import _filter_mcp_children, _leader_start_time, _orphan_stdio_pid_servers, _orphan_stdio_pids, _stdio_pgids, _stdio_pids, _stdio_starttimes -from hermes_cli.process_identity import register_child, update_supervised_pgids +from hermes_cli.process_identity import own_process_groups, register_child, release_process_groups from tools.mcp_tool_common import _core from tools.mcp_tool_node_abi import node_abi_error from tools import mcp_tool_config as _config @@ -338,7 +338,7 @@ class MCPServerTransportMixin: # process (kill -9, crash, force-quit) can't leave this server — or its descendants, e.g. # mcp-remote's spawned `node` — running forever. The graceful paths (shutdown, # _kill_orphaned_mcp_children) still reap as before; this only covers when they never run. - update_supervised_pgids("register", new_pgids.values()) + own_process_groups(new_pgids.values()) def _release_spawned_children(self, new_pids: set[int]) -> None: """Drop the ledger entries; a child (or its pgroup) still alive means SDK teardown failed @@ -360,7 +360,7 @@ class MCPServerTransportMixin: if dropped is not None: released_pgids.append(dropped) _stdio_starttimes.pop(pid, None) - update_supervised_pgids("unregister", released_pgids) + release_process_groups(released_pgids) async def _run_stdio(self, config: dict): """Run the server using stdio transport."""