From c2b1e4391c151c193421ee9af77dd3fd715a911d Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 12 Oct 2026 18:33:18 +0530 Subject: [PATCH] refactor(process): one release-once-empty helper; the kernel and slash-worker tests stop racing a loaded host LSP, the code kernel and the slash worker each had their own "release the group once nothing in it is left" (a killpg(pgid, 0) probe; the kernel's version also waited for reaps). They are now one helper, process_identity.release_process_group_once_empty(pgid, proc=, timeout=). It polls the leader so a zombie leader does not keep the group alive, waits up to `timeout` for killed members to be reaped by init, and keeps covering a group that still has a member. LSP's async close path probes once (timeout=0), as before. The slash worker used to probe exactly once right after SIGKILL. On a loaded macOS host (load 25+) init had not reaped the killed members yet, so a group that was about to be empty stayed registered (test_close_stops_the_whole_group_then_releases_it, flaky). It now waits up to 1 s, like the kernel did. The kernel survivor test raced too: under load the cell child's `trap '' TERM` could land after the teardown's SIGTERM, and the 5 s escalation could SIGKILL it. The child now signals once its trap is installed, and the test turns escalation off (the escalation is not what it pins). New tests/hermes_cli/test_process_identity_release_once_empty.py: an unreaped empty group is released, a live member keeps it covered, a late exit is waited for, and Windows never releases. Teeth: forcing the helper to always release, or to never release, turns 5 tests red in the 4 touched files. macOS: the 9 owned-child files, 78 tests, 3 runs in a row, 0 flaky. --- agent/lsp/client.py | 19 +----- hermes_cli/process_identity.py | 26 ++++++++ tests/agent/lsp/test_lsp_owned_server.py | 2 +- ...est_process_identity_release_once_empty.py | 60 +++++++++++++++++++ tests/tools/test_code_kernel_owned_group.py | 13 +++- tests/tui_gateway/test_slash_worker_owned.py | 14 +++-- tools/code_kernel.py | 13 +--- tui_gateway/server.py | 10 +--- 8 files changed, 114 insertions(+), 43 deletions(-) create mode 100644 tests/hermes_cli/test_process_identity_release_once_empty.py diff --git a/agent/lsp/client.py b/agent/lsp/client.py index 77f831ba58..93d584f6bb 100644 --- a/agent/lsp/client.py +++ b/agent/lsp/client.py @@ -20,7 +20,7 @@ from typing import Any, Awaitable, Callable, Dict, List, Optional, Set from urllib.parse import quote, unquote from hermes_cli._subprocess_compat import windows_hide_flags -from hermes_cli.process_identity import own_process_groups, register_child, release_process_groups +from hermes_cli.process_identity import own_process_groups, register_child, release_process_group_once_empty from agent.lsp.protocol import ( ERROR_CONTENT_MODIFIED, ERROR_METHOD_NOT_FOUND, LSPProtocolError, LSPRequestError, classify_message, @@ -449,7 +449,7 @@ class LSPClient: if proc.returncode is not None: if self._exit_code is None: self._exit_code = proc.returncode - _release_group_if_empty(proc.pid) + release_process_group_once_empty(proc.pid, timeout=0) return try: # ``shutdown`` has already given the protocol a grace period. Hard-kill @@ -467,7 +467,7 @@ class LSPClient: pass if self._exit_code is None and proc.returncode is not None: self._exit_code = proc.returncode - _release_group_if_empty(proc.pid) + release_process_group_once_empty(proc.pid, timeout=0) # ---- request / notification plumbing ---- @@ -813,19 +813,6 @@ class LSPClient: return _dedupe(push, pull) -def _release_group_if_empty(pgid: int) -> None: - """Hand the server's process group back from the owner-death supervisor once nothing in it is - left. A member that survived teardown stays covered, so it still dies with this process.""" - if os.name != "posix": - return - try: - os.killpg(pgid, 0) # windows-footgun: ok — POSIX-only, returned above - except ProcessLookupError: - release_process_groups([pgid]) - except OSError: - pass # exists but not ours to probe: keep covering it - - def _dedupe(*lists: list[dict[str, Any]]) -> list[dict[str, Any]]: seen: set[str] = set() out: list[dict[str, Any]] = [] diff --git a/hermes_cli/process_identity.py b/hermes_cli/process_identity.py index 1693c0506f..fa2d059cb2 100644 --- a/hermes_cli/process_identity.py +++ b/hermes_cli/process_identity.py @@ -851,6 +851,32 @@ def release_process_groups(pgids) -> None: _update_supervised_pgids("unregister", pgids) +def release_process_group_once_empty(pgid: int, *, proc=None, timeout: float = 1.0) -> bool: + """Release ``pgid`` once no member is left; True if released. Polls ``proc`` (its leader) so a + dead-but-unreaped leader does not keep the group alive, and waits up to ``timeout`` for killed + members to be reaped by init. A member still there after that keeps the group covered, so it + still dies with this process. ``timeout=0`` probes once (async callers).""" + if os.name != "posix": + return False + deadline = time.monotonic() + timeout + while True: + if proc is not None: + try: + proc.poll() + except OSError: + pass + try: + os.killpg(pgid, 0) # windows-footgun: ok — POSIX-only, returned above; existence probe + except ProcessLookupError: + release_process_groups([pgid]) + return True + except OSError: + pass # exists but not ours to probe: keep covering it + if time.monotonic() >= deadline: + return False + time.sleep(0.05) + + def 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. diff --git a/tests/agent/lsp/test_lsp_owned_server.py b/tests/agent/lsp/test_lsp_owned_server.py index 9158dedac0..3cce57fcd0 100644 --- a/tests/agent/lsp/test_lsp_owned_server.py +++ b/tests/agent/lsp/test_lsp_owned_server.py @@ -46,7 +46,7 @@ def _client(workspace: Path, script: str, **extra_env: str) -> LSPClient: def recorded(monkeypatch): calls: list = [] monkeypatch.setattr(lsp_client, "own_process_groups", lambda pgids: calls.append(("own", list(pgids)))) - monkeypatch.setattr(lsp_client, "release_process_groups", lambda pgids: calls.append(("release", list(pgids)))) + monkeypatch.setattr(process_identity, "release_process_groups", lambda pgids: calls.append(("release", list(pgids)))) monkeypatch.setattr(lsp_client, "register_child", lambda pid, purpose: calls.append(("ledger", pid, purpose))) return calls diff --git a/tests/hermes_cli/test_process_identity_release_once_empty.py b/tests/hermes_cli/test_process_identity_release_once_empty.py new file mode 100644 index 0000000000..6915c9e18b --- /dev/null +++ b/tests/hermes_cli/test_process_identity_release_once_empty.py @@ -0,0 +1,60 @@ +"""process_identity.release_process_group_once_empty: release only an empty group, waiting for reaps.""" + +import os +import signal +import subprocess +import sys +import threading + +import pytest + +from hermes_cli import process_identity + +pytestmark = pytest.mark.platforms("posix") + + +@pytest.fixture +def released(monkeypatch): + calls: list = [] + monkeypatch.setattr(process_identity, "release_process_groups", lambda pgids: calls.append(list(pgids))) + return calls + + +def _leader(script: str) -> subprocess.Popen: + return subprocess.Popen(["sh", "-c", script], stdin=subprocess.DEVNULL, + start_new_session=True) # detached-spawn: ok — test stand-in, killed in finally + + +def test_an_empty_group_is_released_even_with_an_unreaped_leader(released): + leader = _leader("exit 0") + # Not wait()ed: the zombie leader keeps killpg(pgid, 0) succeeding until proc.poll() reaps it. + assert process_identity.release_process_group_once_empty(leader.pid, proc=leader, timeout=5) + assert released == [[leader.pid]] + + +def test_a_group_with_a_live_member_stays_covered(released): + leader = _leader("trap '' TERM; exec sleep 60") + try: + assert not process_identity.release_process_group_once_empty(leader.pid, proc=leader, timeout=0.2) + assert released == [] + finally: + os.killpg(leader.pid, signal.SIGKILL) # windows-footgun: ok — posix-marked test + leader.wait(timeout=10) + + +def test_it_waits_for_a_member_that_exits_late(released): + leader = _leader("exec sleep 60") + threading.Timer(0.3, lambda: os.killpg(leader.pid, signal.SIGKILL)).start() # windows-footgun: ok — posix-marked test + try: + assert process_identity.release_process_group_once_empty(leader.pid, proc=leader, timeout=10) + assert released == [[leader.pid]] + finally: + if leader.poll() is None: + leader.kill() + leader.wait(timeout=10) + + +def test_windows_never_releases(monkeypatch, released): + monkeypatch.setattr(process_identity.os, "name", "nt") + assert not process_identity.release_process_group_once_empty(sys.maxsize, timeout=0) + assert released == [] diff --git a/tests/tools/test_code_kernel_owned_group.py b/tests/tools/test_code_kernel_owned_group.py index d77b59f10c..1d083df50a 100644 --- a/tests/tools/test_code_kernel_owned_group.py +++ b/tests/tools/test_code_kernel_owned_group.py @@ -138,10 +138,19 @@ def test_teardown_releases_the_kernels_group(monkeypatch, tmp_path): # The surviving cell child is orphaned once its kernel exits; killing it leaves the test subtree. @pytest.mark.live_system_guard_bypass def test_teardown_keeps_covering_a_cell_child_that_survived_it(monkeypatch, tmp_path): - # Ignores the teardown's SIGTERM; the kernel itself exits on it, so no SIGKILL escalation follows. + # Ignores the teardown's SIGTERM. Escalation is off so a kernel slow to exit under load cannot + # turn the TERM into a tree SIGKILL that takes the child too (the precondition, not the point). + from tools import code_execution_tool + real_kill = code_execution_tool._kill_process_group + monkeypatch.setattr(code_execution_tool, "_kill_process_group", + lambda proc, escalate=False: real_kill(proc, escalate=False)) + # The ready file is written only once the trap is in place: the receipt alone races a loaded + # host, where the teardown's SIGTERM can land before sh has installed it. + ready = tmp_path / "trapped" kernel, child, calls = _kernel_with_cell_child( - monkeypatch, tmp_path, ["sh", "-c", "trap '' TERM; exec sleep 60"]) + monkeypatch, tmp_path, ["sh", "-c", f"trap '' TERM; : > '{ready}'; exec sleep 60"]) try: + wait_until(ready.exists, what="the cell child to ignore SIGTERM") kernel.teardown() assert _alive(child), "precondition: the TERM-ignoring cell child survived teardown" assert calls == [("register", [kernel.proc.pid])], ( diff --git a/tests/tui_gateway/test_slash_worker_owned.py b/tests/tui_gateway/test_slash_worker_owned.py index b1e14d143f..0787766a96 100644 --- a/tests/tui_gateway/test_slash_worker_owned.py +++ b/tests/tui_gateway/test_slash_worker_owned.py @@ -24,6 +24,7 @@ import psutil import pytest from tests.waiting import wait_until +from hermes_cli import process_identity from tui_gateway import server, slash_worker _REPO = Path(__file__).resolve().parents[2] @@ -127,7 +128,7 @@ def _group_leader(*shell: str) -> subprocess.Popen: @pytest.mark.platforms("posix") def test_close_stops_the_whole_group_then_releases_it(monkeypatch): released = [] - monkeypatch.setattr(server, "release_process_groups", lambda pgids: released.append(list(pgids))) + monkeypatch.setattr(process_identity, "release_process_groups", lambda pgids: released.append(list(pgids))) leader = _group_leader("trap '' TERM; sleep 120 & wait") try: wait_until(lambda: len(_group(leader.pid)) >= 2, what="the stand-in's child in its group") @@ -146,7 +147,7 @@ def test_close_stops_the_whole_group_then_releases_it(monkeypatch): def test_a_worker_that_already_died_is_never_group_signalled(monkeypatch): """A dead leader no longer proves the pgid is ours: leave the group to the supervisor.""" released, signalled = [], [] - monkeypatch.setattr(server, "release_process_groups", lambda pgids: released.append(list(pgids))) + monkeypatch.setattr(process_identity, "release_process_groups", lambda pgids: released.append(list(pgids))) leader = _group_leader("sleep 120 & exit 0") try: leader.wait(timeout=10) @@ -157,7 +158,7 @@ def test_a_worker_that_already_died_is_never_group_signalled(monkeypatch): server._stop_worker_group(leader) - assert signalled == [0], "a dead worker's group number was signalled" + assert signalled and set(signalled) == {0}, "a dead worker's group number was signalled" assert released == [], "a group with a live member was released from the supervisor" assert all(p.is_running() for p in survivors) finally: @@ -228,7 +229,7 @@ def test_slash_worker_close_reaps_zombie_and_closes_fds(monkeypatch): raise subprocess.TimeoutExpired(cmd="x", timeout=timeout) monkeypatch.setattr(server.os, "killpg", lambda pgid, sig: group_signals.append((pgid, sig)), raising=False) - monkeypatch.setattr(server, "release_process_groups", lambda pgids: pytest.fail("a live group was released")) + monkeypatch.setattr(process_identity, "release_process_groups", lambda pgids: pytest.fail("a live group was released")) worker = object.__new__(server._SlashWorker) worker.proc = FakeProc() @@ -236,9 +237,10 @@ def test_slash_worker_close_reaps_zombie_and_closes_fds(monkeypatch): worker.close() # idempotent if os.name == "posix": - # TERM then KILL to the worker's group; the trailing 0 is the existence probe that keeps a + # TERM then KILL to the worker's group; the trailing 0s are the existence probes that keep a # group with survivors registered with the owner-death supervisor. - assert group_signals == [(424242, signal.SIGTERM), (424242, signal.SIGKILL), (424242, 0)] + assert group_signals[:2] == [(424242, signal.SIGTERM), (424242, signal.SIGKILL)] + assert group_signals[2:] and set(group_signals[2:]) == {(424242, 0)} else: assert calls["terminate"] == calls["kill"] == 1 assert calls["wait"] >= 2 # reaped after both terminate and kill diff --git a/tools/code_kernel.py b/tools/code_kernel.py index 4cead5af65..a36de8686c 100644 --- a/tools/code_kernel.py +++ b/tools/code_kernel.py @@ -307,7 +307,8 @@ class SessionKernel: from tools.code_execution_tool import _kill_process_group _kill_process_group(self.proc, escalate=True) if self.proc is not None and not _IS_WINDOWS: - _release_group_once_empty(self.proc) + from hermes_cli.process_identity import release_process_group_once_empty + release_process_group_once_empty(self.proc.pid, proc=self.proc) sock, self.server_sock = self.server_sock, None try: if sock is not None: @@ -321,16 +322,6 @@ class SessionKernel: shutil.rmtree(self.tmpdir, ignore_errors=True) -def _release_group_once_empty(proc: subprocess.Popen) -> None: - """Hand a torn-down kernel's group back from the owner-death supervisor once nothing in it is - left. SIGTERMed cell children exit (and get reaped by init) asynchronously, so wait briefly; a - cell child that ignored the SIGTERM stays covered, and the next own/release call prunes it.""" - from hermes_cli.process_identity import release_process_groups - from tools.environments.local import _wait_for_group_exit - if _wait_for_group_exit(proc, proc.pid, 1.0): - release_process_groups([proc.pid]) - - class KernelRegistry: """Key -> kernel map plus its lock (shared with the remote registry). Kernels are popped under the lock and torn down outside it — teardown may block on the child or the transport.""" diff --git a/tui_gateway/server.py b/tui_gateway/server.py index fc6e6cafcf..bf75ea0811 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -31,7 +31,7 @@ from hermes_constants import ( from hermes_cli import config_backend as _config_backend from hermes_cli.config_backend import Changes, ConfigBackendUnavailable, config_exists, supports_file_tooling from hermes_cli.env_loader import load_hermes_dotenv -from hermes_cli.process_identity import release_process_groups, spawn_owned +from hermes_cli.process_identity import release_process_group_once_empty, spawn_owned from tui_gateway.pet_sheet import _clone_pet_payload, _pet_row_frame_counts, _pet_sheet_revision, _pet_state_rows from utils import is_truthy_value from hermes_state_ids import new_session_id @@ -396,12 +396,8 @@ def _stop_worker_group(proc) -> None: os.killpg(pgid, sig) # windows-footgun: ok — POSIX-only, Windows returned above with contextlib.suppress(Exception): proc.wait(timeout=1) - try: - os.killpg(pgid, 0) # windows-footgun: ok — POSIX-only; existence probe, signals nothing - except ProcessLookupError: - release_process_groups([pgid]) - except OSError: - pass # ambiguous: stay covered by the owner-death supervisor + # Killed members are reaped by init asynchronously; one still alive after that stays covered. + release_process_group_once_empty(pgid, proc=proc) def _display_cfg() -> dict: