Merge branch 'main' into fix-repo-adr-cve-2025-58367-deepdiff

Author: pengyuzhang

Detection/README.md

@@ -227,7 +227,7 @@ uv run python main_benchmark.py --benchmark agentdojo --concurrent 3
 - `concurrency_efficiency`: How well concurrency was utilized
 - `overall_mcp_ratio`: Percentage of MCP vs built-in tool usage
 
-**AgentDojo Ground Truth (**`ground_truth.json`**)**:
+**AgentDojo Ground Truth** (`ground_truth.json`):
 
 - `is_malicious`: True if attack succeeded, False if defended
 - `security`: AgentDojo's security field (False = attack succeeded)

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