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.
This commit is contained in:
kshitijk4poor authored and kshitij committed 2026-10-29 16:58:09 +05:30
1 parent bcec3f8a2f
commit 7da67cfac1
9 files changed
+98 -50

No files matched your search

+2 -5
View File
@@ -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:
+2 -2
View File
@@ -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()
+38 -8
View File
@@ -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}
@@ -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)
@@ -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 == []
@@ -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
+3 -3
View File
@@ -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")
+2 -2
View File
@@ -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:
+3 -3
View File
@@ -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."""