mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-23 16:36:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(gateway): reap the replaced gateway's orphaned children on POSIX

Browse source 
Builds on jbbottoms's #65178 takeover fix (cherry-picked as the previous
commit). Windows --replace already tree-kills via taskkill /T, but the
POSIX paths signalled only the recorded gateway PID — adapter
subprocesses that outlived their parent kept holding scoped token locks
and blocked the replacement gateway.

- gateway/status.py: _snapshot_gateway_children() captures the old
  gateway's descendants (psutil, recursive) while it is still alive;
  reap_gateway_children() SIGTERMs verified orphans after the main PID
  is confirmed dead, waits bounded, SIGKILLs survivors. Identity-aware
  (psutil is_running is PID+create-time), skips zombies and children
  whose ppid still equals the old gateway (parent actually alive), and
  never raises — best-effort with debug/info logging only.
- take_over_scoped_lock_holder() snapshots before terminating and reaps
  only on a confirmed successful handoff.
- gateway/run.py: start_gateway --replace snapshots before SIGTERM and
  reaps after the old PID is confirmed gone, mirroring taskkill /T.
- tests/gateway/test_replace_child_reap.py: reap/skip/never-raise unit
  coverage plus end-to-end --replace ordering (snapshot → terminate →
  reap) and the no---replace path never touching the old process.
This commit is contained in:
Teknium 2026-07-21 05:46:58 -07:00
parent 50fdf136e8
commit ab76cf836f
3 changed files with 530 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

26
gateway/run.py
View file

@ -22512,6 +22512,17 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
write_takeover_marker(existing_pid)
except Exception as e:
logger.debug("Could not write takeover marker: %s", e)
# Snapshot the old gateway's child processes BEFORE signalling it:
# once it exits, orphans are reparented and can no longer be found
# by a parent walk. On POSIX, adapter subprocesses that outlive
# the gateway keep holding scoped token locks and block the
# replacement (Windows terminate_pid(force=True) already
# tree-kills via taskkill /T). Best-effort — [] on any failure.
try:
from gateway.status import _snapshot_gateway_children
_old_gateway_children = _snapshot_gateway_children(existing_pid)
except Exception:
_old_gateway_children = []
try:
terminate_pid(existing_pid, force=False)
except ProcessLookupError:
@ -22575,6 +22586,21 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
except Exception:
pass
return False
# Old gateway confirmed dead — reap any orphaned child processes
# it left behind (POSIX; mirrors Windows taskkill /T tree-kill).
# Orphaned adapter subprocesses would otherwise keep holding
# scoped token locks against us. Best-effort, never raises.
try:
from gateway.status import reap_gateway_children
reap_gateway_children(
_old_gateway_children, parent_pid=existing_pid
)
except Exception:
logger.debug(
"Child reap for replaced gateway PID %d failed",
existing_pid,
exc_info=True,
)
remove_pid_file()
# remove_pid_file() is a no-op when the PID doesn't match.
# Force-unlink to cover the old-process-crashed case.

132
gateway/status.py
View file

@ -1642,6 +1642,108 @@ def _wait_for_scoped_lock_owner_exit(
return False, _scoped_lock_owner_state(owner_pid, owner_start_time) == "same"
def _snapshot_gateway_children(pid: int) -> list:
"""Best-effort snapshot of ``pid``'s live descendants (POSIX only).
Must be taken while the old gateway is still alive: once the parent
exits, its children are reparented (to init or a subreaper) and can no
longer be discovered by a parent walk. Returns ``[]`` on Windows —
``terminate_pid(force=True)`` there already tree-kills via
``taskkill /T`` — and on any error (missing psutil, process already
gone, access denied). Never raises.
"""
if _IS_WINDOWS:
return []
try:
import psutil # type: ignore
return psutil.Process(int(pid)).children(recursive=True)
except Exception:
logger.debug(
"Could not snapshot children of gateway PID %d", pid, exc_info=True
)
return []
def reap_gateway_children(children: list, *, parent_pid: int, timeout: float = 5.0) -> int:
"""Best-effort reap of a dead gateway's orphaned descendants (POSIX).
Mirrors the Windows ``taskkill /T`` tree-kill for the POSIX ``--replace``
paths: adapter subprocesses that survive their parent keep holding scoped
token locks and block the replacement gateway. Call only AFTER the main
gateway PID is confirmed dead, with a ``children`` snapshot taken via
:func:`_snapshot_gateway_children` while it was still alive.
Safety properties:
- ``psutil.Process.is_running()`` is identity-aware (PID + create time),
so a recycled child PID is never signalled.
- A child whose current ppid still equals ``parent_pid`` is skipped: that
means the parent is in fact alive (caller raced or was mocked) and the
child is not an orphan.
- SIGTERM first, bounded wait, SIGKILL only for survivors.
- Never raises; returns the number of children signalled.
"""
if _IS_WINDOWS or not children:
return 0
reaped = 0
try:
import psutil # type: ignore
live = []
for child in children:
try:
if not child.is_running():
continue
if child.status() == psutil.STATUS_ZOMBIE:
continue
if child.ppid() == parent_pid:
# Parent still alive — this is not an orphan; leave it.
logger.debug(
"Skipping child PID %d of old gateway %d: parent "
"still appears alive",
child.pid,
parent_pid,
)
continue
child.terminate()
live.append(child)
except psutil.NoSuchProcess:
continue
except Exception:
logger.debug(
"Could not terminate child PID %s of old gateway %d",
getattr(child, "pid", "?"),
parent_pid,
exc_info=True,
)
if not live:
return 0
gone, alive = psutil.wait_procs(live, timeout=max(0.0, timeout))
reaped = len(gone)
for child in alive:
try:
child.kill()
reaped += 1
except Exception:
logger.debug(
"Could not force-kill child PID %s of old gateway %d",
getattr(child, "pid", "?"),
parent_pid,
exc_info=True,
)
if reaped:
logger.info(
"Reaped %d orphaned child process(es) of replaced gateway PID %d.",
reaped,
parent_pid,
)
except Exception:
logger.debug(
"Child reap for replaced gateway PID %d failed", parent_pid, exc_info=True
)
return reaped
def take_over_scoped_lock_holder(
record: dict[str, Any],
*,
@ -1656,12 +1758,42 @@ def take_over_scoped_lock_holder(
deliberately stricter than the same-home PID-file replacement path: a
cross-home handoff must place a consumable marker in the target's home or a
service supervisor could revive the target and start a flap loop.
On POSIX, after the owner is confirmed dead, its previously snapshotted
child processes are reaped best-effort (see :func:`reap_gateway_children`)
so orphaned adapter subprocesses cannot keep holding token locks.
"""
owner = _validated_scoped_lock_gateway_owner(record)
if owner is None:
return None
owner_pid, owner_start_time, target_home = owner
# Snapshot descendants while the owner is still alive — after it exits
# they are reparented and undiscoverable (POSIX; [] on Windows where
# taskkill /T already tree-kills).
owner_children = _snapshot_gateway_children(owner_pid)
replaced = _terminate_scoped_lock_owner_once(
owner_pid,
owner_start_time,
target_home,
graceful_attempts=graceful_attempts,
force_attempts=force_attempts,
)
if replaced is not None:
reap_gateway_children(owner_children, parent_pid=owner_pid)
return replaced
def _terminate_scoped_lock_owner_once(
owner_pid: int,
owner_start_time: int,
target_home: Path,
*,
graceful_attempts: int = 20,
force_attempts: int = 20,
) -> Optional[int]:
"""Marker-write + bounded identity-aware termination of a verified owner."""
if not write_takeover_marker(
owner_pid,
target_home=target_home,

372
tests/gateway/test_replace_child_reap.py Normal file
View file

@ -0,0 +1,372 @@
"""Tests for --replace child-process reaping (POSIX taskkill /T parity).
On Windows, ``terminate_pid(force=True)`` tree-kills via ``taskkill /T``.
On POSIX, ``--replace`` historically signalled only the recorded gateway PID,
so adapter subprocesses that survived their parent kept holding scoped token
locks and blocked the replacement gateway. ``_snapshot_gateway_children`` /
``reap_gateway_children`` close that gap: the replacer snapshots the old
gateway's descendants while it is still alive, and reaps them (best-effort,
identity-aware) only after the main PID is confirmed dead.
Also asserts the takeover/reap machinery stays gated on explicit --replace.
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from gateway import status
from gateway.config import GatewayConfig
class _FakeChild:
"""Minimal psutil.Process stand-in for reap tests."""
def __init__(self, pid, *, running=True, ppid=1, zombie=False):
self.pid = pid
self._running = running
self._ppid = ppid
self._zombie = zombie
self.terminated = False
self.killed = False
def is_running(self):
return self._running
def status(self):
return "zombie" if self._zombie else "sleeping"
def ppid(self):
return self._ppid
def terminate(self):
self.terminated = True
def kill(self):
self.killed = True
def _fake_psutil(monkeypatch, *, wait_gone=None, wait_alive=None):
"""Install a stub psutil module for gateway.status's local imports."""
fake = MagicMock()
fake.STATUS_ZOMBIE = "zombie"
fake.NoSuchProcess = type("NoSuchProcess", (Exception,), {})
fake.wait_procs = MagicMock(
side_effect=lambda live, timeout: (
wait_gone if wait_gone is not None else list(live),
wait_alive if wait_alive is not None else [],
)
)
monkeypatch.setitem(sys.modules, "psutil", fake)
return fake
class TestReapGatewayChildren:
def test_reaps_orphaned_children_sigterm_then_wait(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", False)
fake = _fake_psutil(monkeypatch)
orphans = [_FakeChild(101, ppid=1), _FakeChild(102, ppid=1)]
reaped = status.reap_gateway_children(orphans, parent_pid=42)
assert reaped == 2
assert all(c.terminated for c in orphans)
assert not any(c.killed for c in orphans)
fake.wait_procs.assert_called_once()
def test_survivors_of_sigterm_get_sigkill(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", False)
stubborn = _FakeChild(103, ppid=1)
_fake_psutil(monkeypatch, wait_gone=[], wait_alive=[stubborn])
reaped = status.reap_gateway_children([stubborn], parent_pid=42)
assert stubborn.terminated
assert stubborn.killed
assert reaped == 1
def test_child_still_parented_to_live_parent_is_skipped(self, monkeypatch):
"""If a child's ppid still equals the old gateway PID, the parent is
alive and the child is not an orphan — never signal it."""
monkeypatch.setattr(status, "_IS_WINDOWS", False)
_fake_psutil(monkeypatch)
child = _FakeChild(104, ppid=42)
reaped = status.reap_gateway_children([child], parent_pid=42)
assert reaped == 0
assert not child.terminated
assert not child.killed
def test_dead_and_zombie_children_are_skipped(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", False)
_fake_psutil(monkeypatch)
dead = _FakeChild(105, running=False)
zombie = _FakeChild(106, zombie=True)
assert status.reap_gateway_children([dead, zombie], parent_pid=42) == 0
assert not dead.terminated and not zombie.terminated
def test_noop_on_windows_and_empty_snapshot(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", True)
child = _FakeChild(107, ppid=1)
assert status.reap_gateway_children([child], parent_pid=42) == 0
assert not child.terminated
monkeypatch.setattr(status, "_IS_WINDOWS", False)
assert status.reap_gateway_children([], parent_pid=42) == 0
def test_never_raises_when_psutil_explodes(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", False)
fake = _fake_psutil(monkeypatch)
fake.wait_procs.side_effect = RuntimeError("boom")
child = _FakeChild(108, ppid=1)
# Must swallow and return best-effort count, not raise.
assert status.reap_gateway_children([child], parent_pid=42) == 0
class TestSnapshotGatewayChildren:
def test_snapshot_walks_descendants_recursively(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", False)
fake = _fake_psutil(monkeypatch)
kids = [_FakeChild(201), _FakeChild(202)]
fake.Process.return_value.children.return_value = kids
assert status._snapshot_gateway_children(42) == kids
fake.Process.assert_called_once_with(42)
fake.Process.return_value.children.assert_called_once_with(recursive=True)
def test_snapshot_returns_empty_on_windows_or_error(self, monkeypatch):
monkeypatch.setattr(status, "_IS_WINDOWS", True)
assert status._snapshot_gateway_children(42) == []
monkeypatch.setattr(status, "_IS_WINDOWS", False)
fake = _fake_psutil(monkeypatch)
fake.Process.side_effect = RuntimeError("gone")
assert status._snapshot_gateway_children(42) == []
class TestScopedLockTakeoverReapsChildren:
"""take_over_scoped_lock_holder reaps the dead owner's orphans (POSIX)."""
@staticmethod
def _owner_record(target_home: Path, *, pid: int = 4242, start_time: int = 123):
target_home.mkdir(parents=True, exist_ok=True)
record = {
"pid": pid,
"kind": "hermes-gateway",
"argv": ["python", "-m", "hermes_cli.main", "gateway", "run"],
"start_time": start_time,
"hermes_home": str(target_home),
}
(target_home / "gateway.pid").write_text(json.dumps(record))
return record
def _verified_owner_env(self, tmp_path, monkeypatch, *, alive_polls):
replacer_home = tmp_path / "replacer"
target_home = tmp_path / "target"
replacer_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(replacer_home))
record = self._owner_record(target_home)
alive = iter(alive_polls)
monkeypatch.setattr(status, "_pid_exists", lambda _pid: next(alive))
monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123)
monkeypatch.setattr(
status,
"_read_process_cmdline",
lambda _pid: "python -m hermes_cli.main gateway run",
)
return record
def test_successful_takeover_snapshots_then_reaps(self, tmp_path, monkeypatch):
record = self._verified_owner_env(
tmp_path, monkeypatch, alive_polls=[True, True, False]
)
kids = [_FakeChild(301, ppid=1)]
events = []
monkeypatch.setattr(
status,
"_snapshot_gateway_children",
lambda pid: events.append(("snapshot", pid)) or kids,
)
monkeypatch.setattr(
status,
"reap_gateway_children",
lambda children, *, parent_pid, timeout=5.0: events.append(
("reap", parent_pid, children)
)
or len(children),
)
monkeypatch.setattr(
status,
"terminate_pid",
lambda pid, *, force=False: events.append(("terminate", pid, force)),
)
assert status.take_over_scoped_lock_holder(record, graceful_attempts=1) == 4242
# Snapshot taken while owner alive, BEFORE terminate; reap after exit.
assert events == [
("snapshot", 4242),
("terminate", 4242, False),
("reap", 4242, kids),
]
def test_failed_takeover_does_not_reap(self, tmp_path, monkeypatch):
# Owner never exits; safe_to_force stays False via unknown start time.
record = self._verified_owner_env(
tmp_path, monkeypatch, alive_polls=[True] * 50
)
monkeypatch.setattr(status, "_snapshot_gateway_children", lambda pid: [])
starts = iter([123, 123] + [None] * 50)
monkeypatch.setattr(
status, "_get_process_start_time", lambda _pid: next(starts)
)
reap = MagicMock()
monkeypatch.setattr(status, "reap_gateway_children", reap)
monkeypatch.setattr(status, "terminate_pid", lambda pid, *, force=False: None)
monkeypatch.setattr(status.time, "sleep", lambda _s: None)
assert (
status.take_over_scoped_lock_holder(
record, graceful_attempts=1, force_attempts=1
)
is None
)
reap.assert_not_called()
def test_unverified_holder_is_never_snapshotted_or_signalled(
self, tmp_path, monkeypatch
):
"""A non-gateway lock record fails identity validation: no snapshot,
no terminate, no reap — regardless of --replace intent upstream."""
record = {"pid": 4242, "kind": "something-else", "start_time": 123}
snapshot = MagicMock()
terminate = MagicMock()
reap = MagicMock()
monkeypatch.setattr(status, "_snapshot_gateway_children", snapshot)
monkeypatch.setattr(status, "terminate_pid", terminate)
monkeypatch.setattr(status, "reap_gateway_children", reap)
assert status.take_over_scoped_lock_holder(record) is None
snapshot.assert_not_called()
terminate.assert_not_called()
reap.assert_not_called()
@pytest.mark.asyncio
async def test_start_gateway_replace_reaps_old_gateway_children_posix(
monkeypatch, tmp_path
):
"""--replace snapshots the old gateway's children before SIGTERM and
reaps them after the main PID is confirmed dead (POSIX path)."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
events = []
kids = [_FakeChild(401, ppid=1)]
class _CleanExitRunner:
def __init__(self, config):
self.config = config
self.should_exit_cleanly = True
self.exit_reason = None
self.exit_code = None
self.adapters = {}
async def start(self):
assert self._platform_lock_takeover_on_start is True
return True
async def stop(self):
return None
_pid_state = {"alive": True}
monkeypatch.setattr(
"gateway.status.get_running_pid",
lambda: 42 if _pid_state["alive"] else None,
)
monkeypatch.setattr(
"gateway.status.remove_pid_file",
lambda: _pid_state.update(alive=False),
)
monkeypatch.setattr(
"gateway.status.release_all_scoped_locks", lambda **kwargs: 0
)
monkeypatch.setattr(
"gateway.status._snapshot_gateway_children",
lambda pid: events.append(("snapshot", pid)) or kids,
)
monkeypatch.setattr(
"gateway.status.reap_gateway_children",
lambda children, *, parent_pid, timeout=5.0: events.append(
("reap", parent_pid, children)
)
or len(children),
)
def _mock_terminate_pid(pid, force=False):
events.append(("terminate", pid, force))
_pid_state["alive"] = False
monkeypatch.setattr("gateway.status.terminate_pid", _mock_terminate_pid)
monkeypatch.setattr(
"gateway.status._pid_exists", lambda pid: _pid_state["alive"]
)
monkeypatch.setattr("gateway.run.os.getpid", lambda: 100)
monkeypatch.setattr("time.sleep", lambda _: None)
monkeypatch.setattr("tools.skills_sync.sync_skills", lambda quiet=True: None)
monkeypatch.setattr(
"hermes_logging.setup_logging", lambda hermes_home, mode: tmp_path
)
monkeypatch.setattr(
"hermes_logging._add_rotating_handler", lambda *args, **kwargs: None
)
monkeypatch.setattr("gateway.run.GatewayRunner", _CleanExitRunner)
from gateway.run import start_gateway
ok = await start_gateway(config=GatewayConfig(), replace=True, verbosity=None)
assert ok is True
# Snapshot precedes the SIGTERM; reap runs only after the PID is dead.
assert events == [
("snapshot", 42),
("terminate", 42, False),
("reap", 42, kids),
]
@pytest.mark.asyncio
async def test_start_gateway_without_replace_never_touches_old_gateway(
monkeypatch, tmp_path
):
"""Without --replace an existing gateway aborts startup: no takeover
authority is armed, no snapshot/terminate/reap ever runs."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
snapshot = MagicMock()
terminate = MagicMock()
reap = MagicMock()
monkeypatch.setattr("gateway.status.get_running_pid", lambda: 42)
monkeypatch.setattr("gateway.status._snapshot_gateway_children", snapshot)
monkeypatch.setattr("gateway.status.terminate_pid", terminate)
monkeypatch.setattr("gateway.status.reap_gateway_children", reap)
monkeypatch.setattr("gateway.run.os.getpid", lambda: 100)
class _RunnerShouldNotStart:
def __init__(self, config):
raise AssertionError("must not start while another gateway runs")
monkeypatch.setattr("gateway.run.GatewayRunner", _RunnerShouldNotStart)
from gateway.run import start_gateway
ok = await start_gateway(config=GatewayConfig(), replace=False, verbosity=None)
assert ok is False
snapshot.assert_not_called()
terminate.assert_not_called()
reap.assert_not_called()