diff --git a/benchmarks/swebench/README.md b/benchmarks/swebench/README.md index da0b2600c..75595c142 100644 --- a/benchmarks/swebench/README.md +++ b/benchmarks/swebench/README.md @@ -239,9 +239,26 @@ uv run swebench-eval output.jsonl \ uv run swebench-eval output.jsonl --skip-evaluation ``` +**Local Apptainer evaluation:** + +```bash +uv run swebench-eval output.jsonl \ + --run-id my_eval \ + --apptainer \ + --apptainer-sandbox-root ~/.cache/openhands/swebench-apptainer +``` + +The Apptainer evaluator pulls the official SWE-bench instance images, converts +them to reusable writable sandboxes, applies each model patch, runs the +SWE-bench eval script, and grades the resulting test log locally. This is useful +on hosts where Docker is unavailable and Modal is not configured. Apptainer +evaluation currently runs sequentially; `--workers` is accepted for CLI +compatibility but ignored. + The evaluation script will: 1. Convert OpenHands output format to SWE-Bench prediction format -2. Run the official SWE-Bench evaluation harness (unless `--skip-evaluation` is used) +2. Run the official SWE-Bench evaluation harness, or local Apptainer evaluation + when `--apptainer` is used, unless `--skip-evaluation` is set 3. Report pass/fail results for each instance ## References diff --git a/benchmarks/swebench/apptainer_eval.py b/benchmarks/swebench/apptainer_eval.py new file mode 100644 index 000000000..280b1fa79 --- /dev/null +++ b/benchmarks/swebench/apptainer_eval.py @@ -0,0 +1,357 @@ +"""Apptainer-based SWE-bench evaluation.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from collections.abc import Iterable +from pathlib import Path +from typing import Any, cast + +from openhands.sdk import get_logger + + +logger = get_logger(__name__) + + +DEFAULT_APPTAINER_SANDBOX_ROOT = ( + Path.home() / ".cache" / "openhands" / "swebench-apptainer" +) +# Allow Apptainer teardown and artifact writes to finish after SWE-bench timeout. +APPTAINER_EXEC_TIMEOUT_BUFFER_SECONDS = 300 + + +def _run( + cmd: list[str], + log_path: Path | None = None, + timeout: int | None = None, + apptainer_cache: Path | None = None, +) -> subprocess.CompletedProcess[str]: + env = os.environ.copy() + if apptainer_cache is not None: + env["APPTAINER_CACHEDIR"] = str(apptainer_cache) + + proc = subprocess.run( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + timeout=timeout, + env=env, + ) + if log_path is not None: + log_path.write_text(proc.stdout) + return proc + + +def load_predictions(predictions_file: Path) -> dict[str, dict[str, Any]]: + """Load SWE-bench predictions by instance id.""" + from swebench.harness.constants import KEY_INSTANCE_ID + + rows = [] + with predictions_file.open() as f: + for line in f: + if line.strip(): + rows.append(json.loads(line)) + return {row[KEY_INSTANCE_ID]: row for row in rows} + + +def image_uri(instance: dict[str, Any]) -> str: + """Return the official SWE-bench instance image URI.""" + from swebench.harness.constants import KEY_INSTANCE_ID + from swebench.harness.test_spec.test_spec import make_test_spec + + image_template = os.getenv("OPENHANDS_SWEBENCH_IMAGE_TEMPLATE") + if image_template: + instance_id = instance[KEY_INSTANCE_ID] + repo, name = instance_id.split("__") + return "docker://" + image_template.format( + instance_id=instance_id, + repo=repo, + name=name, + arch="x86_64", + ) + + spec = make_test_spec(cast(Any, instance), namespace="swebench") + return "docker://" + spec.instance_image_key + + +def ensure_sandbox( + instance: dict[str, Any], + score_dir: Path, + sandbox_root: Path, + apptainer_cache: Path | None, +) -> Path: + """Build or reuse an Apptainer sandbox for a SWE-bench instance.""" + from swebench.harness.constants import KEY_INSTANCE_ID + + instance_id = instance[KEY_INSTANCE_ID] + sandbox = sandbox_root / instance_id + if sandbox.exists(): + return sandbox + + if shutil.which("apptainer") is None: + raise RuntimeError("Apptainer is not available on PATH") + + tmp = sandbox.with_suffix(".tmp") + if tmp.exists(): + shutil.rmtree(tmp) + tmp.parent.mkdir(parents=True, exist_ok=True) + + logger.info("Building Apptainer sandbox for %s", instance_id) + proc = _run( + ["apptainer", "build", "--sandbox", str(tmp), image_uri(instance)], + score_dir / f"{instance_id}.build.log", + timeout=3600, + apptainer_cache=apptainer_cache, + ) + if proc.returncode != 0: + raise RuntimeError( + f"apptainer build failed for {instance_id}; see " + f"{score_dir / f'{instance_id}.build.log'}" + ) + + tmp.rename(sandbox) + return sandbox + + +def apptainer_base_cmd(sandbox: Path, work_dir: Path) -> list[str]: + """Return the base Apptainer exec command for a sandboxed test run.""" + return [ + "apptainer", + "exec", + "--no-mount", + "hostfs", + # Some clusters bind host /opt into every Apptainer container. Bind the + # sandbox's /opt back over /opt so /opt/miniconda3 from SWE-bench images + # remains visible. + "--bind", + f"{sandbox / 'opt'}:/opt", + "--bind", + f"{work_dir}:/mnt", + "--writable", + str(sandbox), + "bash", + "-lc", + ] + + +def verify_sandbox( + instance: dict[str, Any], + sandbox: Path, + work_dir: Path, + apptainer_cache: Path | None, +) -> None: + """Check that the SWE-bench conda environment is available in the sandbox.""" + from swebench.harness.constants import KEY_INSTANCE_ID + + cmd = apptainer_base_cmd(sandbox, work_dir) + [ + "source /opt/miniconda3/bin/activate && " + "conda activate testbed && " + "cd /testbed && " + "python --version && " + "command -v python && " + "git rev-parse HEAD" + ] + proc = _run( + cmd, + work_dir / "sandbox_verify.log", + timeout=120, + apptainer_cache=apptainer_cache, + ) + if proc.returncode != 0: + raise RuntimeError( + f"sandbox verification failed for {instance[KEY_INSTANCE_ID]}; " + f"see {work_dir / 'sandbox_verify.log'}" + ) + + +def score_shell( + timeout_seconds: int, + apply_patch_fail: str, + tests_timeout: str, +) -> str: + """Return the shell script that applies a patch and runs SWE-bench tests.""" + return f""" +set -uo pipefail +cd /testbed +git config --global --add safe.directory /testbed || true +git reset --hard HEAD +git clean -fd +applied=0 +apply_output=/mnt/apply_patch.log +: > "$apply_output" +for cmd in "git apply --verbose /mnt/patch.diff" "git apply --verbose --reject /mnt/patch.diff" "patch --batch --fuzz=5 -p1 -i /mnt/patch.diff"; do + echo "$cmd" >> "$apply_output" + if bash -lc "$cmd" >> "$apply_output" 2>&1; then + applied=1 + break + fi +done +if [ "$applied" != "1" ]; then + {{ + echo "{apply_patch_fail}" + cat "$apply_output" + }} > /mnt/test_output.txt + exit 0 +fi +git -c core.fileMode=false diff > /mnt/git_diff_before.diff 2>&1 || true +timeout {timeout_seconds} /bin/bash /mnt/eval.sh > /mnt/test_output.txt 2>&1 +status=$? +git -c core.fileMode=false diff > /mnt/git_diff_after.diff 2>&1 || true +if [ "$status" = "124" ]; then + echo "{tests_timeout}" >> /mnt/test_output.txt +fi +exit 0 +""" + + +def score_instance( + instance: dict[str, Any], + prediction: dict[str, Any], + score_dir: Path, + sandbox_root: Path, + timeout_seconds: int, + apptainer_cache: Path | None, +) -> dict[str, Any]: + """Apply a prediction and score one SWE-bench instance in Apptainer.""" + from swebench.harness.constants import ( + APPLY_PATCH_FAIL, + KEY_INSTANCE_ID, + KEY_PREDICTION, + TESTS_TIMEOUT, + ) + from swebench.harness.grading import get_eval_report + from swebench.harness.test_spec.test_spec import make_test_spec + + instance_id = instance[KEY_INSTANCE_ID] + work_dir = score_dir / instance_id + work_dir.mkdir(parents=True, exist_ok=True) + + report_path = work_dir / "report.json" + if report_path.exists(): + return json.loads(report_path.read_text()) + + patch = prediction.get(KEY_PREDICTION) or "" + if not patch.strip(): + report = { + instance_id: { + "patch_is_None": prediction.get(KEY_PREDICTION) is None, + "patch_exists": False, + "patch_successfully_applied": False, + "resolved": False, + "skipped_empty_patch": True, + } + } + report_path.write_text(json.dumps(report, indent=2)) + return report + + test_spec = make_test_spec(cast(Any, instance)) + sandbox = ensure_sandbox(instance, score_dir, sandbox_root, apptainer_cache) + (work_dir / "patch.diff").write_text(patch) + (work_dir / "eval.sh").write_text(test_spec.eval_script) + verify_sandbox(instance, sandbox, work_dir, apptainer_cache) + + shell = score_shell( + timeout_seconds=timeout_seconds, + apply_patch_fail=APPLY_PATCH_FAIL, + tests_timeout=TESTS_TIMEOUT, + ) + proc = _run( + apptainer_base_cmd(sandbox, work_dir) + [shell], + work_dir / "apptainer_exec.log", + timeout=timeout_seconds + APPTAINER_EXEC_TIMEOUT_BUFFER_SECONDS, + apptainer_cache=apptainer_cache, + ) + if proc.returncode != 0: + raise RuntimeError( + f"apptainer exec failed for {instance_id}; " + f"see {work_dir / 'apptainer_exec.log'}" + ) + + report = get_eval_report( + test_spec=test_spec, + prediction=prediction, + test_log_path=str(work_dir / "test_output.txt"), + include_tests_status=True, + ) + report_path.write_text(json.dumps(report, indent=2)) + return report + + +def run_swebench_evaluation_apptainer( + predictions_file: Path, + report_file: Path, + dataset: str, + split: str, + timeout_seconds: int, + workers: int, + score_dir: Path | None = None, + sandbox_root: Path = DEFAULT_APPTAINER_SANDBOX_ROOT, + apptainer_cache: Path | None = None, +) -> None: + """Run SWE-bench evaluation locally with Apptainer.""" + from datasets import load_dataset + from swebench.harness.constants import KEY_INSTANCE_ID + + if workers != 1: + logger.warning( + "Apptainer evaluation currently runs sequentially; ignoring workers=%s", + workers, + ) + + predictions = load_predictions(predictions_file) + wanted = set(predictions) + if score_dir is None: + score_dir = predictions_file.parent / "apptainer_eval" + score_dir.mkdir(parents=True, exist_ok=True) + sandbox_root.mkdir(parents=True, exist_ok=True) + + instances = [ + instance + for instance in cast( + Iterable[dict[str, Any]], load_dataset(dataset, split=split) + ) + if instance[KEY_INSTANCE_ID] in wanted + ] + + reports: dict[str, Any] = {} + for instance in instances: + instance_id = instance[KEY_INSTANCE_ID] + logger.info("Scoring %s with Apptainer", instance_id) + report = score_instance( + instance=instance, + prediction=predictions[instance_id], + score_dir=score_dir, + sandbox_root=sandbox_root, + timeout_seconds=timeout_seconds, + apptainer_cache=apptainer_cache, + ) + reports.update(report) + logger.info( + "%s: resolved=%s patch_applied=%s", + instance_id, + reports[instance_id].get("resolved"), + reports[instance_id].get("patch_successfully_applied"), + ) + + resolved_ids = sorted( + instance_id for instance_id, report in reports.items() if report.get("resolved") + ) + unresolved_ids = sorted(set(reports) - set(resolved_ids)) + report_data = { + "dataset": dataset, + "split": split, + "predictions": str(predictions_file), + "score_dir": str(score_dir), + "total": len(reports), + "resolved": len(resolved_ids), + "unresolved": len(unresolved_ids), + "resolved_ids": resolved_ids, + "unresolved_ids": unresolved_ids, + "reports": reports, + } + report_file.write_text(json.dumps(report_data, indent=2)) + logger.info("Wrote Apptainer SWE-bench report to %s", report_file) diff --git a/benchmarks/swebench/eval_infer.py b/benchmarks/swebench/eval_infer.py index f24aae201..d1ad104ee 100644 --- a/benchmarks/swebench/eval_infer.py +++ b/benchmarks/swebench/eval_infer.py @@ -17,6 +17,10 @@ from pathlib import Path from benchmarks.swebench import constants +from benchmarks.swebench.apptainer_eval import ( + DEFAULT_APPTAINER_SANDBOX_ROOT, + run_swebench_evaluation_apptainer, +) from benchmarks.swebench.config import EVAL_DEFAULTS from benchmarks.utils.constants import MODEL_NAME_OR_PATH from benchmarks.utils.laminar import LaminarService @@ -257,6 +261,37 @@ def main() -> None: help="Timeout in seconds for evaluation", ) + parser.add_argument( + "--apptainer", + action="store_true", + help="Use local Apptainer sandboxes for SWE-bench evaluation", + ) + + parser.add_argument( + "--apptainer-score-dir", + type=Path, + help=( + "Directory for Apptainer scoring logs and per-instance artifacts " + "(default: /apptainer_eval)" + ), + ) + + parser.add_argument( + "--apptainer-sandbox-root", + type=Path, + default=DEFAULT_APPTAINER_SANDBOX_ROOT, + help=( + "Directory for reusable Apptainer sandboxes " + f"(default: {DEFAULT_APPTAINER_SANDBOX_ROOT})" + ), + ) + + parser.add_argument( + "--apptainer-cache", + type=Path, + help="Optional APPTAINER_CACHEDIR value for image pulls/builds", + ) + # Apply EVAL_DEFAULTS from config (for dataset, split, workers, modal, timeout) parser.set_defaults(**EVAL_DEFAULTS) @@ -288,25 +323,39 @@ def main() -> None: convert_to_swebench_format(str(input_file), str(output_file)) if not args.skip_evaluation: - # Run evaluation - run_swebench_evaluation( - str(output_file), - args.run_id, - args.dataset, - args.workers, - split=args.split, - modal=args.modal, - timeout=args.timeout, - ) - - # Move report file to input file directory with .report.json extension - # SWE-Bench creates: {MODEL_NAME_OR_PATH}.{run_id}.json - report_filename = f"{MODEL_NAME_OR_PATH}.{args.run_id}.json" - report_path = output_file.parent / report_filename dest_report_path = input_file.with_suffix(".report.json") - shutil.move(str(report_path), str(dest_report_path)) - logger.info(f"Moved report file to: {dest_report_path}") + if args.apptainer: + run_swebench_evaluation_apptainer( + predictions_file=output_file, + report_file=dest_report_path, + dataset=args.dataset, + split=args.split, + timeout_seconds=args.timeout, + workers=args.workers, + score_dir=args.apptainer_score_dir, + sandbox_root=args.apptainer_sandbox_root, + apptainer_cache=args.apptainer_cache, + ) + else: + # Run evaluation + run_swebench_evaluation( + str(output_file), + args.run_id, + args.dataset, + args.workers, + split=args.split, + modal=args.modal, + timeout=args.timeout, + ) + + # Move report file to input file directory with .report.json extension + # SWE-Bench creates: {MODEL_NAME_OR_PATH}.{run_id}.json + report_filename = f"{MODEL_NAME_OR_PATH}.{args.run_id}.json" + report_path = output_file.parent / report_filename + + shutil.move(str(report_path), str(dest_report_path)) + logger.info(f"Moved report file to: {dest_report_path}") # Update Laminar datapoints with evaluation scores LaminarService.get().update_evaluation_scores( diff --git a/tests/test_swebench_eval_infer.py b/tests/test_swebench_eval_infer.py index 594b9a8ab..29a1df3c3 100644 --- a/tests/test_swebench_eval_infer.py +++ b/tests/test_swebench_eval_infer.py @@ -2,7 +2,17 @@ import json import tempfile +from types import SimpleNamespace +import pytest +from swebench.harness.constants import ( + APPLY_PATCH_FAIL, + KEY_INSTANCE_ID, + KEY_PREDICTION, + TESTS_TIMEOUT, +) + +from benchmarks.swebench import apptainer_eval from benchmarks.swebench.eval_infer import convert_to_swebench_format from benchmarks.utils.constants import MODEL_NAME_OR_PATH @@ -60,3 +70,166 @@ def test_model_name_or_path_uses_constant(self): result = json.loads(f.readline()) assert result["model_name_or_path"] == MODEL_NAME_OR_PATH + + +class TestApptainerEvaluation: + """Tests for Apptainer SWE-bench evaluation helpers.""" + + def test_image_uri_uses_swebench_template(self, monkeypatch): + """Apptainer scoring can use non-Docker-Hub benchmark image mirrors.""" + monkeypatch.setenv( + "OPENHANDS_SWEBENCH_IMAGE_TEMPLATE", + "ghcr.io/epoch-research/swe-bench.eval.{arch}.{instance_id}:latest", + ) + + assert ( + apptainer_eval.image_uri({KEY_INSTANCE_ID: "astropy__astropy-12907"}) + == "docker://ghcr.io/epoch-research/" + "swe-bench.eval.x86_64.astropy__astropy-12907:latest" + ) + + def test_ensure_sandbox_reports_missing_apptainer(self, monkeypatch, tmp_path): + """Missing Apptainer should produce an actionable setup error.""" + monkeypatch.setattr(apptainer_eval.shutil, "which", lambda command: None) + + with pytest.raises(RuntimeError, match="Apptainer is not available"): + apptainer_eval.ensure_sandbox( + instance={KEY_INSTANCE_ID: "django__django-12345"}, + score_dir=tmp_path / "score", + sandbox_root=tmp_path / "sandboxes", + apptainer_cache=None, + ) + + def test_ensure_sandbox_reports_image_build_failure(self, monkeypatch, tmp_path): + """Image pull/build failures should include the per-instance build log path.""" + score_dir = tmp_path / "score" + score_dir.mkdir() + monkeypatch.setattr( + apptainer_eval.shutil, + "which", + lambda command: "/usr/bin/apptainer", + ) + monkeypatch.setattr( + apptainer_eval, + "image_uri", + lambda instance: "docker://swebench/example:latest", + ) + + def fake_run(cmd, log_path, timeout, apptainer_cache): + log_path.write_text("pull failed") + return SimpleNamespace(returncode=1) + + monkeypatch.setattr(apptainer_eval, "_run", fake_run) + + with pytest.raises(RuntimeError, match="apptainer build failed"): + apptainer_eval.ensure_sandbox( + instance={KEY_INSTANCE_ID: "django__django-12345"}, + score_dir=score_dir, + sandbox_root=tmp_path / "sandboxes", + apptainer_cache=None, + ) + assert (score_dir / "django__django-12345.build.log").read_text() == ( + "pull failed" + ) + + def test_score_shell_uses_swebench_sentinel_values(self): + """Patch and timeout failures must emit SWE-bench grading constants.""" + shell = apptainer_eval.score_shell( + timeout_seconds=123, + apply_patch_fail=APPLY_PATCH_FAIL, + tests_timeout=TESTS_TIMEOUT, + ) + + assert f'echo "{APPLY_PATCH_FAIL}"' in shell + assert f'echo "{TESTS_TIMEOUT}"' in shell + assert "timeout 123 /bin/bash /mnt/eval.sh" in shell + assert "{APPLY_PATCH_FAIL}" not in shell + assert "{TESTS_TIMEOUT}" not in shell + + def test_score_instance_empty_patch_writes_unresolved_report(self, tmp_path): + """Empty model patches should be marked unresolved without Apptainer.""" + instance = {KEY_INSTANCE_ID: "django__django-12345"} + prediction = { + KEY_INSTANCE_ID: "django__django-12345", + KEY_PREDICTION: "", + } + + report = apptainer_eval.score_instance( + instance=instance, + prediction=prediction, + score_dir=tmp_path / "score", + sandbox_root=tmp_path / "sandboxes", + timeout_seconds=10, + apptainer_cache=None, + ) + + instance_report = report["django__django-12345"] + assert instance_report["resolved"] is False + assert instance_report["patch_exists"] is False + assert instance_report["skipped_empty_patch"] is True + assert (tmp_path / "score" / "django__django-12345" / "report.json").exists() + + def test_run_swebench_evaluation_apptainer_writes_summary_report( + self, tmp_path, monkeypatch + ): + """The Apptainer entry point should write resolved_ids for callers.""" + predictions = { + "django__django-1": { + KEY_INSTANCE_ID: "django__django-1", + KEY_PREDICTION: "diff --git a/a.py b/a.py", + }, + "django__django-2": { + KEY_INSTANCE_ID: "django__django-2", + KEY_PREDICTION: "", + }, + } + + monkeypatch.setattr( + apptainer_eval, + "load_predictions", + lambda predictions_file: predictions, + ) + monkeypatch.setattr( + "datasets.load_dataset", + lambda dataset, split: [ + {KEY_INSTANCE_ID: "django__django-1"}, + {KEY_INSTANCE_ID: "django__django-2"}, + ], + ) + + def fake_score_instance( + instance, + prediction, + score_dir, + sandbox_root, + timeout_seconds, + apptainer_cache, + ): + instance_id = instance[KEY_INSTANCE_ID] + return { + instance_id: { + "resolved": instance_id == "django__django-1", + "patch_successfully_applied": bool(prediction.get(KEY_PREDICTION)), + } + } + + monkeypatch.setattr(apptainer_eval, "score_instance", fake_score_instance) + + report_file = tmp_path / "output.report.json" + apptainer_eval.run_swebench_evaluation_apptainer( + predictions_file=tmp_path / "output.swebench.jsonl", + report_file=report_file, + dataset="princeton-nlp/SWE-bench_Verified", + split="test", + timeout_seconds=10, + workers=2, + score_dir=tmp_path / "score", + sandbox_root=tmp_path / "sandboxes", + apptainer_cache=None, + ) + + report = json.loads(report_file.read_text()) + assert report["total"] == 2 + assert report["resolved"] == 1 + assert report["resolved_ids"] == ["django__django-1"] + assert report["unresolved_ids"] == ["django__django-2"]