diff --git a/docker/base-image/agent_server/services/orphan_sweeper.py b/docker/base-image/agent_server/services/orphan_sweeper.py index 025114f52..efdf2f1af 100644 --- a/docker/base-image/agent_server/services/orphan_sweeper.py +++ b/docker/base-image/agent_server/services/orphan_sweeper.py @@ -100,37 +100,23 @@ def _active_execution_pids() -> Iterable[int]: than risking a false kill of the agent-server itself on a registry glitch. """ + # #912: delegated to `ProcessRegistry.active_execution_pids()` so the + # periodic sweep and the drain-time sweep in `subprocess_pgroup` share + # one canonical source. The registry method returns a list (with + # duplicate-tolerant pid/pgid entries — the allowlist resolver dedupes + # via descendant walk); convert to a set here for the sweeper's + # interface contract. try: from .process_registry import get_process_registry # lazy except Exception: # noqa: BLE001 return () try: - registry = get_process_registry() - running = registry.list_running() + return set(get_process_registry().active_execution_pids()) except Exception: # noqa: BLE001 logger.exception("[OrphanSweeper] failed to enumerate active executions") return () - pids: set[int] = set() - for entry in running: - if not isinstance(entry, dict): - continue - # ``pid`` exposed in list_running's shape (#817 follow-up); the - # allowlist resolver walks descendants via ppid so claude's - # tool subprocesses are covered automatically. - pid = entry.get("pid") - if isinstance(pid, int) and pid > 0: - pids.add(pid) - # ``pgid`` captured at register time — covers grandchildren that - # were spawned with ``setsid`` (escaping the ppid chain) but - # remain in the original pgid. - meta = entry.get("metadata") or {} - pgid = meta.get("pgid") - if isinstance(pgid, int) and pgid > 0: - pids.add(pgid) - return pids - async def run_orphan_sweep_loop( interval_seconds: float | None = None, diff --git a/docker/base-image/agent_server/services/process_registry.py b/docker/base-image/agent_server/services/process_registry.py index 23687d5a8..d2ef34ccc 100644 --- a/docker/base-image/agent_server/services/process_registry.py +++ b/docker/base-image/agent_server/services/process_registry.py @@ -149,29 +149,23 @@ def terminate(self, execution_id: str, graceful_timeout: int = 5) -> dict: # Issue #817 follow-up: cgroup-walk sweep for descendants # that escaped the pgid kill via setsid, FD detachment, or # env stripping. Best-effort — never fail termination on - # this. Preserve every OTHER active execution by passing - # their PIDs/pgids as extra_pids so we don't kill them - # while terminating this one. + # this. Issue #912: delegated to active_execution_pids() + # which is the single canonical source for this allowlist, + # also used by the periodic orphan sweeper and the drain- + # time sweep in subprocess_pgroup. Excluding ``execution_id`` + # so the sweep doesn't try to preserve the process we just + # killed. try: - preserve: list[int] = [] - with self._lock: - for other_id, other_entry in self._processes.items(): - if other_id == execution_id: - continue - other_proc = other_entry["process"] - if other_proc.poll() is not None: - continue - preserve.append(other_proc.pid) - other_pgid = (other_entry.get("metadata") or {}).get("pgid") - if isinstance(other_pgid, int) and other_pgid > 0: - preserve.append(other_pgid) + preserve = self.active_execution_pids( + exclude_execution_id=execution_id + ) killed = kill_cgroup_orphans(extra_pids=preserve) if killed: logger.info( f"[ProcessRegistry] Cgroup sweep killed {killed} " f"orphan(s) after terminating {execution_id} " - f"(preserved {len(preserve)} pid(s) for " - f"{len(self._processes) - 1} other execution(s))" + f"(preserved {len(preserve)} pid(s) for other " + f"in-flight execution(s))" ) except Exception: logger.exception( @@ -233,6 +227,42 @@ def list_running(self) -> list: }) return result + def active_execution_pids(self, exclude_execution_id: Optional[str] = None) -> List[int]: + """Snapshot of pids + captured pgids for currently-running executions. + + Returned for the orphan-sweep allowlist. Issue #912: any caller of + :func:`kill_cgroup_orphans` that runs while *other* executions are + still in flight must pass these so the sweep doesn't SIGKILL their + claude subprocesses. The list intentionally includes both the + ``pid`` (resolves descendants via ppid walk in + :mod:`orphan_allowlist`) and the captured ``pgid`` from metadata + (covers grandchildren spawned with ``setsid`` that escape the ppid + chain). Duplicates are fine — the allowlist resolver de-dupes. + + Args: + exclude_execution_id: When set, the entry with that id is + omitted from the snapshot. Used by ``terminate()`` so the + cgroup sweep doesn't try to preserve a process we just + killed. + + Returns: + A new list (snapshot under the registry lock) of ints. Empty + list when no other executions are running. + """ + result: List[int] = [] + with self._lock: + for exec_id, entry in self._processes.items(): + if exclude_execution_id is not None and exec_id == exclude_execution_id: + continue + process = entry["process"] + if process.poll() is not None: + continue + result.append(process.pid) + pgid = (entry.get("metadata") or {}).get("pgid") + if isinstance(pgid, int) and pgid > 0: + result.append(pgid) + return result + def cleanup_finished(self) -> int: """ Remove entries for finished processes. diff --git a/docker/base-image/agent_server/utils/subprocess_pgroup.py b/docker/base-image/agent_server/utils/subprocess_pgroup.py index 1b6338ac3..305292f6d 100644 --- a/docker/base-image/agent_server/utils/subprocess_pgroup.py +++ b/docker/base-image/agent_server/utils/subprocess_pgroup.py @@ -347,12 +347,22 @@ async def drain_reader_threads( finally: # Issue #817 follow-up: cgroup-walk runs on every exit path. # Best-effort — never fail the drain on sweep exceptions. + # Issue #912: forward all *other* in-flight executions' pids/pgids + # as the allowlist so the sweep doesn't SIGKILL a legitimate + # concurrent claude subprocess running another task in the same + # cgroup. The pre-#912 bare call was the bug — it left only the + # sweep's own pid + parents on the allowlist, so any concurrent + # task's claude got killed whenever a sibling task drained. + # The draining process itself has already exited at this point + # so it doesn't need to be in the allowlist. try: - killed = kill_cgroup_orphans() + extra_pids = _active_execution_pids_for_drain() + killed = kill_cgroup_orphans(extra_pids=extra_pids) if killed: logger.info( - "[Subprocess] Cgroup sweep killed %d orphan(s) after drain", - killed, + "[Subprocess] Cgroup sweep killed %d orphan(s) after drain " + "(preserved %d pid(s) for other in-flight execution(s))", + killed, len(extra_pids), ) except Exception: # noqa: BLE001 logger.exception( @@ -361,6 +371,28 @@ async def drain_reader_threads( ) +def _active_execution_pids_for_drain() -> list[int]: + """Lazy registry read for the #912 drain-time allowlist. + + Lives in ``utils`` and lazy-imports ``services.process_registry`` to + avoid pulling FastAPI into the import graph of this low-level helper. + Returns an empty list on any failure path — the allowlist is best- + effort; a registry hiccup at drain time must never crash the drain. + """ + try: + from ..services.process_registry import get_process_registry # lazy + except Exception: # noqa: BLE001 + return [] + try: + return get_process_registry().active_execution_pids() + except Exception: # noqa: BLE001 + logger.exception( + "[Subprocess] active_execution_pids() raised during drain sweep — " + "continuing with empty allowlist" + ) + return [] + + def signal_process_tree( process: subprocess.Popen, sig: int, diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md index 90383765d..365f3ab36 100644 --- a/docs/memory/feature-flows.md +++ b/docs/memory/feature-flows.md @@ -11,6 +11,7 @@ | Date | ID | Feature | Flow | |------|-----|---------|------| +| 2026-05-25 | #912 | fix(orphan-sweep): drain-time cgroup sweep now forwards an allowlist of in-flight execution pids/pgids so concurrent legitimate claude subprocesses don't get SIGKILLed when a sibling task drains in the same agent cgroup. Single canonical source via `ProcessRegistry.active_execution_pids(exclude_execution_id=…)` — used by the periodic orphan sweeper (#817), `ProcessRegistry.terminate()`, and the new `subprocess_pgroup._active_execution_pids_for_drain()` helper. Fixes the silent SIGKILL of multi-minute tasks visible as "exit code -9 / 0 tool calls / 0 turns" whenever any other task finished in the same container. 8 unit tests + in-container behavioural check. | [execution-termination.md](feature-flows/execution-termination.md), [parallel-headless-execution.md](feature-flows/parallel-headless-execution.md) | | 2026-05-18 | #887 | fix(read-only): guard moved to base image (`/opt/trinity/hooks/`, root-owned 0555); MultiEdit bypass fixed; fail-closed via `run_hook()`; lifecycle always syncs config on start (stale-volume fix); config file protected by `path_deny` + `bash_deny` in guardrails-baseline.json; 18 unit tests | [read-only-mode.md](feature-flows/read-only-mode.md) | | 2026-05-18 | #888 | write_user_memory MCP tool — per-user memory write with server-side email resolution, fixing PII cross-user memory leak | [write-user-memory.md](feature-flows/write-user-memory.md) | | 2026-05-17 | #35d4e78 | fix(credentials): map agent-server connect errors to 503 on `import_credentials` and `export_credentials` — `httpx.RequestError` (ConnectError/TimeoutException/ReadError) now surfaces 503 instead of 500 when the agent container is up but its FastAPI server isn't reachable yet. Mirrors the inject/agent-files pattern. | [credential-injection.md](feature-flows/credential-injection.md) | diff --git a/tests/unit/test_912_orphan_sweep_drain_allowlist.py b/tests/unit/test_912_orphan_sweep_drain_allowlist.py new file mode 100644 index 000000000..8aa6fa7cf --- /dev/null +++ b/tests/unit/test_912_orphan_sweep_drain_allowlist.py @@ -0,0 +1,189 @@ +""" +Tests for Issue #912 — drain-time orphan sweep must forward an +allowlist of in-flight execution pids/pgids so concurrent legitimate +claude subprocesses don't get SIGKILLed when a sibling task drains. + +Three surfaces: + + * ``ProcessRegistry.active_execution_pids()`` — single canonical + source of the allowlist, replacing the duplicated walks that lived + in ``orphan_sweeper`` and ``ProcessRegistry.terminate``. + * ``subprocess_pgroup._active_execution_pids_for_drain`` — the lazy + registry-read helper used in the drain-time finally block. + * ``subprocess_pgroup.drain_reader_threads`` — proof-of-wire-up that + ``kill_cgroup_orphans`` is called with ``extra_pids=`` populated. + +Tests bypass FastAPI and the real cgroup; ``kill_cgroup_orphans`` is +monkey-patched into a recording stub. +""" +from __future__ import annotations + +import asyncio +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +_REPO = Path(__file__).resolve().parents[2] +_BASE_IMAGE = _REPO / "docker" / "base-image" +if str(_BASE_IMAGE) not in sys.path: + sys.path.insert(0, str(_BASE_IMAGE)) + + +def _make_proc_entry(pid: int, *, poll_result=None, pgid: int | None = None) -> dict: + """Build a registry-shaped entry: process is a Mock that reports the + requested pid and poll() result; metadata carries the optional pgid.""" + process = MagicMock(spec=subprocess.Popen) + process.pid = pid + process.poll.return_value = poll_result + return { + "process": process, + "started_at": __import__("datetime").datetime.utcnow(), + "metadata": {"pgid": pgid} if pgid is not None else {}, + } + + +# --------------------------------------------------------------------------- +# ProcessRegistry.active_execution_pids +# --------------------------------------------------------------------------- + + +def test_active_execution_pids_includes_pid_and_pgid(): + from agent_server.services.process_registry import ProcessRegistry + + registry = ProcessRegistry() + registry._processes = { + "exec-a": _make_proc_entry(1001, pgid=1001), + "exec-b": _make_proc_entry(2002, pgid=3003), + } + + pids = registry.active_execution_pids() + # pid + pgid for each entry; duplicates fine (allowlist resolver dedupes). + assert sorted(pids) == [1001, 1001, 2002, 3003] + + +def test_active_execution_pids_excludes_self(): + from agent_server.services.process_registry import ProcessRegistry + + registry = ProcessRegistry() + registry._processes = { + "exec-a": _make_proc_entry(1001, pgid=1001), + "exec-b": _make_proc_entry(2002, pgid=2002), + } + + pids = registry.active_execution_pids(exclude_execution_id="exec-a") + assert 1001 not in pids + assert sorted(pids) == [2002, 2002] + + +def test_active_execution_pids_skips_finished_processes(): + """A process whose ``.poll()`` returned non-None is no longer + running; do not forward its pid as a preserve-target.""" + from agent_server.services.process_registry import ProcessRegistry + + registry = ProcessRegistry() + registry._processes = { + "exec-alive": _make_proc_entry(1001, poll_result=None, pgid=1001), + "exec-dead": _make_proc_entry(2002, poll_result=0, pgid=2002), + } + + pids = registry.active_execution_pids() + assert sorted(pids) == [1001, 1001] + + +def test_active_execution_pids_skips_pgid_when_missing_or_invalid(): + from agent_server.services.process_registry import ProcessRegistry + + registry = ProcessRegistry() + registry._processes = { + "no-pgid": _make_proc_entry(1001), # no metadata.pgid + "bad-pgid": _make_proc_entry(2002, pgid=0), # 0 = invalid + "neg-pgid": _make_proc_entry(3003, pgid=-1), # negative = invalid + "good-pgid": _make_proc_entry(4004, pgid=4004), + } + + pids = registry.active_execution_pids() + # All pids in; only the good pgid (4004) gets appended. + assert sorted(pids) == [1001, 2002, 3003, 4004, 4004] + + +def test_active_execution_pids_empty_registry(): + from agent_server.services.process_registry import ProcessRegistry + + assert ProcessRegistry().active_execution_pids() == [] + + +# --------------------------------------------------------------------------- +# subprocess_pgroup._active_execution_pids_for_drain +# --------------------------------------------------------------------------- + + +def test_drain_helper_returns_registry_pids(monkeypatch): + from agent_server.utils import subprocess_pgroup + + fake_registry = MagicMock() + fake_registry.active_execution_pids.return_value = [1001, 1001, 2002] + monkeypatch.setattr( + "agent_server.services.process_registry.get_process_registry", + lambda: fake_registry, + ) + assert subprocess_pgroup._active_execution_pids_for_drain() == [1001, 1001, 2002] + + +def test_drain_helper_swallows_registry_errors(monkeypatch): + """A registry hiccup at drain time must never crash the drain — the + finally block runs even when readers are wedged.""" + from agent_server.utils import subprocess_pgroup + + def _raises(): + raise RuntimeError("registry exploded") + + fake_registry = MagicMock() + fake_registry.active_execution_pids.side_effect = _raises + monkeypatch.setattr( + "agent_server.services.process_registry.get_process_registry", + lambda: fake_registry, + ) + assert subprocess_pgroup._active_execution_pids_for_drain() == [] + + +# --------------------------------------------------------------------------- +# drain_reader_threads forwards the allowlist +# --------------------------------------------------------------------------- + + +def test_drain_reader_threads_forwards_extra_pids(monkeypatch): + """The bug-fix proof: ``kill_cgroup_orphans`` is called with + ``extra_pids=`` from the registry. Pre-#912 it was called bare, + causing the false-kill of concurrent claude subprocesses.""" + from agent_server.utils import subprocess_pgroup + + # Record kill_cgroup_orphans calls. + calls = [] + + def _stub_kill(extra_pids=(), sweep_pid=None, dry_run=False): # noqa: ARG001 + calls.append({"extra_pids": list(extra_pids), "sweep_pid": sweep_pid}) + return 0 + + monkeypatch.setattr(subprocess_pgroup, "kill_cgroup_orphans", _stub_kill) + monkeypatch.setattr( + subprocess_pgroup, + "_active_execution_pids_for_drain", + lambda: [1001, 1001, 2002], + ) + + # A quickly-exiting subprocess + no reader threads to wedge: drain + # short-circuits and falls straight into the finally sweep. + proc = subprocess.Popen(["true"]) + proc.wait() + + asyncio.run( + subprocess_pgroup.drain_reader_threads( + proc, grace=1, post_kill_grace=1, pgid=None + ) + ) + + assert len(calls) == 1 + assert calls[0]["extra_pids"] == [1001, 1001, 2002]