|
|
|
|
| """
|
| 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"}
|
|
|
|
|
| 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")
|
|
|
| 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 {})})
|
| 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"<uploaded_files>\n/testbed\n</uploaded_files>\nI've uploaded a python code repository in the directory /testbed. Consider the following issue "
|
| f"description:\n\n<issue>\n{inst['problem_statement']}\n</issue>\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")
|
| {"list": cmd_list, "gold": cmd_gold, "goldall": cmd_goldall, "agent": cmd_agent}[a.cmd](a)
|
|
|
