#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
13_swe_native.py - SWE-bench Verified (Mini) WITHOUT Docker or conda, on a plain Linux VM, driving the real Claude Code CLI.
It reuses swebench 4.0.4's own per-repo specs, setup/eval scripts and log parsers (pip install --no-deps --target /content/sb4 swebench==4.0.4),
translating the conda lines to `uv venv` environments, one /testbed checkout at a time.
python 13_swe_native.py list the instances of SWE-bench Verified Mini with their repo / version
python 13_swe_native.py gold INSTANCE_ID build the environment and check that the GOLD patch resolves the instance
python 13_swe_native.py agent --ids a,b,c --out results_swe_k448.json --endpoint http://127.0.0.1:18080 --model k448 --user agent
[--max-turns 60] [--timeout 2700] Claude Code solves each instance, the patch is graded by the official parsers
A resolved instance = every FAIL_TO_PASS test passes and no PASS_TO_PASS test breaks (the official SWE-bench definition). THE MODEL'S COMMANDS
AND THE REPOSITORIES' TESTS ARE EXECUTED: use a disposable VM; the agent runs as a separate unix user.
"""
import argparse
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import time
from pathlib import Path
sys.path.insert(0, "/content/sb4")
ENVS, TESTBED, WORK = Path("/content/sb_envs"), Path("/testbed"), Path("/content/sb_work")
CONSTRAINTS = Path("/content/sb_constraints.txt")
PY_MAP = {"3.5": "3.8", "3.6": "3.8", "3.7": "3.8"} # uv offers CPython 3.8+ ; these old pins run fine on 3.8
def rows():
from datasets import load_dataset
return [dict(r) for r in load_dataset("MariusHobbhahn/swe-bench-verified-mini", split="test")]
def get(iid):
for r in rows():
if r["instance_id"] == iid:
return r
sys.exit(f"unknown instance {iid}")
def test_spec(inst):
from swebench.harness.test_spec.test_spec import make_test_spec
return make_test_spec(inst)
def env_dir(inst):
return ENVS / f"{inst['repo'].replace('/', '__')}__{inst['version']}__v4"
def adapt(script, venv, py):
"""conda lines -> uv venv lines (the scripts are plain bash otherwise)."""
s = script.replace("source /opt/miniconda3/bin/activate\n", "")
s = re.sub(r"conda create -n testbed python=([\d.]+)\s+-y", lambda m: f"uv venv --seed --python {PY_MAP.get(m.group(1), m.group(1))} {venv} && {venv}/bin/python -m pip install 'setuptools<70'", s)
s = s.replace("conda activate testbed", f"source {venv}/bin/activate")
s = s.replace('echo "Current environment: $CONDA_DEFAULT_ENV"', "true")
# sphinx: tox would build its own interpreter without pytest here; run what its tox.ini runs, directly in the venv
s = re.sub(r"tox --current-env -e\S+ -v -- ", "PYTHONWARNINGS=all python -X dev -m pytest --durations 25 -rA ", s)
return s
def bash(script, log, timeout, env=None):
Path(log).parent.mkdir(parents=True, exist_ok=True)
f = Path(log).with_suffix(".sh")
f.write_text(script, encoding="utf-8")
try:
r = subprocess.run(["bash", str(f)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=timeout, errors="replace",
env={**os.environ, "PIP_CONSTRAINT": str(CONSTRAINTS), **(env or {})}) # one stream, in order: the grader reads the output between the START/END markers
out = r.stdout
code = r.returncode
except subprocess.TimeoutExpired as e:
out = ((e.stdout or b"").decode("utf-8", "replace") if isinstance(e.stdout, bytes) else (e.stdout or "")) + "\n[timeout]"
code = 124
Path(log).write_text(out, encoding="utf-8")
return code, out
def prepare(inst):
"""Environment (cached per repo+version) and a fresh /testbed at the base commit with the repo installed."""
ts = test_spec(inst)
venv = env_dir(inst)
py = re.search(r"python=([\d.]+)", ts.setup_env_script)
py = PY_MAP.get(py.group(1), py.group(1)) if py else "3.9"
if not (venv / ".done").exists():
shutil.rmtree(venv, ignore_errors=True)
ENVS.mkdir(parents=True, exist_ok=True)
code, out = bash(adapt(ts.setup_env_script, venv, py), WORK / "logs" / f"{venv.name}_env.log", 3000)
if code != 0 or not (venv / "bin" / "python").exists():
raise RuntimeError(f"environment build failed (exit {code}): {out[-600:]}")
(venv / ".done").write_text("ok")
shutil.rmtree(TESTBED, ignore_errors=True)
code, out = bash(adapt(ts.install_repo_script, venv, py), WORK / "logs" / f"{inst['instance_id']}_repo.log", 2400)
if code != 0 or not (TESTBED / ".git").exists():
raise RuntimeError(f"repo install failed (exit {code}): {out[-600:]}")
return ts, venv
def alias_docstring_lines(out):
"""Python 3.8 prints `test_x (mod.Class)` and the docstring line separately; add the one-line form that the dataset's test names use."""
res, prev = [], None
for line in out.split("\n"):
res.append(line)
t = line.strip()
m = re.match(r"^(test\S* \([\w.]+\))$", t)
if m:
prev = m.group(1)
elif prev and " ... " in t:
res.append(f"{prev} ... {t.rsplit(' ... ', 1)[1]}")
prev = None
else:
prev = None
return "\n".join(res)
def evaluate(inst, ts, venv, patch):
"""Fresh checkout, apply the model patch, run the official eval script, grade with the official parser. -> (resolved, note)"""
from swebench.harness.grading import get_eval_report
py = re.search(r"python=([\d.]+)", ts.setup_env_script)
py = PY_MAP.get(py.group(1), py.group(1)) if py else "3.9"
subprocess.run("git reset --hard -q && git clean -fdq", shell=True, cwd=TESTBED)
if not patch.strip():
return False, "empty patch"
pf = WORK / f"{inst['instance_id']}.patch"
pf.write_text(patch if patch.endswith("\n") else patch + "\n", encoding="utf-8")
r = subprocess.run(["git", "apply", "-v", str(pf)], cwd=TESTBED, capture_output=True, text=True)
if r.returncode != 0:
return False, "patch does not apply: " + (r.stdout + r.stderr)[-200:]
code, out = bash(adapt(ts.eval_script, venv, py), WORK / "logs" / f"{inst['instance_id']}_eval.log", 2700)
logp = WORK / "logs" / f"{inst['instance_id']}_report.log"
logp.write_text(">>>>> Applied Patch\n" + alias_docstring_lines(out), encoding="utf-8")
try:
rep = get_eval_report(test_spec=ts, prediction={"instance_id": inst["instance_id"], "model_name_or_path": "k448", "model_patch": patch},
test_log_path=str(logp), include_tests_status=True)[inst["instance_id"]]
except Exception as e:
return False, f"grading error: {e!r}"
ts_ = rep.get("tests_status", {})
f2p, p2p = ts_.get("FAIL_TO_PASS", {}), ts_.get("PASS_TO_PASS", {})
note = f"F2P {len(f2p.get('success', []))}/{len(f2p.get('success', [])) + len(f2p.get('failure', []))}, P2P ok {len(p2p.get('success', []))} broken {len(p2p.get('failure', []))}"
return bool(rep.get("resolved")), note
def cmd_goldall(a):
"""Gold-patch check of every instance: the ones that resolve are the runnable benchmark."""
out = Path(a.out)
res = json.loads(out.read_text(encoding="utf-8")) if out.exists() else {}
for inst in rows():
iid = inst["instance_id"]
if iid in res:
continue
t0 = time.time()
try:
ts, venv = prepare(inst)
ok, note = evaluate(inst, ts, venv, inst["patch"])
except Exception as e:
ok, note = False, f"error: {e!r}"[:300]
res[iid] = {"ok": ok, "note": note, "secs": round(time.time() - t0)}
out.write_text(json.dumps(res, indent=1), encoding="utf-8")
print(f"{iid:36s} gold_resolved={ok} {note[:100]} ({time.time() - t0:.0f} s)", flush=True)
print(f"runnable {sum(v['ok'] for v in res.values())}/{len(res)}")
def cmd_list(a):
for r in rows():
print(f"{r['instance_id']:36s} {r['repo']:20s} {r['version']:5s} F2P={len(json.loads(r['FAIL_TO_PASS']) if isinstance(r['FAIL_TO_PASS'], str) else r['FAIL_TO_PASS'])}")
def cmd_gold(a):
inst = get(a.instance)
t0 = time.time()
ts, venv = prepare(inst)
print(f"prepared in {time.time() - t0:.0f} s", flush=True)
ok, note = evaluate(inst, ts, venv, inst["patch"])
print(f"GOLD {inst['instance_id']}: resolved={ok} {note} ({time.time() - t0:.0f} s)")
return 0 if ok else 1
def agent_prompt(inst):
return (f"\n/testbed\n\nI've uploaded a python code repository in the directory /testbed. Consider the following issue "
f"description:\n\n\n{inst['problem_statement']}\n\n\nCan you help me implement the necessary changes to the repository so that the "
"requirements specified in the issue are met? I've already taken care of all changes to any of the test files described in the issue, so you do NOT "
"need to modify the testing logic or any of the tests in any way. Make the minimal changes to non-test files in /testbed so that the issue is resolved. "
"You can run the existing tests to check your work. When you are done, stop.")
def parse_stream(text):
calls = errs = 0
last, final = "", {}
for line in text.splitlines():
if not line.startswith("{"):
continue
try:
ev = json.loads(line)
except ValueError:
continue
t = ev.get("type")
if t == "assistant":
for b in (ev.get("message", {}).get("content") or []):
if b.get("type") == "tool_use":
calls += 1
elif b.get("type") == "text" and b.get("text", "").strip():
last = b["text"].strip()
elif t == "user":
c = ev.get("message", {}).get("content")
if isinstance(c, list):
errs += sum(1 for b in c if isinstance(b, dict) and b.get("type") == "tool_result" and b.get("is_error"))
elif t == "result":
final = ev
u = final.get("usage") or {}
return {"turns": final.get("num_turns", 0), "tool_calls": calls, "tool_errors": errs, "out_tok": u.get("output_tokens", 0),
"in_tok": u.get("input_tokens", 0) + u.get("cache_read_input_tokens", 0) + u.get("cache_creation_input_tokens", 0),
"stop": final.get("subtype") or ("no_result" if not final else "?"), "last": last[-300:]}
def cmd_agent(a):
ids = a.ids.split(",") if a.ids else [r["instance_id"] for r in rows()]
out = Path(a.out)
res = json.loads(out.read_text(encoding="utf-8")).get("results", {}) if out.exists() else {}
logs = Path(str(out.with_suffix("")) + "_logs")
logs.mkdir(parents=True, exist_ok=True)
home = WORK / "_home"
home.mkdir(parents=True, exist_ok=True)
t0 = time.time()
for k, iid in enumerate(ids, 1):
if iid in res:
continue
inst = get(iid)
s0 = time.time()
info = {"repo": inst["repo"], "version": inst["version"]}
try:
ts, venv = prepare(inst)
subprocess.run(["chown", "-R", f"{a.user}:{a.user}", str(TESTBED), str(home)], check=True)
env = {"PATH": f"{venv}/bin:" + os.environ.get("PATH", ""), "VIRTUAL_ENV": str(venv), "HOME": str(home), "LANG": "C.UTF-8",
"ANTHROPIC_BASE_URL": a.endpoint, "ANTHROPIC_AUTH_TOKEN": "local", "ANTHROPIC_API_KEY": "local", "ANTHROPIC_MODEL": a.model,
"ANTHROPIC_SMALL_FAST_MODEL": a.model, "ANTHROPIC_DEFAULT_HAIKU_MODEL": a.model, "ANTHROPIC_DEFAULT_SONNET_MODEL": a.model,
"ANTHROPIC_DEFAULT_OPUS_MODEL": a.model, "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", "DISABLE_TELEMETRY": "1",
"DISABLE_AUTOUPDATER": "1", "DISABLE_ERROR_REPORTING": "1", "IS_SANDBOX": "1"}
cmd = ["runuser", "-u", a.user, "--", "env"] + [f"{kk}={vv}" for kk, vv in env.items()] + shlex.split(a.claude) + [
"-p", agent_prompt(inst), "--output-format", "stream-json", "--verbose", "--dangerously-skip-permissions",
"--max-turns", str(a.max_turns), "--model", a.model]
try:
r = subprocess.run(cmd, cwd=TESTBED, capture_output=True, text=True, timeout=a.timeout, errors="replace", stdin=subprocess.DEVNULL)
text, err, timed_out = r.stdout, r.stderr, False
except subprocess.TimeoutExpired as e:
text = ((e.stdout or b"").decode("utf-8", "replace") if isinstance(e.stdout, bytes) else (e.stdout or ""))
err, timed_out = "timeout", True
(logs / f"{iid}.jsonl").write_text(text + ("\n[stderr]\n" + err[-2000:] if err else ""), encoding="utf-8")
info.update(parse_stream(text))
info["timed_out"] = timed_out
subprocess.run("git add -A && git diff --cached HEAD > /tmp/model.patch", shell=True, cwd=TESTBED)
patch = Path("/tmp/model.patch").read_text(encoding="utf-8", errors="replace")
info["patch_lines"] = patch.count("\n")
info["pass"], info["note"] = evaluate(inst, ts, venv, patch)
except Exception as e:
info.update({"pass": False, "note": f"harness error: {e!r}"[:300]})
info["secs"] = round(time.time() - s0)
res[iid] = info
out.write_text(json.dumps({"kind": "swe", "name": a.model, "seconds": round(time.time() - t0), "results": res}, indent=1), encoding="utf-8")
print(f"{k:3d}/{len(ids)} {iid:34s} {'RESOLVED' if info['pass'] else 'no':8s} turns={info.get('turns', 0):3d} calls={info.get('tool_calls', 0):3d} "
f"tool_errors={info.get('tool_errors', 0):2d} patch={info.get('patch_lines', 0):3d} lines {info['secs']:5d} s {info.get('note', '')[:80]}", flush=True)
n = len(res)
print(f"\nresolved {sum(v['pass'] for v in res.values())}/{n}")
if __name__ == "__main__":
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
sub.add_parser("list")
ga = sub.add_parser("goldall")
ga.add_argument("--out", default="/content/gold_all.json")
g = sub.add_parser("gold")
g.add_argument("instance")
ag = sub.add_parser("agent")
ag.add_argument("--ids")
ag.add_argument("--out", default="results_swe.json")
ag.add_argument("--endpoint", default="http://127.0.0.1:18080")
ag.add_argument("--model", default="k448")
ag.add_argument("--user", default="agent")
ag.add_argument("--claude", default="claude")
ag.add_argument("--max-turns", type=int, default=60)
ag.add_argument("--timeout", type=int, default=2700)
a = ap.parse_args()
WORK.mkdir(parents=True, exist_ok=True)
CONSTRAINTS.write_text("Pygments==2.18.0\n", encoding="utf-8") # the version current when the official images were built
{"list": cmd_list, "gold": cmd_gold, "goldall": cmd_goldall, "agent": cmd_agent}[a.cmd](a)