fix(#912): forward in-flight allowlist to drain-time cgroup sweep

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,

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.

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,

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) |

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]