From b28f23f32c68b3315f364bdee5ffd50f034c36e8 Mon Sep 17 00:00:00 2001 From: Bjjj834 Date: Mon, 13 Jul 2026 16:34:07 -0400 Subject: [PATCH] Add SWE-bench agent runner --- .gitignore | 6 + docs/swebench_experiment.md | 73 ++++-- pyproject.toml | 3 + scripts/convert_swebench_report.py | 93 ++++++++ scripts/run_swebench_agent.py | 112 +++++++++ src/codebench/__init__.py | 4 + src/codebench/swebench_results.py | 129 ++++++++++ src/codebench/swebench_runner.py | 362 +++++++++++++++++++++++++++++ tests/test_swebench_results.py | 130 +++++++++++ tests/test_swebench_runner.py | 256 ++++++++++++++++++++ 10 files changed, 1150 insertions(+), 18 deletions(-) create mode 100644 scripts/convert_swebench_report.py create mode 100644 scripts/run_swebench_agent.py create mode 100644 src/codebench/swebench_results.py create mode 100644 src/codebench/swebench_runner.py create mode 100644 tests/test_swebench_results.py create mode 100644 tests/test_swebench_runner.py diff --git a/.gitignore b/.gitignore index a7e9bea..9f61718 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,9 @@ build/ # Regenerable SWE-bench artifacts (sample contains gold patches — keep out of git) data/swebench_verified_sample.jsonl predictions/swebench_predictions_template.jsonl + +# SWE-bench runner artifacts (repo clones, per-attempt logs, raw prediction runs) +workspaces/ +attempts/ +predictions/*.jsonl +!predictions/*_v*.jsonl diff --git a/docs/swebench_experiment.md b/docs/swebench_experiment.md index cce232f..8df9678 100644 --- a/docs/swebench_experiment.md +++ b/docs/swebench_experiment.md @@ -63,41 +63,78 @@ sampled task in the official SWE-bench harness format: ### 3. Run agents to generate patches -For each task, give the agent only the agent-safe fields (`task.description`, -`task.repo`, `base_commit`, tags). The agent checks out `repo` at -`base_commit`, works on the issue, and produces a unified diff. Fill that diff -into the `model_patch` field of the corresponding template line. (Full agent -automation is intentionally not implemented yet.) +`scripts/run_swebench_agent.py` automates real attempts. Per task × attempt it +checks out the target repo at `base_commit` in an isolated git worktree +(clones are cached under `workspaces/repos/`), runs your agent command there, +captures `git diff` against `base_commit` as `model_patch`, and writes one +official predictions JSONL per attempt index. + +```bash +# dry-run the whole pipeline first (no real agent, empty patches) +python scripts/run_swebench_agent.py --agent-cmd noop --run-name dryrun + +# 3 independent attempts per task with Claude Code +python scripts/run_swebench_agent.py \ + --model-name anote-code --attempts 3 --timeout 1800 --run-name v1 \ + --agent-cmd 'claude -p "$(cat {prompt_file})" --permission-mode acceptEdits' +# → predictions/v1_attempt1.jsonl, v1_attempt2.jsonl, v1_attempt3.jsonl +``` + +`--agent-cmd` is a shell template with `{prompt_file}`, `{workdir}`, +`{instance_id}`, and `{repo}` placeholders, so any agent CLI plugs in. The +agent's prompt and working directory contain only agent-safe data — the +runner strips `hidden_reference` at load time (`load_agent_safe_tasks`) and +never writes gold fields into prompts, logs, or attempt records +(`attempts/{run}/{instance}/attempt-{k}/`). Failed or timed-out attempts are +recorded with an empty patch so they still count as rollouts. Use `--resume` +to continue an interrupted run and `--attempts N` for multiple independent +rollouts per task (the harness allows one prediction per instance per file, +hence one predictions file per attempt index). ### 4. Evaluate with the official SWE-bench harness ```bash -pip install swebench -python -m swebench.harness.run_evaluation \ - --dataset_name SWE-bench/SWE-bench_Verified \ - --predictions_path predictions/swebench_predictions_template.jsonl \ - --max_workers 4 \ - --run_id codebench-swebench-v0 +pip install swebench # or: pip install -e ".[eval]" +for k in 1 2 3; do + python -m swebench.harness.run_evaluation \ + --dataset_name SWE-bench/SWE-bench_Verified \ + --predictions_path predictions/v1_attempt$k.jsonl \ + --max_workers 2 \ + --run_id v1-attempt$k +done ``` -The harness applies each `model_patch` in a containerized environment and runs -the hidden FAIL_TO_PASS / PASS_TO_PASS tests. +The harness applies each `model_patch` in a containerized environment +(Docker must be running) and runs the hidden FAIL_TO_PASS / PASS_TO_PASS +tests. Each run writes a summary `{model}.{run_id}.json` plus per-instance +reports under `logs/run_evaluation/{run_id}/{model}/`. ### 5. Convert results back into CodeBench metrics -From the harness report, build `ExecutionResult` objects per (task, agent): +```bash +python scripts/convert_swebench_report.py \ + --run-ids v1-attempt1 v1-attempt2 v1-attempt3 \ + --model-name anote-code -k 3 \ + --output data/swebench_results_anote-code_v1.json +``` + +Each evaluated (instance, attempt) becomes one `ExecutionResult` rollout: - `tests_passed` / `tests_total` — from the FAIL_TO_PASS + PASS_TO_PASS outcomes - `regression_count` — PASS_TO_PASS tests that now fail - `execution_success` — the instance is marked *resolved* -These feed directly into the existing `codebench.evaluate` leaderboard -(pass@k, regression rate, cost-adjusted score) without any API changes. +Pooling one harness run per attempt gives exactly the rollout shape +`codebench.evaluate.reliability_at_k` consumes (n = attempts, c = resolved +attempts), replacing synthetic rollouts with real ones in the H-experiment +pipeline — no API changes. ## Testing -Adapter tests run offline with fake instances (no Hugging Face download): +Adapter, runner, and results tests all run offline (fake instances, local +git fixture repos, fixture harness reports — no Hugging Face download, no +Docker): ```bash -python -m pytest tests/test_swebench_adapter.py -q +python -m pytest tests/test_swebench_adapter.py tests/test_swebench_runner.py tests/test_swebench_results.py -q ``` diff --git a/pyproject.toml b/pyproject.toml index d29eb9b..ecd7042 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,9 @@ dependencies = [ "tqdm>=4.65", ] +[project.optional-dependencies] +eval = ["swebench>=2.1"] + [tool.setuptools.packages.find] where = ["src"] diff --git a/scripts/convert_swebench_report.py b/scripts/convert_swebench_report.py new file mode 100644 index 0000000..967d38e --- /dev/null +++ b/scripts/convert_swebench_report.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Convert official SWE-bench harness reports into CodeBench metrics. + +Pools one or more harness runs (one per attempt index) into ExecutionResult +rollouts, computes reliability@k / pass-rate / regression aggregates, and +writes a versioned results JSON. + +Usage: + python scripts/convert_swebench_report.py \\ + --run-ids v1-attempt1 v1-attempt2 v1-attempt3 \\ + --model-name anote-code -k 3 \\ + --output data/swebench_results_anote-code_v1.json +""" + +import argparse +import json +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from codebench.swebench_results import ( + DEFAULT_LOGS_DIR, + aggregate_swebench_results, + harness_report_to_execution_results, +) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--run-ids", nargs="+", required=True, + help="Harness run ids to pool (one per attempt index)", + ) + parser.add_argument("--model-name", required=True, help="model_name_or_path used in the runs") + parser.add_argument("-k", type=int, default=5, help="k for reliability@k (default: 5)") + parser.add_argument( + "--summary-dir", default=".", + help="Directory holding {model}.{run_id}.json summaries (default: cwd)", + ) + parser.add_argument( + "--logs-dir", default=DEFAULT_LOGS_DIR, + help=f"Harness evaluation logs dir (default: {DEFAULT_LOGS_DIR})", + ) + parser.add_argument("--output", default=None, help="Results JSON output path") + args = parser.parse_args() + + safe_model = args.model_name.replace("/", "__") + all_results = [] + per_run = {} + for run_id in args.run_ids: + summary_path = os.path.join(args.summary_dir, f"{safe_model}.{run_id}.json") + if not os.path.exists(summary_path): + sys.exit(f"error: harness summary not found: {summary_path}") + results = harness_report_to_execution_results( + summary_path=summary_path, + run_id=run_id, + model_name=args.model_name, + logs_dir=args.logs_dir, + ) + per_run[run_id] = { + "n_instances": len(results), + "resolved": sum(1 for r in results if r.execution_success), + } + all_results.extend(results) + + aggregates = aggregate_swebench_results(all_results, k=args.k) + + print(f"Model: {args.model_name} runs: {', '.join(args.run_ids)}") + for run_id, stats in per_run.items(): + print(f" {run_id}: {stats['resolved']}/{stats['n_instances']} resolved") + print(f"reliability@{args.k}: {aggregates[f'reliability@{args.k}']:.4f}") + print(f"resolve rate: {aggregates['resolve_rate']:.4f}") + print(f"mean pass rate: {aggregates['mean_pass_rate']:.4f}") + print(f"mean regression rate: {aggregates['mean_regression_rate']:.4f}") + + if args.output: + payload = { + "model": args.model_name, + "run_ids": args.run_ids, + "k": args.k, + "per_run": per_run, + "aggregates": aggregates, + "rollouts": [r.model_dump(mode="json") for r in all_results], + } + os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) + with open(args.output, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2) + print(f"\nWrote results to {os.path.abspath(args.output)}") + + +if __name__ == "__main__": + main() diff --git a/scripts/run_swebench_agent.py b/scripts/run_swebench_agent.py new file mode 100644 index 0000000..925da3e --- /dev/null +++ b/scripts/run_swebench_agent.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Run real agent attempts against a SWE-bench Verified sample. + +For each task x attempt: checks out the target repo at base_commit in an +isolated git worktree, runs the agent command there, captures git diff as +model_patch, and writes one official predictions JSONL per attempt index. + +The agent only ever sees the issue description, repo name, base commit, +difficulty, and tags — never the gold patch or hidden tests. + +Usage: + # dry-run the pipeline (no real agent, empty patches) + python scripts/run_swebench_agent.py --agent-cmd noop --run-name dryrun + + # 3 attempts per task with Claude Code + python scripts/run_swebench_agent.py \\ + --model-name anote-code --attempts 3 --run-name v1 \\ + --agent-cmd 'claude -p "$(cat {prompt_file})" --permission-mode acceptEdits' +""" + +import argparse +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from codebench.swebench_runner import ( + DEFAULT_GIT_BASE_URL, + DEFAULT_TIMEOUT_S, + run_swebench_attempts, +) + +ROOT = os.path.join(os.path.dirname(__file__), "..") +DEFAULT_SAMPLE = os.path.join(ROOT, "data", "swebench_verified_sample.jsonl") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--sample", + default=DEFAULT_SAMPLE, + help="Task sample JSONL from prepare_swebench_sample.py", + ) + parser.add_argument( + "--agent-cmd", + default="noop", + help=( + "Shell command template with {prompt_file}/{workdir}/{instance_id}/{repo} " + "placeholders, or 'noop' for a dry run (default: noop)" + ), + ) + parser.add_argument("--model-name", default="anote-code", help="model_name_or_path value") + parser.add_argument("--attempts", type=int, default=1, help="Attempts per task (default: 1)") + parser.add_argument("--run-name", default="run", help="Name for this run's artifacts") + parser.add_argument( + "--timeout", type=int, default=DEFAULT_TIMEOUT_S, + help=f"Per-attempt agent timeout in seconds (default: {DEFAULT_TIMEOUT_S})", + ) + parser.add_argument( + "--workspaces-dir", default=os.path.join(ROOT, "workspaces"), + help="Where repo clones and worktrees live", + ) + parser.add_argument( + "--attempts-dir", default=os.path.join(ROOT, "attempts"), + help="Where per-attempt records (prompt/log/patch/meta) are written", + ) + parser.add_argument( + "--predictions-dir", default=os.path.join(ROOT, "predictions"), + help="Where prediction JSONL files are written", + ) + parser.add_argument( + "--git-base-url", default=DEFAULT_GIT_BASE_URL, + help="Base URL for cloning task repos (default: https://github.com)", + ) + parser.add_argument( + "--resume", action="store_true", + help="Skip (task, attempt) pairs that already have a recorded meta.json", + ) + parser.add_argument( + "--keep-workspaces", action="store_true", + help="Keep per-attempt worktrees for debugging", + ) + args = parser.parse_args() + + prediction_files = run_swebench_attempts( + sample_path=args.sample, + agent_cmd=args.agent_cmd, + model_name=args.model_name, + attempts=args.attempts, + run_name=args.run_name, + workspaces_dir=args.workspaces_dir, + attempts_dir=args.attempts_dir, + predictions_dir=args.predictions_dir, + timeout_s=args.timeout, + resume=args.resume, + keep_workspaces=args.keep_workspaces, + git_base_url=args.git_base_url, + ) + print(f"Run '{args.run_name}' complete: {len(prediction_files)} prediction file(s)") + for attempt, path in sorted(prediction_files.items()): + print(f" attempt {attempt}: {os.path.abspath(path)}") + print( + "\nNext: evaluate each file with the official harness, e.g.\n" + " python -m swebench.harness.run_evaluation " + "--dataset_name SWE-bench/SWE-bench_Verified \\\n" + f" --predictions_path {os.path.abspath(prediction_files[1])} \\\n" + f" --max_workers 2 --run_id {args.run_name}-attempt1" + ) + + +if __name__ == "__main__": + main() diff --git a/src/codebench/__init__.py b/src/codebench/__init__.py index 7abb304..39f13af 100644 --- a/src/codebench/__init__.py +++ b/src/codebench/__init__.py @@ -37,6 +37,8 @@ write_swebench_sample, write_predictions_jsonl, ) +from .swebench_runner import run_swebench_attempts +from .swebench_results import harness_report_to_execution_results __all__ = [ "TaskDifficulty", @@ -68,4 +70,6 @@ "swebench_instance_to_codetask", "write_swebench_sample", "write_predictions_jsonl", + "run_swebench_attempts", + "harness_report_to_execution_results", ] diff --git a/src/codebench/swebench_results.py b/src/codebench/swebench_results.py new file mode 100644 index 0000000..55316b1 --- /dev/null +++ b/src/codebench/swebench_results.py @@ -0,0 +1,129 @@ +"""Convert official SWE-bench harness reports into CodeBench metrics. + +The harness (``python -m swebench.harness.run_evaluation``) produces, per run: + +- a summary JSON ``{model}.{run_id}.json`` in the invocation directory, with + id lists such as ``resolved_ids``, ``unresolved_ids``, ``error_ids``, and + ``empty_patch_ids``; +- per-instance reports at + ``logs/run_evaluation/{run_id}/{model}/{instance_id}/report.json`` with a + ``tests_status`` breakdown of FAIL_TO_PASS / PASS_TO_PASS outcomes. + +Each evaluated (instance, run) becomes one :class:`ExecutionResult` rollout: + +- ``tests_total`` = |FAIL_TO_PASS| + |PASS_TO_PASS| +- ``tests_passed`` = successes across both groups +- ``regression_count`` = PASS_TO_PASS failures +- ``execution_success`` = the instance was resolved + +With one harness run per attempt index, the pooled results are exactly the +rollout shape :func:`codebench.evaluate.reliability_at_k` consumes. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List, Union + +from .core import ExecutionResult +from .evaluate import regression_rate, reliability_at_k + +DEFAULT_LOGS_DIR = "logs/run_evaluation" + + +def load_harness_summary(summary_path: Union[str, Path]) -> Dict[str, Any]: + """Load a harness summary JSON (``{model}.{run_id}.json``).""" + return json.loads(Path(summary_path).read_text(encoding="utf-8")) + + +def _instance_report_path( + logs_dir: Union[str, Path], run_id: str, model_name: str, instance_id: str +) -> Path: + # The harness sanitizes "/" in model names when building directories. + safe_model = model_name.replace("/", "__") + return Path(logs_dir) / run_id / safe_model / instance_id / "report.json" + + +def _result_from_instance_report( + instance_id: str, + model_name: str, + report: Dict[str, Any], +) -> ExecutionResult: + entry = report.get(instance_id, report) + status = entry.get("tests_status", {}) + f2p = status.get("FAIL_TO_PASS", {}) + p2p = status.get("PASS_TO_PASS", {}) + f2p_pass = len(f2p.get("success", [])) + f2p_fail = len(f2p.get("failure", [])) + p2p_pass = len(p2p.get("success", [])) + p2p_fail = len(p2p.get("failure", [])) + return ExecutionResult( + task_id=instance_id, + agent_name=model_name, + tests_passed=f2p_pass + p2p_pass, + tests_total=f2p_pass + f2p_fail + p2p_pass + p2p_fail, + regression_count=p2p_fail, + execution_success=bool(entry.get("resolved", False)), + ) + + +def harness_report_to_execution_results( + summary_path: Union[str, Path], + run_id: str, + model_name: str, + logs_dir: Union[str, Path] = DEFAULT_LOGS_DIR, +) -> List[ExecutionResult]: + """Convert one harness run into ExecutionResult rollouts. + + Instances without a per-instance report (errored or empty-patch attempts) + become zero-score rollouts with ``execution_success=False``, so failed + attempts still count toward reliability@k denominators. + """ + summary = load_harness_summary(summary_path) + instance_ids: List[str] = sorted( + set(summary.get("resolved_ids", [])) + | set(summary.get("unresolved_ids", [])) + | set(summary.get("error_ids", [])) + | set(summary.get("empty_patch_ids", [])) + ) + + results: List[ExecutionResult] = [] + resolved = set(summary.get("resolved_ids", [])) + for instance_id in instance_ids: + report_path = _instance_report_path(logs_dir, run_id, model_name, instance_id) + if report_path.exists(): + report = json.loads(report_path.read_text(encoding="utf-8")) + results.append(_result_from_instance_report(instance_id, model_name, report)) + else: + results.append( + ExecutionResult( + task_id=instance_id, + agent_name=model_name, + tests_passed=0, + tests_total=0, + regression_count=0, + execution_success=instance_id in resolved, + ) + ) + return results + + +def aggregate_swebench_results( + results: List[ExecutionResult], + k: int = 5, +) -> Dict[str, Any]: + """Aggregate pooled rollouts into CodeBench headline metrics.""" + n = len(results) + tasks = sorted({r.task_id for r in results}) + resolved_rollouts = sum(1 for r in results if r.execution_success) + return { + "n_rollouts": n, + "n_tasks": len(tasks), + "resolved_rollouts": resolved_rollouts, + "resolve_rate": resolved_rollouts / max(n, 1), + "mean_pass_rate": sum(r.pass_rate for r in results) / max(n, 1), + "mean_regression_rate": sum(regression_rate(r) for r in results) / max(n, 1), + f"reliability@{k}": reliability_at_k(results, k=k), + "tasks": tasks, + } diff --git a/src/codebench/swebench_runner.py b/src/codebench/swebench_runner.py new file mode 100644 index 0000000..cbee313 --- /dev/null +++ b/src/codebench/swebench_runner.py @@ -0,0 +1,362 @@ +"""Runner for real SWE-bench Verified agent attempts. + +Orchestrates: load agent-safe tasks -> check out each target repo at +``base_commit`` -> run a coding agent in the checkout -> capture ``git diff`` +as ``model_patch`` -> export official SWE-bench prediction JSONL files +(one per attempt index, since the harness allows one prediction per +instance_id per file). + +Leakage policy +-------------- +:func:`load_agent_safe_tasks` is the single choke point that reads sample +files written by :func:`codebench.swebench_adapter.write_swebench_sample`. +It keeps only agent-safe fields; ``hidden_reference`` (gold patch, +test_patch, FAIL_TO_PASS, PASS_TO_PASS) is dropped at parse time and never +reaches any prompt, log, or attempt record. The agent works in a clean +checkout of the *target* repo — no benchmark files are present there. +""" + +from __future__ import annotations + +import json +import shlex +import subprocess +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Union + +from pydantic import BaseModel, Field + +from .swebench_adapter import write_predictions_jsonl + +#: Sentinel value for --agent-cmd: run no agent, produce an empty patch. +NOOP_AGENT = "noop" + +DEFAULT_GIT_BASE_URL = "https://github.com" +DEFAULT_TIMEOUT_S = 1800 + + +class AgentSafeTask(BaseModel): + """The only view of a SWE-bench task the runner ever holds. + + Constructed exclusively by :func:`load_agent_safe_tasks`, which drops + gold fields at parse time. + """ + + task_id: str + repo: str + base_commit: str + description: str + difficulty: str + tags: List[str] = Field(default_factory=list) + + +class AttemptRecord(BaseModel): + """Outcome of one agent attempt on one task.""" + + task_id: str + attempt: int + model_name: str + exit_code: Optional[int] + timed_out: bool + latency_ms: float + model_patch: str + record_dir: str + + +def load_agent_safe_tasks(sample_path: Union[str, Path]) -> List[AgentSafeTask]: + """Load tasks from a sample JSONL, keeping only agent-safe fields. + + Gold data (``hidden_reference``) and infra-only fields are discarded + here and never stored on any runner object. + """ + tasks: List[AgentSafeTask] = [] + with Path(sample_path).open("r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + record = json.loads(line) + task = record["task"] + tasks.append( + AgentSafeTask( + task_id=task["task_id"], + repo=task["repo"], + base_commit=record["base_commit"], + description=task["description"], + difficulty=task["difficulty"], + tags=list(task.get("tags", [])), + ) + ) + return tasks + + +def build_agent_prompt(task: AgentSafeTask) -> str: + """Build the agent-facing prompt from agent-safe fields only.""" + return ( + f"You are working in a checkout of the repository `{task.repo}` " + f"at commit `{task.base_commit}` (the current working directory).\n\n" + f"Resolve the following GitHub issue by modifying the source code. " + f"Focus on the root cause rather than editing tests. " + f"Do not create git commits; leave your changes in the working tree.\n\n" + f"--- ISSUE ({task.task_id}, difficulty: {task.difficulty}) ---\n\n" + f"{task.description}\n" + ) + + +def _git(args: List[str], cwd: Union[str, Path]) -> str: + result = subprocess.run( + ["git", *args], + cwd=str(cwd), + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"git {' '.join(args)} failed in {cwd}: {result.stderr.strip()}" + ) + return result.stdout + + +def ensure_repo_clone( + repo: str, + workspaces_dir: Union[str, Path], + git_base_url: str = DEFAULT_GIT_BASE_URL, +) -> Path: + """Clone ``repo`` once into the workspace cache, or fetch if present.""" + clone_path = Path(workspaces_dir) / "repos" / repo.replace("/", "__") + if clone_path.exists(): + _git(["fetch", "--all", "--quiet"], cwd=clone_path) + else: + clone_path.parent.mkdir(parents=True, exist_ok=True) + url = f"{git_base_url.rstrip('/')}/{repo}" + result = subprocess.run( + ["git", "clone", "--quiet", url, str(clone_path)], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"git clone {url} failed: {result.stderr.strip()}") + return clone_path + + +def create_worktree( + clone_path: Union[str, Path], + base_commit: str, + worktree_path: Union[str, Path], +) -> Path: + """Create a detached worktree of the cached clone at ``base_commit``.""" + worktree_path = Path(worktree_path) + worktree_path.parent.mkdir(parents=True, exist_ok=True) + _git( + ["worktree", "add", "--detach", str(worktree_path), base_commit], + cwd=clone_path, + ) + return worktree_path + + +def remove_worktree(clone_path: Union[str, Path], worktree_path: Union[str, Path]) -> None: + """Remove a worktree created by :func:`create_worktree`.""" + subprocess.run( + ["git", "worktree", "remove", "--force", str(worktree_path)], + cwd=str(clone_path), + capture_output=True, + text=True, + ) + subprocess.run( + ["git", "worktree", "prune"], + cwd=str(clone_path), + capture_output=True, + text=True, + ) + + +def capture_model_patch(worktree_path: Union[str, Path], base_commit: str) -> str: + """Capture the agent's work as a unified diff against ``base_commit``. + + ``git add -A`` stages new untracked files; diffing the index against the + explicit base commit is robust even if the agent created commits. + """ + _git(["add", "-A"], cwd=worktree_path) + patch = _git(["diff", "--cached", base_commit], cwd=worktree_path) + for line in patch.splitlines(): + if line.startswith("diff --git"): + parts = line.split() + paths = [p[2:] for p in parts[2:4] if len(p) > 2] + if any(p.startswith(("/", "..")) or p.startswith(".git/") for p in paths): + raise RuntimeError(f"Refusing patch touching unsafe path: {line}") + return patch + + +def run_agent_command( + agent_cmd: str, + prompt_file: Union[str, Path], + workdir: Union[str, Path], + task: AgentSafeTask, + timeout_s: int = DEFAULT_TIMEOUT_S, + log_path: Optional[Union[str, Path]] = None, +) -> Dict[str, Any]: + """Run one agent attempt and return exit metadata. + + ``agent_cmd`` is a shell template with ``{prompt_file}``, ``{workdir}``, + ``{instance_id}``, and ``{repo}`` placeholders. The sentinel ``"noop"`` + runs nothing (empty patch), for dry runs and tests. + """ + if agent_cmd.strip() == NOOP_AGENT: + return {"exit_code": 0, "timed_out": False, "latency_ms": 0.0} + + command = agent_cmd.format( + prompt_file=shlex.quote(str(prompt_file)), + workdir=shlex.quote(str(workdir)), + instance_id=shlex.quote(task.task_id), + repo=shlex.quote(task.repo), + ) + start = time.monotonic() + timed_out = False + exit_code: Optional[int] = None + stdout = stderr = "" + try: + result = subprocess.run( + command, + shell=True, + cwd=str(workdir), + capture_output=True, + text=True, + timeout=timeout_s, + ) + exit_code = result.returncode + stdout, stderr = result.stdout, result.stderr + except subprocess.TimeoutExpired as exc: + timed_out = True + stdout = exc.stdout or "" if isinstance(exc.stdout, str) else "" + stderr = exc.stderr or "" if isinstance(exc.stderr, str) else "" + latency_ms = (time.monotonic() - start) * 1000.0 + + if log_path is not None: + log_path = Path(log_path) + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + f"$ {command}\n\n--- stdout ---\n{stdout}\n--- stderr ---\n{stderr}\n", + encoding="utf-8", + ) + return {"exit_code": exit_code, "timed_out": timed_out, "latency_ms": latency_ms} + + +def _run_single_attempt( + task: AgentSafeTask, + attempt: int, + agent_cmd: str, + model_name: str, + record_dir: Path, + clone_path: Path, + worktree_path: Path, + timeout_s: int, + keep_workspaces: bool, +) -> AttemptRecord: + record_dir.mkdir(parents=True, exist_ok=True) + prompt_file = record_dir / "prompt.md" + prompt_file.write_text(build_agent_prompt(task), encoding="utf-8") + + create_worktree(clone_path, task.base_commit, worktree_path) + try: + outcome = run_agent_command( + agent_cmd, + prompt_file=prompt_file, + workdir=worktree_path, + task=task, + timeout_s=timeout_s, + log_path=record_dir / "agent.log", + ) + if outcome["timed_out"] or (outcome["exit_code"] not in (0, None)): + model_patch = "" # failed attempt still counts as a rollout + else: + model_patch = capture_model_patch(worktree_path, task.base_commit) + finally: + if not keep_workspaces: + remove_worktree(clone_path, worktree_path) + + record = AttemptRecord( + task_id=task.task_id, + attempt=attempt, + model_name=model_name, + exit_code=outcome["exit_code"], + timed_out=outcome["timed_out"], + latency_ms=outcome["latency_ms"], + model_patch=model_patch, + record_dir=str(record_dir), + ) + (record_dir / "model.patch").write_text(model_patch, encoding="utf-8") + (record_dir / "meta.json").write_text( + json.dumps(record.model_dump(exclude={"model_patch"}), indent=2), + encoding="utf-8", + ) + return record + + +def run_swebench_attempts( + sample_path: Union[str, Path], + agent_cmd: str, + model_name: str = "anote-code", + attempts: int = 1, + run_name: str = "run", + workspaces_dir: Union[str, Path] = "workspaces", + attempts_dir: Union[str, Path] = "attempts", + predictions_dir: Union[str, Path] = "predictions", + timeout_s: int = DEFAULT_TIMEOUT_S, + resume: bool = False, + keep_workspaces: bool = False, + git_base_url: str = DEFAULT_GIT_BASE_URL, +) -> Dict[int, Path]: + """Run ``attempts`` independent agent attempts per sampled task. + + Returns a mapping of attempt index -> written predictions JSONL path + (one file per attempt, in the official SWE-bench prediction format). + """ + if attempts < 1: + raise ValueError(f"attempts must be >= 1, got {attempts}") + + tasks = load_agent_safe_tasks(sample_path) + if not tasks: + raise ValueError(f"No tasks found in {sample_path}") + + records: Dict[int, List[AttemptRecord]] = {k: [] for k in range(1, attempts + 1)} + for task in tasks: + clone_path = ensure_repo_clone(task.repo, workspaces_dir, git_base_url) + for attempt in range(1, attempts + 1): + record_dir = Path(attempts_dir) / run_name / task.task_id / f"attempt-{attempt}" + meta_path = record_dir / "meta.json" + if resume and meta_path.exists(): + meta = json.loads(meta_path.read_text(encoding="utf-8")) + patch = (record_dir / "model.patch").read_text(encoding="utf-8") + records[attempt].append(AttemptRecord(**meta, model_patch=patch)) + continue + worktree_path = ( + Path(workspaces_dir) / "runs" / run_name / f"{task.task_id}__a{attempt}" + ) + records[attempt].append( + _run_single_attempt( + task=task, + attempt=attempt, + agent_cmd=agent_cmd, + model_name=model_name, + record_dir=record_dir, + clone_path=clone_path, + worktree_path=worktree_path, + timeout_s=timeout_s, + keep_workspaces=keep_workspaces, + ) + ) + + prediction_files: Dict[int, Path] = {} + for attempt, attempt_records in records.items(): + predictions = [ + { + "instance_id": r.task_id, + "model_name_or_path": model_name, + "model_patch": r.model_patch, + } + for r in attempt_records + ] + out = Path(predictions_dir) / f"{run_name}_attempt{attempt}.jsonl" + prediction_files[attempt] = write_predictions_jsonl(predictions, out) + return prediction_files diff --git a/tests/test_swebench_results.py b/tests/test_swebench_results.py new file mode 100644 index 0000000..f817dfc --- /dev/null +++ b/tests/test_swebench_results.py @@ -0,0 +1,130 @@ +"""Tests for codebench.swebench_results (fixture harness reports, offline).""" + +import json + +import pytest + +from codebench.core import ExecutionResult +from codebench.swebench_results import ( + aggregate_swebench_results, + harness_report_to_execution_results, +) + +MODEL = "anote-code" +RUN_ID = "v1-attempt1" + + +def write_fixture_run(tmp_path, model=MODEL, run_id=RUN_ID): + """Build a fake harness output tree: summary + per-instance reports.""" + summary = { + "resolved_ids": ["acme__widget-1"], + "unresolved_ids": ["acme__widget-2"], + "error_ids": ["acme__widget-3"], + "empty_patch_ids": [], + } + summary_path = tmp_path / f"{model}.{run_id}.json" + summary_path.write_text(json.dumps(summary), encoding="utf-8") + + logs_dir = tmp_path / "logs" / "run_evaluation" + reports = { + "acme__widget-1": { + "resolved": True, + "tests_status": { + "FAIL_TO_PASS": {"success": ["t1", "t2"], "failure": []}, + "PASS_TO_PASS": {"success": ["t3", "t4", "t5"], "failure": ["t6"]}, + }, + }, + "acme__widget-2": { + "resolved": False, + "tests_status": { + "FAIL_TO_PASS": {"success": [], "failure": ["t1", "t2"]}, + "PASS_TO_PASS": {"success": ["t3", "t4", "t5", "t6"], "failure": []}, + }, + }, + # acme__widget-3 errored: no report.json on purpose + } + for instance_id, report in reports.items(): + report_dir = logs_dir / run_id / model / instance_id + report_dir.mkdir(parents=True) + (report_dir / "report.json").write_text( + json.dumps({instance_id: report}), encoding="utf-8" + ) + return summary_path, logs_dir + + +def test_field_mapping_from_instance_report(tmp_path): + summary_path, logs_dir = write_fixture_run(tmp_path) + results = harness_report_to_execution_results( + summary_path, run_id=RUN_ID, model_name=MODEL, logs_dir=logs_dir + ) + by_id = {r.task_id: r for r in results} + assert set(by_id) == {"acme__widget-1", "acme__widget-2", "acme__widget-3"} + + resolved = by_id["acme__widget-1"] + assert resolved.agent_name == MODEL + assert resolved.tests_total == 6 # 2 F2P + 4 P2P + assert resolved.tests_passed == 5 # 2 F2P + 3 P2P successes + assert resolved.regression_count == 1 # 1 P2P failure + assert resolved.execution_success is True + + unresolved = by_id["acme__widget-2"] + assert unresolved.tests_total == 6 + assert unresolved.tests_passed == 4 + assert unresolved.regression_count == 0 + assert unresolved.execution_success is False + + +def test_missing_report_becomes_zero_rollout(tmp_path): + summary_path, logs_dir = write_fixture_run(tmp_path) + results = harness_report_to_execution_results( + summary_path, run_id=RUN_ID, model_name=MODEL, logs_dir=logs_dir + ) + errored = {r.task_id: r for r in results}["acme__widget-3"] + assert errored.tests_total == 0 + assert errored.tests_passed == 0 + assert errored.execution_success is False + assert errored.pass_rate == 0.0 + + +def test_model_name_with_slash_is_sanitized(tmp_path): + model = "org/model" + summary_path, logs_dir = write_fixture_run(tmp_path, model="org__model") + results = harness_report_to_execution_results( + summary_path, run_id=RUN_ID, model_name=model, logs_dir=logs_dir + ) + # reports live under the sanitized dir but agent_name keeps the real name + assert all(r.agent_name == model for r in results) + assert any(r.tests_total > 0 for r in results) + + +def make_rollout(task_id, success): + return ExecutionResult( + task_id=task_id, + agent_name=MODEL, + tests_passed=6 if success else 3, + tests_total=6, + regression_count=0 if success else 1, + execution_success=success, + ) + + +def test_aggregate_reliability_over_pooled_attempts(): + # one task, 3 attempts (harness runs), 2 resolved + rollouts = [ + make_rollout("acme__widget-1", True), + make_rollout("acme__widget-1", True), + make_rollout("acme__widget-1", False), + ] + agg = aggregate_swebench_results(rollouts, k=1) + assert agg["n_rollouts"] == 3 + assert agg["n_tasks"] == 1 + assert agg["resolved_rollouts"] == 2 + assert agg["reliability@1"] == pytest.approx(2 / 3) # pass_at_k(3, 2, 1) + agg3 = aggregate_swebench_results(rollouts, k=3) + assert agg3["reliability@3"] == 1.0 # pass_at_k(3, 2, 3) + + +def test_aggregate_empty_results(): + agg = aggregate_swebench_results([], k=5) + assert agg["n_rollouts"] == 0 + assert agg["reliability@5"] == 0.0 diff --git a/tests/test_swebench_runner.py b/tests/test_swebench_runner.py new file mode 100644 index 0000000..7c49354 --- /dev/null +++ b/tests/test_swebench_runner.py @@ -0,0 +1,256 @@ +"""Tests for codebench.swebench_runner (offline: local git repos, no network).""" + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from codebench.swebench_runner import ( + build_agent_prompt, + capture_model_patch, + create_worktree, + ensure_repo_clone, + load_agent_safe_tasks, + run_swebench_attempts, +) + +GOLD_SENTINEL = "XGOLDPATCHX-must-never-leak" +HIDDEN_TEST_SENTINEL = "XHIDDENTESTX-must-never-leak" + + +def _git(args, cwd): + subprocess.run( + ["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], + cwd=str(cwd), check=True, capture_output=True, text=True, + ) + + +def _sha(cwd): + return subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=str(cwd), + check=True, capture_output=True, text=True, + ).stdout.strip() + + +@pytest.fixture +def fake_remote(tmp_path): + """A local 'remote' repo acme/widget with two commits. + + base_commit is the FIRST commit, so tests prove worktrees check out + base_commit rather than the branch tip. + """ + repo = tmp_path / "remotes" / "acme" / "widget" + repo.mkdir(parents=True) + _git(["init", "-q"], repo) + (repo / "app.py").write_text("VERSION = 1\n", encoding="utf-8") + _git(["add", "-A"], repo) + _git(["commit", "-q", "-m", "base"], repo) + base_commit = _sha(repo) + (repo / "app.py").write_text("VERSION = 2\n", encoding="utf-8") + _git(["add", "-A"], repo) + _git(["commit", "-q", "-m", "tip"], repo) + return {"git_base_url": str(tmp_path / "remotes"), "repo": "acme/widget", + "base_commit": base_commit} + + +@pytest.fixture +def sample_file(tmp_path, fake_remote): + """A poisoned sample JSONL: gold fields carry sentinel strings.""" + record = { + "task": { + "task_id": "acme__widget-1", + "repo": fake_remote["repo"], + "description": "The widget crashes when VERSION is read.", + "difficulty": "medium", + "tags": ["swebench", "acme/widget", "medium"], + "test_file": "swebench://hidden-tests", + "reference_solution": "swebench://hidden-gold-patch", + }, + "base_commit": fake_remote["base_commit"], + "environment_setup_commit": "envsha", + "hidden_reference": { + "patch": GOLD_SENTINEL, + "test_patch": HIDDEN_TEST_SENTINEL, + "FAIL_TO_PASS": f'["{HIDDEN_TEST_SENTINEL}::t1"]', + "PASS_TO_PASS": "[]", + "_note": "reference-only", + }, + } + path = tmp_path / "sample.jsonl" + path.write_text(json.dumps(record) + "\n", encoding="utf-8") + return path + + +def run_kwargs(tmp_path, sample_file, fake_remote, **overrides): + kwargs = dict( + sample_path=sample_file, + agent_cmd="noop", + model_name="test-agent", + attempts=1, + run_name="t", + workspaces_dir=tmp_path / "workspaces", + attempts_dir=tmp_path / "attempts", + predictions_dir=tmp_path / "predictions", + git_base_url=fake_remote["git_base_url"], + ) + kwargs.update(overrides) + return kwargs + + +# ------------------------------------------------------------- safe loading + +def test_load_agent_safe_tasks_drops_hidden_fields(sample_file): + tasks = load_agent_safe_tasks(sample_file) + assert len(tasks) == 1 + task = tasks[0] + assert task.task_id == "acme__widget-1" + assert task.repo == "acme/widget" + assert task.base_commit + dumped = json.dumps(task.model_dump()) + assert GOLD_SENTINEL not in dumped + assert HIDDEN_TEST_SENTINEL not in dumped + assert "hidden_reference" not in dumped + + +def test_prompt_contains_only_allowed_fields(sample_file): + task = load_agent_safe_tasks(sample_file)[0] + prompt = build_agent_prompt(task) + assert task.description in prompt + assert task.repo in prompt + assert task.base_commit in prompt + assert GOLD_SENTINEL not in prompt + assert HIDDEN_TEST_SENTINEL not in prompt + + +# ---------------------------------------------------------------- checkout + +def test_worktree_checks_out_base_commit_not_tip(tmp_path, fake_remote): + clone = ensure_repo_clone( + fake_remote["repo"], tmp_path / "workspaces", fake_remote["git_base_url"] + ) + wt = create_worktree(clone, fake_remote["base_commit"], tmp_path / "wt") + assert (wt / "app.py").read_text() == "VERSION = 1\n" # not the tip's VERSION = 2 + + +def test_clone_is_cached(tmp_path, fake_remote): + c1 = ensure_repo_clone( + fake_remote["repo"], tmp_path / "workspaces", fake_remote["git_base_url"] + ) + c2 = ensure_repo_clone( + fake_remote["repo"], tmp_path / "workspaces", fake_remote["git_base_url"] + ) + assert c1 == c2 and c1.exists() + + +# ------------------------------------------------------------- diff capture + +def test_capture_model_patch_includes_new_and_edited_files(tmp_path, fake_remote): + clone = ensure_repo_clone( + fake_remote["repo"], tmp_path / "workspaces", fake_remote["git_base_url"] + ) + wt = create_worktree(clone, fake_remote["base_commit"], tmp_path / "wt") + (wt / "app.py").write_text("VERSION = 1\nFIXED = True\n", encoding="utf-8") + (wt / "new_module.py").write_text("x = 1\n", encoding="utf-8") + patch = capture_model_patch(wt, fake_remote["base_commit"]) + assert "FIXED = True" in patch + assert "new_module.py" in patch + + +# ------------------------------------------------------- end-to-end running + +def test_noop_run_produces_empty_patch_predictions(tmp_path, sample_file, fake_remote): + files = run_swebench_attempts(**run_kwargs(tmp_path, sample_file, fake_remote)) + assert set(files) == {1} + rows = [json.loads(l) for l in files[1].read_text().splitlines()] + assert rows == [ + {"instance_id": "acme__widget-1", "model_name_or_path": "test-agent", + "model_patch": ""} + ] + record_dir = tmp_path / "attempts" / "t" / "acme__widget-1" / "attempt-1" + assert (record_dir / "prompt.md").exists() + assert (record_dir / "meta.json").exists() + assert (record_dir / "model.patch").read_text() == "" + + +def test_scripted_agent_patch_roundtrip(tmp_path, sample_file, fake_remote): + # a "real" agent: writes fix.py into its working directory + agent_cmd = ( + f'{sys.executable} -c ' + f'"import pathlib; pathlib.Path(\'fix.py\').write_text(\'patched\')"' + ) + files = run_swebench_attempts( + **run_kwargs(tmp_path, sample_file, fake_remote, agent_cmd=agent_cmd) + ) + row = json.loads(files[1].read_text().splitlines()[0]) + assert set(row) == {"instance_id", "model_name_or_path", "model_patch"} + assert "fix.py" in row["model_patch"] + assert "patched" in row["model_patch"] + # worktree cleaned up by default + assert not (tmp_path / "workspaces" / "runs" / "t").exists() or not any( + (tmp_path / "workspaces" / "runs" / "t").iterdir() + ) + + +def test_multiple_attempts_one_predictions_file_each(tmp_path, sample_file, fake_remote): + files = run_swebench_attempts( + **run_kwargs(tmp_path, sample_file, fake_remote, attempts=3) + ) + assert set(files) == {1, 2, 3} + for k, path in files.items(): + assert path.name == f"t_attempt{k}.jsonl" + assert len(path.read_text().splitlines()) == 1 + for k in (1, 2, 3): + assert (tmp_path / "attempts" / "t" / "acme__widget-1" / f"attempt-{k}").exists() + + +def test_failed_agent_yields_empty_patch(tmp_path, sample_file, fake_remote): + files = run_swebench_attempts( + **run_kwargs(tmp_path, sample_file, fake_remote, + agent_cmd=f"{sys.executable} -c \"raise SystemExit(3)\"") + ) + row = json.loads(files[1].read_text().splitlines()[0]) + assert row["model_patch"] == "" + meta = json.loads( + (tmp_path / "attempts" / "t" / "acme__widget-1" / "attempt-1" / "meta.json").read_text() + ) + assert meta["exit_code"] == 3 + + +def test_resume_skips_completed_attempts(tmp_path, sample_file, fake_remote): + run_swebench_attempts(**run_kwargs(tmp_path, sample_file, fake_remote)) + meta_path = tmp_path / "attempts" / "t" / "acme__widget-1" / "attempt-1" / "meta.json" + before = meta_path.stat().st_mtime_ns + # an agent that would fail if actually executed + files = run_swebench_attempts( + **run_kwargs(tmp_path, sample_file, fake_remote, resume=True, + agent_cmd="false && this-should-never-run") + ) + assert meta_path.stat().st_mtime_ns == before # not re-run + row = json.loads(files[1].read_text().splitlines()[0]) + assert row["instance_id"] == "acme__widget-1" + + +def test_invalid_attempts_rejected(tmp_path, sample_file, fake_remote): + with pytest.raises(ValueError): + run_swebench_attempts(**run_kwargs(tmp_path, sample_file, fake_remote, attempts=0)) + + +# ---------------------------------------------------------------- leakage + +def test_no_gold_data_anywhere_in_run_artifacts(tmp_path, sample_file, fake_remote): + run_swebench_attempts( + **run_kwargs(tmp_path, sample_file, fake_remote, attempts=2) + ) + artifact_roots = [tmp_path / "attempts", tmp_path / "predictions"] + checked = 0 + for root in artifact_roots: + for path in root.rglob("*"): + if path.is_file(): + content = path.read_text(encoding="utf-8", errors="replace") + assert GOLD_SENTINEL not in content, path + assert HIDDEN_TEST_SENTINEL not in content, path + assert "hidden_reference" not in content, path + checked += 1 + assert checked >= 7 # prompts, logs, patches, metas, predictions