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,65 @@ def terminate_processes(
         t.join()
 
 
+def _session_processes(session: int) -> set[int]:
+    """Return the current members of a POSIX session."""
+    try:
+        listing = subprocess.run(
+            ["ps", "-A", "-o", "pid=,sid="],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        ).stdout
+        return {
+            pid
+            for line in listing.splitlines()
+            for pid, sid in [map(int, line.split())]
+            if sid == session
+        }
+    except (OSError, subprocess.SubprocessError, ValueError):
+        logger.warning("Could not read POSIX session %d", session)
+        return set()
+
+
+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 root not in members:
+        with contextlib.suppress(Exception):
+            process.kill()
+        process.join(timeout=timeout)
+        return
+
+    while members:
+        for pid in members:
+            with contextlib.suppress(ProcessLookupError, PermissionError):
+                os.kill(pid, signal.SIGSTOP)
+        discovered = _session_processes(root)
+        if discovered <= members:
+            break
+        members.update(discovered)
+
+    for pid in members:
+        with contextlib.suppress(ProcessLookupError, PermissionError):
+            os.kill(pid, signal.SIGKILL)
+    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