benchmarks/scripts/13_swe_native.py

#!/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"<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")      # the version current when the official images were built
    {"list": cmd_list, "gold": cmd_gold, "goldall": cmd_goldall, "agent": cmd_agent}[a.cmd](a)