cactus183/patchbench-dev

1import ast2import os3import re4import subprocess5import tempfile6from pathlib import Path7from typing import Any8 9PYTEST_TIMEOUT_SECONDS = 1510 11 12def _parse_pytest_output(stdout: str, stderr: str) -> tuple[set[str], set[str]]:13    """14    Parse pytest verbose output to extract passing and failing test names.15    Returns (passing_tests, failing_tests) as sets of test names.16    """17    passing = set()18    failing = set()19    combined = stdout + "\n" + stderr20 21    line_pattern = re.compile(r"(test_\w+\.py::[\w\[\]\-]+)\s+(PASSED|FAILED|ERROR)")22    for match in line_pattern.finditer(combined):23        test_name = match.group(1).split("::", 1)[1]24        status = match.group(2)25        if status == "PASSED":26            passing.add(test_name)27        else:28            failing.add(test_name)29 30    return passing, failing31 32 33def _is_valid_python(code: str) -> bool:34    try:35        ast.parse(code)36        return True37    except (SyntaxError, ValueError):38        return False39 40 41def grade_patch(42    patched_code: str,43    test_code: str,44    baseline_passing: set[str],45    baseline_failing: set[str],46) -> tuple[float, dict[str, Any]]:47    """48    Grade a proposed patch by running pytest in a subprocess sandbox.49    Returns (reward in [0.0, 1.0], info dict). Never raises.50    """51    info: dict[str, Any] = {52        "is_valid_python": False,53        "tests_passing": 0,54        "tests_failing": 0,55        "newly_passing": 0,56        "regressions": 0,57        "all_tests_pass": False,58        "grader_error": None,59    }60 61    # Step 1: syntax check62    if not _is_valid_python(patched_code):63        info["grader_error"] = "invalid_python_syntax"64        return (0.0, info)65    info["is_valid_python"] = True66 67    # Step 2: run pytest in sandbox68    try:69        with tempfile.TemporaryDirectory() as tmp_dir:70            tmp_path = Path(tmp_dir)71            (tmp_path / "solution.py").write_text(patched_code)72            (tmp_path / "test_solution.py").write_text(test_code)73 74            parent_env = os.environ.copy()75            parent_env["PYTHONPATH"] = str(tmp_path) + os.pathsep + parent_env.get("PYTHONPATH", "")76            parent_env["PYTHONDONTWRITEBYTECODE"] = "1"77 78            result = subprocess.run(79                ["python", "-m", "pytest", "test_solution.py", "-v", "--tb=no", "-p", "no:cacheprovider"],80                cwd=str(tmp_path),81                capture_output=True,82                text=True,83                timeout=PYTEST_TIMEOUT_SECONDS,84                env=parent_env,85            )86            stdout = result.stdout or ""87            stderr = result.stderr or ""88 89            currently_passing, currently_failing = _parse_pytest_output(stdout, stderr)90 91            # If we parsed nothing at all, fall back to exit code heuristic92            if not currently_passing and not currently_failing:93                if result.returncode == 0:94                    currently_passing = baseline_passing | baseline_failing95                    currently_failing = set()96                else:97                    currently_passing = set()98                    currently_failing = baseline_passing | baseline_failing99 100    except subprocess.TimeoutExpired:101        info["grader_error"] = "pytest_timeout"102        currently_passing = set()103        currently_failing = baseline_passing | baseline_failing104    except Exception as exc:105        info["grader_error"] = f"subprocess_error:{type(exc).__name__}"106        return (0.1, info)107 108    # Step 3: compute diffs109    newly_passing = (currently_passing - baseline_passing) & baseline_failing110    regressions = baseline_passing - currently_passing111    total_originally_failing = len(baseline_failing)112    all_pass = len(currently_failing) == 0 and len(currently_passing) > 0113 114    info.update({115        "tests_passing": len(currently_passing),116        "tests_failing": len(currently_failing),117        "newly_passing": len(newly_passing),118        "regressions": len(regressions),119        "all_tests_pass": all_pass,120    })121 122    # Step 4: reward shaping123    raw = 0.3  # valid python bonus124    if total_originally_failing > 0:125        raw += 0.4 * (len(newly_passing) / total_originally_failing)126    raw -= 0.5 * len(regressions)127    raw -= 0.1  # step cost128    if all_pass:129        raw += 0.3130 131    # Normalize to [0.0, 1.0]: map [-0.6, +1.0] -> [0.0, 1.0]132    normalized = (raw + 0.6) / 1.6133    reward = max(0.0, min(1.0, normalized))134 135    info["reward_breakdown"] = {136        "syntax_valid_bonus": 0.3 if info["is_valid_python"] else 0.0,137        "progress_reward": round(0.4 * (len(newly_passing) / total_originally_failing), 4) if total_originally_failing > 0 else 0.0,138        "regression_penalty": round(-0.5 * len(regressions), 4),139        "step_cost": -0.1,140        "terminal_bonus": 0.3 if all_pass else 0.0,141        "raw_sum": round(raw, 4),142        "normalized": round(reward, 4),143    }144 145    return (reward, info)146