feat: async test subprocess execution and --parallel-candidates flag

Add async_execute_test_subprocess() for running pytest via
asyncio.create_subprocess_exec with stdout/stderr capture and
timeout support.

Add --parallel-candidates N CLI argument to control worktree pool
size. Add anyio dependency for structured concurrency.

Author: KRRT7

codeflash/cli_cmds/cli.py

@@ -500,6 +500,13 @@ def _build_parser() -> ArgumentParser:
     )
     parser.add_argument("--no-draft", default=False, action="store_true", help="Skip optimization for draft PRs")
     parser.add_argument("--worktree", default=False, action="store_true", help="Use worktree for optimization")
+    parser.add_argument(
+        "--parallel-candidates",
+        type=int,
+        default=0,
+        metavar="N",
+        help="Evaluate up to N optimization candidates in parallel using git worktrees (0 = sequential)",
+    )
     parser.add_argument(
         "--testgen-review", default=False, action="store_true", help="Enable AI review and repair of generated tests"
     )

codeflash/languages/python/test_runner.py

@@ -10,7 +10,6 @@
 
 from codeflash.cli_cmds.console import logger
 from codeflash.code_utils.code_utils import custom_addopts
-from codeflash.code_utils.shell_utils import get_cross_platform_subprocess_run_args
 from codeflash.languages.registry import get_language_support
 
 # Pattern to extract timing from stdout markers: !######...:<duration_ns>######!
@@ -92,11 +91,35 @@ def _ensure_runtime_files(project_root: Path, language: str = "javascript") -> N
 
 def execute_test_subprocess(
     cmd_list: list[str], cwd: Path, env: dict[str, str] | None, timeout: int = 600
-) -> subprocess.CompletedProcess:
+) -> subprocess.CompletedProcess[str]:
     """Execute a subprocess with the given command list, working directory, environment variables, and timeout."""
     logger.debug(f"executing test run with command: {' '.join(cmd_list)}")
     with custom_addopts():
-        run_args = get_cross_platform_subprocess_run_args(
-            cwd=cwd, env=env, timeout=timeout, check=False, text=True, capture_output=True
-        )
-        return subprocess.run(cmd_list, **run_args)  # noqa: PLW1510
+        return subprocess.run(cmd_list, cwd=cwd, env=env, timeout=timeout, check=False, text=True, capture_output=True)
+
+
+async def async_execute_test_subprocess(
+    cmd_list: list[str], cwd: Path, env: dict[str, str] | None, timeout: int = 600
+) -> subprocess.CompletedProcess[str]:
+    """Execute a test subprocess asynchronously using anyio."""
+    import os as _os
+
+    import anyio
+
+    logger.debug(f"async executing test run with command: {' '.join(cmd_list)}")
+
+    merged_env = _os.environ.copy()
+    if env:
+        merged_env.update(env)
+
+    with custom_addopts():
+        try:
+            with anyio.fail_after(timeout):
+                result = await anyio.run_process(cmd_list, cwd=cwd, env=merged_env, check=False)
+        except TimeoutError as e:
+            raise subprocess.TimeoutExpired(cmd_list, timeout) from e
+
+    stdout = result.stdout.decode("utf-8", errors="replace") if result.stdout else ""
+    stderr = result.stderr.decode("utf-8", errors="replace") if result.stderr else ""
+
+    return subprocess.CompletedProcess(args=cmd_list, returncode=result.returncode, stdout=stdout, stderr=stderr)

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
     "tomlkit>=0.14.0",
     "attrs>=26.1.0",
     "requests>=2.32.5",
+    "anyio>=4.4.0",
     "junitparser>=4.0.2",
     "pydantic>=2.13.3",
     "humanize>=4.13.0",