fix(detection): kill leaked subprocess on task timeout, remove dead scan (#20)

* fix(detection): kill leaked subprocess on task timeout, remove dead scan

_execute_command ran the claude CLI via asyncio.create_subprocess_exec +
asyncio.wait_for(process.communicate(), timeout=...). wait_for only
cancels the await, not the child process, so on timeout the claude CLI
(and any MCP servers it spawned) kept running unkilled. The synchronous
equivalent in guardrail/adr_agent/adr_baseline.py already gets this
right via subprocess.run(..., timeout=...), which kills the child on
TimeoutExpired - this mirrors that same kill-and-reap behavior for the
async path.

Also removes a dead computation in execute_ads_task: existing_sessions
was built via a full recursive rglob("*.jsonl") over the host-wide,
ever-growing ~/.claude/projects/ directory on every single task, then
passed into _process_results, which never referenced it. Confirmed via
grep - pure wasted I/O, safe to delete along with the now-unused
parameter.

Added TestTaskExecutorExecuteCommand covering the timeout-kill behavior
(verified by temporarily reverting the fix and confirming the new test
fails against the original code - the leaked process's returncode was
still None after the call), the success path, and non-zero exit
handling.

Fixes #18

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* fix(detection): kill whole process group on timeout, not just the direct child

Addresses review feedback on PR #20 from @pengyuzhang, who ran a
differential harness (child spawning a grandchild subprocess, probing
both PIDs after _execute_command returns) and found two real gaps in
the previous kill-on-timeout fix:

1. process.kill() only signals the claude CLI itself - any descendant
   it spawned (e.g. an MCP server) survives.
2. await process.wait() can block far past the configured timeout -
   potentially indefinitely - if a descendant inherited the CLI's
   stdout/stderr pipes and holds them open, since asyncio's subprocess
   transport only reports completion once those pipes close. Measured
   at 120s in the review's harness for a 2s timeout; an unbounded
   orphan would wedge one of the max_concurrent execution slots
   forever.

Fix, per the review's suggested approach: start_new_session=True puts
the CLI and everything it spawns in one process group. On timeout,
signal the whole group - SIGTERM first for a clean shutdown, bounded
wait, SIGKILL escalation if it hasn't exited within the grace period.
Falls back to killing just the direct child on platforms without
process-group support (os.killpg), since start_new_session and
os.killpg are POSIX-only - this benchmark harness has no other
Windows-specific handling, but the fallback keeps the code from
raising if it's ever imported there.

Added test_kills_grandchild_process_holding_inherited_stdio,
reproducing the review's exact scenario: a child that spawns its own
child without redirecting stdout/stderr, so the grandchild inherits
the same pipes asyncio set up for the direct child. Asserts the call
returns promptly (not hanging toward the grandchild's full sleep
duration) and that both processes are actually gone afterward, polling
via os.kill(pid, 0) rather than a single check to avoid a false
negative against a not-yet-reaped zombie.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Pengyu Zhang <zhangelsu@gmail.com>

Author: 3 people

Detection/main_benchmark.py

@@ -9,9 +9,11 @@
 import asyncio
 import json
 import logging
+import os
 import random
 import re
 import shutil
+import signal
 import sys
 import time
 from datetime import datetime
@@ -675,11 +677,6 @@ async def execute_ads_task(self, task: Dict[str, Any], workspace_dir: Path,
                 capabilities = config.get("capabilities", [])
                 mcp_tools.extend([f"mcp__{server}__{tool}" for tool in capabilities])
 
-            claude_projects_dir = Path.home() / ".claude" / "projects"
-            existing_sessions = set()
-            if claude_projects_dir.exists():
-                existing_sessions = {f.name for f in claude_projects_dir.rglob("*.jsonl")}
-
             cmd = self.command_builder.build_claude_command(task, mcp_tools)
 
             success, error_message, result = await self._execute_command(cmd, workspace_dir)
@@ -689,9 +686,7 @@ async def execute_ads_task(self, task: Dict[str, Any], workspace_dir: Path,
 
             execution_time = time.time() - start_time
 
-            return await self._process_results(
-                workspace_dir, existing_sessions, result, execution_time
-            )
+            return await self._process_results(workspace_dir, result, execution_time)
 
         except Exception as e:
             print(f"❌ Error executing task: {e}")
@@ -781,12 +776,17 @@ async def execute_agentdojo_task(self, task: Dict[str, Any], output_dir: Path) -
 
     async def _execute_command(self, cmd: List[str], workspace_dir: Path) -> Tuple[bool, Optional[str], Optional[Dict[str, Any]]]:
         """Execute Claude CLI command."""
+        process = None
         try:
             process = await asyncio.create_subprocess_exec(
                 *cmd,
                 cwd=workspace_dir,
                 stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE
+                stderr=asyncio.subprocess.PIPE,
+                # Puts the CLI and everything it spawns (MCP servers, etc.) in
+                # one process group, so a timeout can signal all of them at
+                # once instead of just the direct child. POSIX only.
+                start_new_session=(os.name == "posix"),
             )
 
             stdout, stderr = await asyncio.wait_for(
@@ -802,11 +802,48 @@ async def _execute_command(self, cmd: List[str], workspace_dir: Path) -> Tuple[b
             return True, None, result
 
         except asyncio.TimeoutError:
+            # asyncio.wait_for() only cancels the await - it does not touch the
+            # child process. Killing just the direct child isn't enough either:
+            # it doesn't reach any descendants the CLI spawned (e.g. MCP
+            # servers), and if one of those descendants inherited the CLI's
+            # stdout/stderr pipes, it can hold them open indefinitely -
+            # asyncio's subprocess transport only reports completion once
+            # those pipes close, so process.wait() could hang forever instead
+            # of just leaking. Signal the whole process group instead: SIGTERM
+            # first for a clean shutdown, then SIGKILL if it hasn't exited
+            # within the grace period.
+            if process is not None and process.returncode is None:
+                await self._kill_process_tree(process)
             return False, "Task execution timed out", None
         except Exception as e:
             return False, str(e), None
 
-    async def _process_results(self, workspace_dir: Path, existing_sessions: set,
+    async def _kill_process_tree(self, process: "asyncio.subprocess.Process", grace_period: float = 5.0) -> None:
+        """Terminate a process and everything it spawned.
+
+        Falls back to killing just the direct child on platforms without
+        process-group support (e.g. Windows), matching create_subprocess_exec's
+        start_new_session=(os.name == "posix") above.
+        """
+        if not hasattr(os, "killpg"):
+            process.kill()
+            await process.wait()
+            return
+
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except ProcessLookupError:
+            return
+
+        try:
+            await asyncio.wait_for(process.wait(), timeout=grace_period)
+        except asyncio.TimeoutError:
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+    async def _process_results(self, workspace_dir: Path,
                               result: Dict[str, Any], execution_time: float) -> Tuple[bool, Optional[str], Optional[Dict[str, Any]]]:
         """Process execution results using session ID from Claude CLI JSON output."""
         session_id = result.get("session_id")

Detection/tests/test_main_benchmark.py

@@ -1,11 +1,15 @@
 """Tests for main_benchmark helpers."""
 
+import asyncio
+import os
+import sys
+import time
 from pathlib import Path
 from unittest.mock import MagicMock
 
 import pytest
 
-from main_benchmark import CommandBuilder, Config, MCPServerManager, TaskManager
+from main_benchmark import CommandBuilder, Config, MCPServerManager, TaskExecutor, TaskManager
 
 
 class TestConfig:
@@ -101,3 +105,134 @@ def test_rejects_zero_concurrency(self):
         with pytest.raises(ValueError, match="max_concurrent_tasks must be >= 1"):
             if max_concurrent < 1:
                 raise ValueError(f"max_concurrent_tasks must be >= 1, got {max_concurrent}")
+
+
+class TestTaskExecutorExecuteCommand:
+    """Covers _execute_command's process lifecycle, in particular that a timed-out
+    claude CLI subprocess is actually killed rather than left running (previously
+    asyncio.wait_for's timeout only cancelled the await, not the child process)."""
+
+    def _make_executor(self, tmp_path: Path, monkeypatch, max_execution_time: float) -> TaskExecutor:
+        monkeypatch.chdir(tmp_path)
+        config = Config()
+        config._config["execution"]["max_execution_time"] = max_execution_time
+        return TaskExecutor(config)
+
+    @pytest.mark.asyncio
+    async def test_kills_process_on_timeout(self, tmp_path: Path, monkeypatch):
+        executor = self._make_executor(tmp_path, monkeypatch, max_execution_time=0.05)
+
+        real_create_subprocess_exec = asyncio.create_subprocess_exec
+        spawned = {}
+
+        async def spying_create_subprocess_exec(*args, **kwargs):
+            process = await real_create_subprocess_exec(*args, **kwargs)
+            spawned["process"] = process
+            return process
+
+        monkeypatch.setattr(asyncio, "create_subprocess_exec", spying_create_subprocess_exec)
+
+        cmd = [sys.executable, "-c", "import time; time.sleep(5)"]
+        success, error_message, result = await executor._execute_command(cmd, tmp_path)
+
+        assert success is False
+        assert "timed out" in error_message.lower()
+        assert result is None
+
+        # The process must have been killed and reaped, not left running in the
+        # background after _execute_command returns.
+        process = spawned["process"]
+        assert process.returncode is not None
+
+    @staticmethod
+    def _wait_until_process_gone(pid: int, timeout: float = 3.0) -> bool:
+        """Poll until a pid is fully reaped rather than asserting immediately -
+        os.kill(pid, 0) on a not-yet-reaped zombie still succeeds, so a single
+        check right after sending the kill signal can be a false negative."""
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            try:
+                os.kill(pid, 0)
+            except ProcessLookupError:
+                return True
+            except PermissionError:
+                return False
+            time.sleep(0.1)
+        return False
+
+    @pytest.mark.asyncio
+    @pytest.mark.skipif(os.name != "posix", reason="process-group kill is POSIX-only")
+    async def test_kills_grandchild_process_holding_inherited_stdio(self, tmp_path: Path, monkeypatch):
+        """A descendant that inherits the CLI's stdout/stderr pipes must also be
+        killed on timeout - process.kill() alone only signals the direct child,
+        and a surviving descendant holding those pipes open can block
+        process.wait() indefinitely instead of just leaking (see PR #20 review)."""
+        executor = self._make_executor(tmp_path, monkeypatch, max_execution_time=0.2)
+
+        real_create_subprocess_exec = asyncio.create_subprocess_exec
+        spawned = {}
+
+        async def spying_create_subprocess_exec(*args, **kwargs):
+            process = await real_create_subprocess_exec(*args, **kwargs)
+            spawned["process"] = process
+            return process
+
+        monkeypatch.setattr(asyncio, "create_subprocess_exec", spying_create_subprocess_exec)
+
+        pidfile = tmp_path / "grandchild.pid"
+        # The direct child spawns its own child ("grandchild") without
+        # redirecting its stdout/stderr, so it inherits the same pipes
+        # asyncio set up for the direct child - reproducing the scenario
+        # where a descendant keeps those pipes open past the timeout.
+        script = (
+            "import subprocess, sys, time\n"
+            "gc = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])\n"
+            "open(sys.argv[1], 'w').write(str(gc.pid))\n"
+            "time.sleep(30)\n"
+        )
+        cmd = [sys.executable, "-c", script, str(pidfile)]
+
+        start = time.monotonic()
+        success, error_message, result = await executor._execute_command(cmd, tmp_path)
+        elapsed = time.monotonic() - start
+
+        assert success is False
+        assert "timed out" in error_message.lower()
+
+        # Must return promptly, not hang for anywhere near the grandchild's
+        # 30s sleep - proves process.wait() wasn't blocked on inherited pipes.
+        assert elapsed < 10
+
+        process = spawned["process"]
+        assert process.returncode is not None
+
+        for _ in range(30):
+            if pidfile.exists():
+                break
+            await asyncio.sleep(0.1)
+        assert pidfile.exists(), "grandchild never reported its pid"
+
+        grandchild_pid = int(pidfile.read_text())
+        assert self._wait_until_process_gone(grandchild_pid), "grandchild process was left running"
+
+    @pytest.mark.asyncio
+    async def test_returns_parsed_json_on_success(self, tmp_path: Path, monkeypatch):
+        executor = self._make_executor(tmp_path, monkeypatch, max_execution_time=30)
+
+        cmd = [sys.executable, "-c", "import json, sys; sys.stdout.write(json.dumps({'session_id': 'abc123'}))"]
+        success, error_message, result = await executor._execute_command(cmd, tmp_path)
+
+        assert success is True
+        assert error_message is None
+        assert result == {"session_id": "abc123"}
+
+    @pytest.mark.asyncio
+    async def test_reports_nonzero_exit_without_leaving_error_message_empty(self, tmp_path: Path, monkeypatch):
+        executor = self._make_executor(tmp_path, monkeypatch, max_execution_time=30)
+
+        cmd = [sys.executable, "-c", "import sys; sys.stderr.write('boom'); sys.exit(1)"]
+        success, error_message, result = await executor._execute_command(cmd, tmp_path)
+
+        assert success is False
+        assert "boom" in error_message
+        assert result is None