fix(v1): reap env pool worker subprocesses

verifiers/utils/process_utils.py

@@ -1,8 +1,10 @@
 """Process lifecycle utilities."""
 
+import contextlib
 import logging
 import os
 import signal
+import subprocess
 import threading
 from multiprocessing.connection import Connection
 from multiprocessing.process import BaseProcess
@@ -83,6 +85,127 @@ def terminate_processes(
         t.join()
 
 
+def _proc_session_processes(session: int) -> dict[int, bool] | None:
+    """Read session membership from procfs, or return None when unavailable."""
+    try:
+        entries = os.scandir("/proc")
+    except OSError:
+        return None
+
+    members: dict[int, bool] = {}
+    with entries:
+        for entry in entries:
+            if not entry.name.isdigit():
+                continue
+            try:
+                with open(f"/proc/{entry.name}/stat") as stat:
+                    fields = stat.read().rsplit(")", 1)[1].split()
+                if int(fields[3]) == session:
+                    members[int(entry.name)] = fields[0] != "Z"
+            except (IndexError, OSError, ValueError):
+                # Processes can exit between scandir and opening their stat file. A live
+                # member is seen on the next fixed-point scan below.
+                continue
+    return members
+
+
+def _session_processes(session: int) -> dict[int, bool] | None:
+    """Map current POSIX session PIDs to whether each process is non-zombie."""
+    if (members := _proc_session_processes(session)) is not None:
+        return members
+
+    try:
+        listing = subprocess.run(
+            ["ps", "-A", "-o", "pid=,sid=,stat="],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        ).stdout
+    except (OSError, subprocess.SubprocessError):
+        logger.warning("Could not read POSIX session %d", session)
+        return None
+
+    members = {}
+    for line in listing.splitlines():
+        try:
+            raw_pid, raw_sid, state = line.split()
+            pid, sid = int(raw_pid), int(raw_sid)
+        except ValueError:
+            continue
+        if sid == session:
+            members[pid] = not state.startswith("Z")
+    return members
+
+
+def kill_process_session(process: BaseProcess, timeout: float = 10.0) -> None:
+    """Kill a worker-owned POSIX session, including its separate process groups.
+
+    Pool workers are session leaders, and framework subprocess runtimes create their own
+    process groups inside that session.  Session membership therefore survives worker
+    exit and does not rely on a stale parent-PID snapshot.  Members are stopped until the
+    session is stable, preventing new forks between discovery and ``SIGKILL``.
+
+    Call this only after giving the process a chance to terminate gracefully.  It is a
+    POSIX helper; if the process did not become its own session leader, only its PID is
+    killed so an unrelated caller session is never targeted.
+    """
+    root = process.pid
+    assert root is not None
+    # The caller waits on the process sentinel without joining first.  An exited worker
+    # therefore remains a zombie and keeps its PID/SID reserved until this cleanup joins it.
+    members = _session_processes(root)
+    if members is None:
+        logger.error(
+            "Cannot discover worker session %d; killing only the worker process", root
+        )
+        with contextlib.suppress(Exception):
+            process.kill()
+        process.join(timeout=timeout)
+        return
+    if root not in members:
+        with contextlib.suppress(Exception):
+            process.kill()
+        process.join(timeout=timeout)
+        return
+
+    live = {pid for pid, is_live in members.items() if is_live}
+    discovery_failures = 0
+    try:
+        while live:
+            for pid in live:
+                with contextlib.suppress(ProcessLookupError, PermissionError):
+                    os.kill(pid, signal.SIGSTOP)
+            discovered = _session_processes(root)
+            if discovered is None:
+                discovery_failures += 1
+                if discovery_failures < 3:
+                    continue
+                logger.error("Could not verify stopped worker session %d", root)
+                break
+            discovery_failures = 0
+            current = {pid for pid, is_live in discovered.items() if is_live}
+            if current <= live:
+                live = current
+                break
+            live = current
+    finally:
+        # Refresh away exited/zombie PIDs before killing. The nested finally guarantees
+        # that even KeyboardInterrupt during discovery cannot strand a stopped worker.
+        try:
+            if (current := _session_processes(root)) is not None:
+                live = {pid for pid, is_live in current.items() if is_live}
+        finally:
+            for pid in live:
+                with contextlib.suppress(BaseException):
+                    if os.getsid(pid) == root:
+                        os.kill(pid, signal.SIGKILL)
+            with contextlib.suppress(BaseException):
+                process.kill()
+            with contextlib.suppress(BaseException):
+                process.join(timeout=timeout)
+
+
 def use_threading_tqdm_lock() -> None:
     """Pin tqdm to a threading lock so it never lazily creates an `mp.RLock` a killed worker would
     leak (a `resource_tracker` semaphore warning). No-op if tqdm isn't installed."""

verifiers/v1/runtimes/subprocess.py

@@ -54,14 +54,14 @@ async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult:
             cwd=self.workdir,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
-            start_new_session=True,  # own process group, so we can reap the whole tree
+            process_group=0,  # own group inside the pool worker's cleanup session
         )
         try:
             stdout, stderr = await proc.communicate()
         finally:
             # If the await didn't finish, the caller cancelled it (e.g. the rollout's
             # scoring_timeout / harness_timeout fired): communicate() leaves the process
-            # running, so SIGKILL its whole group (start_new_session => pgid == pid) — otherwise
+            # running, so SIGKILL its whole group (process_group=0 => pgid == pid) — otherwise
             # a hung child (a wedged uv/sympy verify) outlives the rollout and leaks CPU. A
             # no-op once it has exited on its own.
             if proc.returncode is None:
@@ -88,7 +88,7 @@ async def run_background(
                 cwd=self.workdir,
                 stdout=f,
                 stderr=asyncio.subprocess.STDOUT,
-                start_new_session=True,  # own process group, so cleanup() reaps the whole tree
+                process_group=0,  # own group inside the pool worker's cleanup session
             )
         self._background.append(
             proc
@@ -104,7 +104,7 @@ async def write(self, path: str, data: bytes) -> None:
 
     def cleanup(self) -> None:
         for proc in self._background:
-            # Kill the whole group (start_new_session => pgid == pid), not just proc.pid, so a
+            # Kill the whole group (process_group=0 => pgid == pid), not just proc.pid, so a
             # background server's children (sh -> uv -> python) are reaped too.
             with contextlib.suppress(ProcessLookupError, PermissionError):
                 os.killpg(os.getpgid(proc.pid), signal.SIGTERM)

verifiers/v1/serve/pool.py

@@ -28,8 +28,10 @@
 import os
 import signal
 import threading
+import time
 import uuid
 from collections.abc import Callable
+from multiprocessing.connection import wait
 
 import msgpack
 import zmq
@@ -38,6 +40,7 @@
 from verifiers.v1.env import EnvConfig
 from verifiers.v1.serve.server import EnvServer
 from verifiers.v1.serve.types import HealthResponse, RunGroupRequest
+from verifiers.utils.process_utils import kill_process_session
 
 logger = logging.getLogger(__name__)
 
@@ -117,7 +120,7 @@ def _spawn_worker(self) -> None:
         address = f"ipc://{self._worker_path(i)}"
         parent_conn, child_conn = self._mpctx.Pipe()
         proc = self._mpctx.Process(
-            target=serve_env,
+            target=_serve_pool_worker,
             kwargs=dict(
                 max_workers=1,
                 address=address,
@@ -246,12 +249,17 @@ def _shutdown(self) -> None:
                 w["pipe"].close()
             with contextlib.suppress(Exception):
                 w["process"].terminate()
+
+        # Wait on sentinels rather than join/is_alive: an exited worker remains a zombie,
+        # reserving the PID that identifies its cleanup session until that session is reaped.
+        pending = {w["process"].sentinel for w in self.workers}
+        deadline = time.monotonic() + 10
+        while pending and (remaining := deadline - time.monotonic()) > 0:
+            pending.difference_update(wait(pending, timeout=remaining))
+
         for w in self.workers:
             with contextlib.suppress(Exception):
-                w["process"].join(timeout=10)
-            if w["process"].is_alive():
-                with contextlib.suppress(Exception):
-                    w["process"].kill()
+                kill_process_session(w["process"])
             with contextlib.suppress(Exception):
                 w["dealer"].close()
             with contextlib.suppress(OSError):
@@ -261,6 +269,12 @@ def _shutdown(self) -> None:
         logger.info("EnvServerPool down")
 
 
+def _serve_pool_worker(**kwargs) -> None:
+    """Run a pool worker as the leader of the session that owns its subprocesses."""
+    os.setsid()
+    serve_env(**kwargs)
+
+
 def env_config_data(config) -> dict:
     """The picklable `EnvConfig` fields of a (possibly dynamically-narrowed, unpicklable)
     config object — ship this across a process boundary, then rebuild via