小模型写、大模型审：SWE-bench Pro Hard 与 DeepSWE 上的结对实验

#!/usr/bin/env python3
"""SWE-bench Pro V2 上跑结对：agent 在题目自带的容器里改代码，判分在全新容器里跑官方 tests/test.sh。

模型都经一个兼容 OpenAI Responses 接口的模型服务调用（deepseek-flash / glm-5.3-flash / gpt-6.1-sol / gpt-6-astra）。
- 每道题一个镜像 ghcr.io/scaleapi/swe-bench_pro-v2:<id>；同一道题的几个做法共用一次拉取，全部判完就删镜像。
- codex（官方 musl 静态版 0.155.1）只读挂进容器，用 docker exec 在容器里跑，所以 agent 的命令都在题目环境里执行。
- 结对协议：开发者写完 → 独立测试员只按题目写自己的测试去跑 → REVISE 带反例打回，最多 5 轮。
- 判分照官方 patch_replay：抓 agent 容器里的 git diff，套到全新容器，跑 tests/test.sh，读 /logs/verifier/reward.txt。
模型服务地址/令牌从环境变量 PAIRBENCH_GATEWAY_URL / PAIRBENCH_GATEWAY_TOKEN 读；令牌只按变量名透传给 docker exec，从不打印。
用法：source ~/.pairbench.env && python3 run_swepro.py <arms> [--tasks sample100.json] [--par 4]
      python3 run_swepro.py --smoke <id>...   # 官方答案应判对、空补丁应判错
"""
import argparse, json, os, re, shutil, subprocess, sys, threading, time, traceback, hashlib
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

T = Path(__file__).resolve().parent
# 两套题共用这个脚本：PB_BENCH=swepro（SWE-bench Pro V2）或 deepswe（DeepSWE v1.1，从零出题防泄题）
BENCH = os.environ.get("PB_BENCH", "swepro")
TASKS = T / ("deep-swe/tasks" if BENCH == "deepswe" else "SWE-bench_Pro-os/v2/tasks")
RES = T / os.environ.get("RES", "results")
WORK = T / os.environ.get("PB_WORK", "work")
BIN = T / "bin"                       # codex + codex-code-mode-host + rg
GW = os.environ.get("PAIRBENCH_GATEWAY_URL", "").rstrip("/")
# 官方锁定协议：每题 50 分钟硬上限（结对两方共用），容器里用 timeout 掐，到点照样抓 diff 去判
# DeepSWE 官方 agent 上限 10800s（3 小时）
BUDGET_S, MAX_CONT, MAX_REVIEW = (10800 if BENCH == "deepswe" else 50 * 60) - 60, 6, 5
AGENT_GW = "http://127.0.0.1:18080"   # agent 容器断网，只能经这个回环端口 -> 宿主转发器 -> 模型服务
GW_PATH = os.environ.get("PB_GW_PATH", "/v1")  # local-pool proxy path override
SOCK = Path(os.environ.get("PB_SOCK", T / ("sock-" + BENCH)))   # 每套评测一个转发口：两个进程抢同一个 gw.sock 会互相顶掉
GRADE_CAP_S = 1800 if BENCH == "deepswe" else 3000
# 每个容器的资源上限；PAR_FILE 里写同时跑几道题（跑着跑着可以改，autoscale 会改它）
CPUS, MEM = os.environ.get("PB_CPUS", "2"), os.environ.get("PB_MEM", "6g")
PAR_FILE = Path(os.environ.get("PAR_FILE", T / "par.txt"))
LOCK = threading.Lock()

ARMS = {"solo_dsf": ("dsf", None), "solo_glm": ("glm", None),
        "dsf_t_dsf": ("dsf", "dsf"), "dsf_t_sol": ("dsf", "sol"), "glm_t_dsf": ("glm", "dsf"),
        "dsf_t_astra": ("dsf", "astra"),
        "solo_sol": ("sol", None), "solo_astra": ("astra", None),
        "sol_t_sol": ("sol", "sol"),
        "solo_solmax": ("solmax", None)}   # 10-22 加：Sol 推理开到 max 单干（榜单用 max；不设时模型服务默认 medium）   # 10-22 加：Sol 写 + Sol 审   # 10-22 加：大模型单干对照   # 10-21 加：最强的 GPT 当审查者
MODELS = {"glm": ("glm-5.3-flash", "low"), "dsf": ("deepseek-flash", ""), "sol": ("gpt-6.1-sol", ""), "astra": ("gpt-6-astra", ""), "solmax": ("gpt-6.1-sol", "max")}


def log(*a):
    with LOCK: print(time.strftime("%H:%M:%S"), *a, flush=True)


def sh(cmd, timeout=600, **kw):
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, **kw)


def has_marker(text, *marks):
    """收尾标记只认单独成行（子串匹配会把正文里的 finished 当交卷）。"""
    lines = [re.sub(r"[*`_#>\s.:!]+", " ", l).strip().upper() for l in (text or "").splitlines() if l.strip()]
    return any(l in marks for l in lines[-3:])


def verdict(text):
    m = re.findall(r"VERDICT:\s*(APPROVED|REVISE)", text or "", re.I)
    return m[-1].upper() if m else ("APPROVED" if re.search(r"\bAPPROVED\b", text or "") else "REVISE")


class GatewayDown(RuntimeError):
    pass


# ---------- 容器 ----------
def image_of(tid):
    m = re.search(r'docker_image\s*=\s*"([^"]+)"', (TASKS / tid / "task.toml").read_text())
    return m.group(1)


MIRROR = os.environ.get("PB_MIRROR", "ghcr.nju.edu.cn")   # 国内直连 ghcr 只有 ~130KB/s，走国内镜像站


def official_digest(img):
    """向 ghcr.io 官方查这个 tag 的校验值（只取清单，很快），再按校验值从镜像站拉：快，而且内容不会被换掉。"""
    repo, tag = img.split("/", 1)[1].rsplit(":", 1)
    tok = json.loads(sh(["curl", "-sS", "-m", "30", f"https://ghcr.io/token?scope=repository:{repo}:pull"]).stdout)["token"]
    acc = ("application/vnd.oci.image.index.v1+json,application/vnd.oci.image.manifest.v1+json,"
           "application/vnd.docker.distribution.manifest.v2+json,application/vnd.docker.distribution.manifest.list.v2+json")
    h = sh(["curl", "-sSI", "-m", "30", "-H", f"Authorization: Bearer {tok}", "-H", f"Accept: {acc}",
            f"https://ghcr.io/v2/{repo}/manifests/{tag}"]).stdout
    m = re.search(r"docker-content-digest:\s*(sha256:[0-9a-f]{64})", h, re.I)
    if not m: raise RuntimeError("no digest for " + tag[:60])
    return repo, m.group(1)


def pull(img):
    if not img.startswith("ghcr.io/"):   # DeepSWE 在 public.ecr.aws，没有国内镜像站，直接拉
        for i in range(4):
            r = sh(["docker", "pull", "-q", img], timeout=3600)
            if r.returncode == 0: return
            log("pull retry", i, img.split(":")[-1][:60], r.stderr.strip()[-200:]); time.sleep(30)
        raise RuntimeError("pull failed " + img)
    for i in range(4):
        try:
            repo, dg = official_digest(img)
            r = sh(["docker", "pull", "-q", f"{MIRROR}/{repo}@{dg}"], timeout=1800)
            if r.returncode == 0:
                sh(["docker", "tag", f"{MIRROR}/{repo}@{dg}", img]); return
            err = r.stderr.strip()[-200:]
        except Exception as e:
            err = repr(e)[:200]
        log("pull retry", i, img.split(":")[-1][:60], err); time.sleep(30)
    raise RuntimeError("pull failed " + img)


def drop_image(img):
    if os.environ.get("PB_KEEP_IMAGE"): return   # 同一道题重复跑时别互相删镜像
    repo = img.split("/", 1)[1].rsplit(":", 1)[0]
    ids = sh(["docker", "images", "-q", img]).stdout.split()
    sh(["docker", "rmi", "-f", img] + ids, timeout=600)   # 镜像 1–4G，判完就删（tag 和按校验值拉的那份一起删）


def start(name, img, home=None, offline=False):
    # --cpu-shares 64（默认 1024）：容器里的进程不吃宿主上的 nice，靠这个在共用机器上让着别人
    sh(["docker", "rm", "-f", name])
    # 只让容器「看到」分给它的那几个核：不加 cpuset 时 go test / jest / pytest -n auto 按宿主几十核开并行，
    # 一个容器里冒出上千线程挤在 2 核里排队，10-21 把两台机器的负载顶到上千。按容器名散列挑一段连续的核。
    ncpu, k = os.cpu_count() or 2, max(1, int(float(CPUS) + 0.999))
    first = (int(hashlib.md5(name.encode()).hexdigest(), 16) % max(1, ncpu // k)) * k
    cpuset = f"{first}-{first + k - 1}" if k > 1 else str(first)
    cmd = ["docker", "run", "-d", "--name", name, "--cpus", CPUS, "--cpuset-cpus", cpuset, "-e", f"GOMAXPROCS={k}",
           "--memory", MEM, "--pids-limit", "4096",
           "--label", "pairbench=1", "--cpu-shares", "64", "--entrypoint", "sleep",
           # 国内连不上 proxy.golang.org / sum.golang.org：改走国内镜像，否则改了 go.mod 的题连标准答案都编不过
           "-e", "GOPROXY=https://goproxy.cn,direct", "-e", "GOSUMDB=sum.golang.google.cn", "-v", f"{BIN}:/opt/pbcodex:ro"]
    if offline and not home: cmd += ["--network", "none"]   # DeepSWE 判分也断网（官方 verifier network_mode = no-network）
    if home: cmd += ["--network", "none", "-v", f"{home}:/pbhome", "-v", f"{SOCK}:/pbsock"]   # 官方：agent 阶段只许连模型接口
    r = sh(cmd + [img, "infinity"], timeout=300)
    if r.returncode: raise RuntimeError("docker run: " + r.stderr[-300:])
    if home:   # agent 容器：codex 跑命令用登录 shell，/etc/profile 会重置 PATH，这里把镜像原 PATH 补回去
        sh(["docker", "exec", name, "bash", "-c", 'echo "export PATH=$PATH" > /etc/profile.d/zz-pb-path.sh'])
        sh(["docker", "exec", "-d", name, "python3", "/opt/pbcodex/inrelay.py"])
        probe = ("import socket,time\nfor i in range(50):\n    try: socket.create_connection(('127.0.0.1',18080),1).close(); print('up'); break\n"
                 "    except Exception: time.sleep(0.2)\n"
                 "try: socket.create_connection(('1.1.1.1',443),3); print('LEAK')\nexcept Exception: print('sealed')")
        r = sh(["docker", "exec", name, "python3", "-c", probe]).stdout
        if "up" not in r or "sealed" not in r: raise RuntimeError("network lock probe failed: " + r[-200:])
    return name


def dexec(name, script, timeout=600, env=()):
    # 用 bash -c 不用 -lc：登录 shell 会读 /etc/profile 把 PATH 重置，镜像里的 /usr/local/go/bin 之类就丢了（vuls 标准答案得 0 分）
    cmd = ["docker", "exec"] + sum([["-e", e] for e in env], []) + [name, "bash", "-c", script]
    return sh(cmd, timeout=timeout)


def repo_root(name):
    return dexec(name, "cd /app 2>/dev/null && pwd || (cd /testbed && pwd)").stdout.strip() or "/app"


def start_gw_relay():
    """宿主上的转发器：SOCK/gw.sock -> 模型服务 host:port。agent 容器断网后唯一能出去的路，目的地写死成模型服务。"""
    import socket
    from urllib.parse import urlparse
    u = urlparse(GW); dst = (u.hostname, u.port or (443 if u.scheme == "https" else 80))
    SOCK.mkdir(exist_ok=True); os.chmod(SOCK, 0o777)
    path = SOCK / "gw.sock"
    path.unlink(missing_ok=True)
    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM); srv.bind(str(path)); os.chmod(path, 0o777); srv.listen(256)

    def pump(a, b):
        try:
            while True:
                d = a.recv(65536)
                if not d: break
                b.sendall(d)
        except Exception: pass
        finally:
            for x in (a, b):
                try: x.shutdown(socket.SHUT_RDWR)
                except Exception: pass

    def handle(c):
        try: g = socket.create_connection(dst, 15)
        except Exception: c.close(); return
        g.settimeout(None)
        threading.Thread(target=pump, args=(c, g), daemon=True).start(); pump(g, c)

    def loop():
        while True:
            c, _ = srv.accept(); threading.Thread(target=handle, args=(c,), daemon=True).start()
    threading.Thread(target=loop, daemon=True).start()


# ---------- agent ----------
class DockerCodex:
    """容器里的一个 codex 会话：第一轮 exec，之后 exec resume <thread>。会话记录落在挂进来的 /pbhome/codex-<who>。"""

    def __init__(self, who, cname, root, role, deadline):
        self.model, self.effort = MODELS[who]; self.deadline = deadline
        self.cname, self.root, self.home, self.thread = cname, root, f"/pbhome/codex-{role}", None
        self.usage = {"model": self.model, "input": 0, "cached": 0, "output": 0, "turns": 0}

    def turn(self, text):
        cfg = ["-c", "model_provider=pb",
               "-c", f"model_providers.pb={{ name='pb', base_url='{AGENT_GW}{GW_PATH}', env_key='PAIRBENCH_GATEWAY_TOKEN', wire_api='responses' }}",
               "-c", "model_context_window=258400", "-c", "web_search='disabled'"]
        if self.effort: cfg += ["-c", f"model_reasoning_effort={self.effort}"]
        common = ["--json", "--skip-git-repo-check", "--ignore-user-config", "--dangerously-bypass-approvals-and-sandbox",
                  "-m", self.model] + cfg
        sub = ["exec", "resume", self.thread] if self.thread else ["exec", "-C", self.root]
        cmd = ["docker", "exec", "-w", self.root, "-e", "PAIRBENCH_GATEWAY_TOKEN", "-e", f"CODEX_HOME={self.home}",
               self.cname, "bash", "-c", f'export PATH=/opt/pbcodex:$PATH; mkdir -p {self.home} && exec "$@"', "_", "/opt/pbcodex/codex"] + sub + common + [text]
        left = int(self.deadline - time.time())
        if left < 60: return "(no reply; budget exhausted)"
        cmd[cmd.index("/opt/pbcodex/codex")] = "timeout"   # 容器里的 timeout：客户端断了 codex 也会停
        cmd[cmd.index("timeout") + 1:cmd.index("timeout") + 1] = ["-k", "20", str(left), "/opt/pbcodex/codex"]
        try:
            p = subprocess.run(cmd, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=left + 60)
            out, perr = p.stdout, p.stderr
        except subprocess.TimeoutExpired as e:
            out, perr = (e.stdout or b"").decode() if isinstance(e.stdout, bytes) else (e.stdout or ""), "turn timeout"
            dexec(self.cname, "pkill -f /opt/pbcodex/codex || true")
        reply, err = "", ""
        for line in out.splitlines():
            try: j = json.loads(line)
            except Exception: continue
            t = j.get("type")
            if t == "thread.started": self.thread = j["thread_id"]
            elif t == "item.completed" and j["item"].get("type") == "agent_message": reply = j["item"].get("text") or reply
            elif t == "turn.completed":
                u = j.get("usage") or {}
                self.usage["input"] += u.get("input_tokens", 0); self.usage["cached"] += u.get("cached_input_tokens", 0)
                self.usage["output"] += u.get("output_tokens", 0); self.usage["turns"] += 1
            elif t in ("turn.failed", "error"): err = json.dumps(j)[:400]
        if not reply and re.search(r"401|403|unauthorized|invalid_api_key", err, re.I): raise GatewayDown(err)
        if not reply: log("NOREPLY", self.cname, " ".join((err or perr[-300:]).split())[:300])   # 试水排错：别把错误吞掉
        return reply or f"(no reply; {err or perr[-300:]})"


GOAL = """You are resolving an issue in the repository at {root}. This is a normal engineering task.

{instruction}

Rules: do not use the internet (the model API is the only network you may use); do not look at git history (no git log/show of other commits, no reflog); work autonomously and never ask the human.
Implement the change in the source code. You may add tests and run the repository's existing tests to check your work.
When you are completely done, reply with a short summary and the line: FINISHED"""
CONTINUE = "Continue working until the issue is fully resolved and tested. When you are completely done, reply with a short summary and the line: FINISHED"
TESTER_CARD = """You are the independent TESTER in a pair-programming session. The developer is another agent resolving an issue in this repository: {root}.
Your replies are relayed to the developer verbatim.
How you work: from the issue alone, decide what correct behaviour is, then write your OWN reproduction / acceptance tests (put them under /tmp/pairtest/, or in new files whose names contain "pairtester") and run them against the developer's current working tree. Read the existing code to understand behaviour, but judge the fix by what your tests show, not by reading the developer's diff. The developer saying "tested" is not evidence.
REVISE must include a concrete failing case (the test you ran, its output, and what the issue says should happen). APPROVED must say which tests you ran and what passed.
Do not edit non-test source files. Delete any test files you added inside the repository before you reply APPROVED. Do not use the internet or git history.
End every reply with exactly one line: VERDICT: APPROVED  or  VERDICT: REVISE

The issue:
{instruction}"""
TO_T = "\nNow write a message to the tester: what you changed and how you tested it. End with the line: TO TESTER"

CAPTURE = ('cd "{root}" && rm -rf /tmp/pairtest; git ls-files --others --exclude-standard | grep -i pairtester | xargs -r rm -f; '
           'git add -A 2>/dev/null; git diff --cached --binary > /pbhome/model.patch; git reset -q; wc -c < /pbhome/model.patch')


def gold_patch(tid):
    return TASKS / tid / ("solution/solution.patch" if BENCH == "deepswe" else "solution/gold_patch.diff")


def base_commit(tid):
    return json.loads((TASKS / tid / "tests/config.json").read_text())["base_commit"]


def grade_deepswe(tid, img, patch_path, tag):
    """DeepSWE 官方：全新断网容器 + /logs/artifacts/model.patch + 不改动的 tests/test.sh，读 reward.json。"""
    name = f"pbg-{tag}"
    start(name, img, offline=True)
    try:
        sh(["docker", "cp", str(TASKS / tid / "tests"), f"{name}:/tests"])
        dexec(name, "mkdir -p /logs/artifacts /logs/verifier")
        sh(["docker", "cp", str(patch_path), f"{name}:/logs/artifacts/model.patch"])
        r = dexec(name, "bash /tests/test.sh", timeout=GRADE_CAP_S + 120)
        rj = dexec(name, "cat /logs/verifier/reward.json 2>/dev/null").stdout.strip()
        try: d = json.loads(rj)
        except Exception: d = {"reward": 0, "parse_error": rj[:200]}
        return {"reward": int(d.get("reward", 0) == 1), "apply_rc": int(d.get("apply_failed", 0)), "detail": d,
                "apply_tail": "", "test_tail": (r.stdout + r.stderr)[-1500:]}
    finally:
        sh(["docker", "rm", "-f", name])


# DeepSWE 官方 collect：只取已提交的改动（git diff 起点..HEAD），题面都要求 "commit everything when you are done"
CAPTURE_DEEPSWE = ('cd "{root}" && git config --global --add safe.directory "{root}"; '
                   'git diff --binary {base} HEAD > /pbhome/model.patch; wc -c < /pbhome/model.patch')


def grade(tid, img, patch_path, tag):
    if BENCH == "deepswe": return grade_deepswe(tid, img, patch_path, tag)
    """官方 patch_replay：全新容器 + 补丁 + 不改动的 tests/test.sh。"""
    name = f"pbg-{tag}"
    start(name, img)
    try:
        sh(["docker", "cp", str(TASKS / tid / "tests"), f"{name}:/tests"])
        # 6 道题的判分脚本写死 GOPROXY=proxy.golang.org，国内连不上 → 新依赖下不来、标准答案都编不过。
        # 换成内容相同的国内镜像 goproxy.cn，判分标准不变。
        dexec(name, "sed -i 's#https://proxy.golang.org#https://goproxy.cn#g' /tests/*.sh")
        sh(["docker", "cp", str(patch_path), f"{name}:/tmp/replay.patch"])
        ap = dexec(name, 'cd /app 2>/dev/null || cd /testbed; [ -s /tmp/replay.patch ] || exit 0; '
                         'git apply --verbose /tmp/replay.patch || git apply --3way /tmp/replay.patch || patch --fuzz=3 -p1 -i /tmp/replay.patch',
                   timeout=300)
        r = dexec(name, "mkdir -p /logs/verifier; bash /tests/test.sh", timeout=GRADE_CAP_S + 120)
        rw = dexec(name, "cat /logs/verifier/reward.txt 2>/dev/null").stdout.strip()
        return {"reward": int(rw or 0), "apply_rc": ap.returncode, "apply_tail": (ap.stdout + ap.stderr)[-600:],
                "test_tail": (r.stdout + r.stderr)[-1500:]}
    finally:
        sh(["docker", "rm", "-f", name])


def run_job(arm, tid, img):
    out = RES / arm / f"{tid}.json"
    tag = hashlib.md5((f"{arm}/{tid}" + os.environ.get("PB_REP", "")).encode()).hexdigest()[:10]
    home = WORK / arm / tid; shutil.rmtree(home, ignore_errors=True); home.mkdir(parents=True)
    os.chmod(home, 0o777)
    cname = start(f"pba-{tag}", img, home)
    rec = {"arm": arm, "id": tid, "started": time.time(), "reviews": [], "turns": 0}
    t0 = time.time(); dev = rev = None
    try:
        root = repo_root(cname)
        instr = (TASKS / tid / "instruction.md").read_text()
        who, tester = ARMS[arm]
        deadline = t0 + BUDGET_S
        dev = DockerCodex(who, cname, root, "dev", deadline)
        if tester:
            card = TESTER_CARD.format(root=root, instruction=instr)
            rev = DockerCodex(tester, cname, root, "tester", deadline)
            rev.turn(card + "\n\nFor now reply with just OK; the developer's first message will follow.")
        reply = dev.turn(GOAL.format(root=root, instruction=instr)); rec["turns"] += 1
        for _ in range(MAX_CONT):
            if has_marker(reply, "FINISHED") or time.time() > deadline - 60: break
            reply = dev.turn(CONTINUE); rec["turns"] += 1
        if rev:
            msg = dev.turn(TO_T)
            for rnd in range(MAX_REVIEW):
                r = rev.turn(f"Message from the developer:\n\n{msg}")
                v = verdict(r)
                rec["reviews"].append({"t": round(time.time() - t0), "round": rnd, "verdict": v, "text": r[:3000]})
                if v == "APPROVED" or time.time() > deadline - 60: break
                msg = dev.turn(f"Tester's reply:\n\n{r}\n\nReproduce what the tester found, fix it, re-run the tests, then write your next message to the tester." + TO_T)
        rec["elapsed_s"] = time.time() - t0
        cap = CAPTURE_DEEPSWE.format(root=root, base=base_commit(tid)) if BENCH == "deepswe" else CAPTURE.format(root=root)
        rec["patch_bytes"] = int((dexec(cname, cap, timeout=300).stdout.strip() or "0").split()[-1])
    except GatewayDown as e:
        rec["error"] = "gateway_down " + str(e)[:200]
    except Exception as e:
        rec["error"] = repr(e)[:300]; traceback.print_exc()
    finally:
        rec["usage"] = {"dev": dev.usage if dev else None, "rev": rev.usage if rev else None}
        try: dexec(cname, "chmod -R a+rwX /pbhome", timeout=1800)   # 容器里是 root 写的，放开权限，宿主上才删得掉
        except Exception as e: log("CHMOD SKIP", arm, tid[:60], repr(e)[:80])   # node_modules 巨大时会超时；别让它吞掉结果和删容器
        sh(["docker", "rm", "-f", cname])
    if rec.get("error"):
        log("NOT RECORDED", arm, tid[:60], rec["error"][:120]); return rec
    (RES / arm).mkdir(parents=True, exist_ok=True)
    shutil.copy(home / "model.patch", RES / arm / f"{tid}.patch")
    try: rec["grade"] = g = grade(tid, img, home / "model.patch", tag)
    except subprocess.TimeoutExpired:   # 补丁让测试卡死：记失败并打标，别让整道题崩掉、丢掉做题记录
        rec["grade"] = g = {"reward": 0, "timeout": True}; rec["grade_timeout"] = True
    rec["passed"] = g["reward"] == 1
    out.write_text(json.dumps(rec, ensure_ascii=False, indent=1))
    log(arm, tid[:70], "PASS" if rec["passed"] else "fail", f"{rec['elapsed_s']/60:.0f}min reviews={len(rec['reviews'])} patch={rec['patch_bytes']}B")
    return rec


def run_task(tid, arms):
    todo = [a for a in arms if not (RES / a / f"{tid}.json").exists()]
    if not todo: return
    if not os.environ.get("PB_NO_CLAIM"):   # 加跑的做法不占题：别让多机分题以为这题已被本机领走
        (T / "started").mkdir(exist_ok=True); (T / "started" / tid).touch()
    img = image_of(tid)
    try:
        pull(img)
        with ThreadPoolExecutor(len(todo)) as ex: list(ex.map(lambda a: run_job(a, tid, img), todo))
        gf = RES / "_gold" / f"{tid}.json"   # 顺手判一次标准答案：它都过不了的题，统计时剔除（环境坏了不怪模型）
        if not gf.exists():
            gf.parent.mkdir(parents=True, exist_ok=True)
            g = grade(tid, img, gold_patch(tid), "gold-" + hashlib.md5(tid.encode()).hexdigest()[:8])
            gf.write_text(json.dumps(g, ensure_ascii=False, indent=1)); log("GOLD", tid[:60], g["reward"])
    except Exception as e:
        log("TASK CRASH", tid[:60], repr(e)[:200])
    finally:
        drop_image(img)


def smoke(ids):
    for tid in ids:
        img = image_of(tid); pull(img)
        empty = WORK / "empty.patch"; empty.parent.mkdir(parents=True, exist_ok=True); empty.write_text("")
        g1 = grade(tid, img, gold_patch(tid), "smoke-gold")
        g0 = grade(tid, img, empty, "smoke-empty")
        log("SMOKE", tid[:60], "gold", g1["reward"], "empty", g0["reward"], "OK" if (g1["reward"], g0["reward"]) == (1, 0) else "BAD")
        drop_image(img)


if __name__ == "__main__":
    ap = argparse.ArgumentParser(); ap.add_argument("arms", nargs="?"); ap.add_argument("--tasks", default="sample100.json")
    ap.add_argument("--par", type=int, default=2); ap.add_argument("--ids"); ap.add_argument("--smoke", nargs="*")
    ap.add_argument("--order", default="fwd", choices=["fwd", "rev", "mid"])
    a = ap.parse_args()
    if a.smoke is not None: smoke(a.smoke); sys.exit()
    if not GW or not os.environ.get("PAIRBENCH_GATEWAY_TOKEN"): sys.exit("source ~/.pairbench.env first")
    start_gw_relay()
    arms = a.arms.split(","); assert all(x in ARMS for x in arms), arms
    ids = a.ids.split(",") if a.ids else json.load(open(T / a.tasks))
    if a.order == "rev": ids = ids[::-1]
    elif a.order == "mid": ids = ids[len(ids) // 2:] + ids[:len(ids) // 2]
    if not PAR_FILE.exists(): PAR_FILE.write_text(str(a.par))
    log("tasks", len(ids), "arms", arms)
    busy = []; lk = threading.Lock(); queue = list(ids)   # queue：还没开跑的题，按原顺序先来先到

    def par():
        try: return max(1, int(PAR_FILE.read_text().strip()))
        except Exception: return a.par

    def worker(tid):
        while True:   # 同时跑的题数由 par.txt 现读决定，便于逐步加并发
            with lk:
                if len(busy) < par() and queue and queue[0] == tid: queue.pop(0); busy.append(tid); break
            time.sleep(2)
        try: run_task(tid, arms)
        finally:
            with lk: busy.remove(tid)

    with ThreadPoolExecutor(32) as ex: list(ex.map(worker, ids))
    log("ALL DONE")

# 跑在 agent 容器里（容器是 --network none，只有回环）：127.0.0.1:18080 -> /pbsock/gw.sock（宿主上的转发器，只通模型服务）
import socket, threading, sys
def pump(a, b):
    try:
        while True:
            d = a.recv(65536)
            if not d: break
            b.sendall(d)
    except Exception: pass
    finally:
        for s in (a, b):
            try: s.shutdown(socket.SHUT_RDWR)
            except Exception: pass
def handle(c):
    u = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try: u.connect("/pbsock/gw.sock")
    except Exception: c.close(); return
    threading.Thread(target=pump, args=(c, u), daemon=True).start(); pump(u, c)
srv = socket.socket(); srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("127.0.0.1", 18080)); srv.listen(64)
while True:
    c, _ = srv.accept(); threading.Thread(target=handle, args=(c,), daemon=True).start()

# 10-22 判分超时的答案：机器空闲时把判分时限放宽到 2 小时重判一次（答案不重做）。旧判分留在 grade_before_long_regrade。
import json, os, sys
from concurrent.futures import ThreadPoolExecutor
import run_swepro as r
r.GRADE_CAP_S = 7200
jobs = [x.split("/") for x in sys.argv[2:]]   # arm/tid
def one(j):
    arm, tid = j; f = r.RES / arm / f"{tid}.json"; d = json.load(open(f))
    p = r.RES / arm / f"{tid}.patch"
    img = r.image_of(tid); r.pull(img)
    try: g = r.grade(tid, img, p.resolve(), "lg-" + str(abs(hash(arm + tid)) % 10**8))
    except Exception as e: g = {"reward": 0, "timeout": True, "error": repr(e)[:200]}
    d["grade_before_long_regrade"] = d["grade"]; d["grade"] = g; d["passed"] = bool(g.get("reward"))
    d["grade_timeout"] = bool(g.get("timeout")); d["regraded"] = "10-22 判分超时，空闲时放宽到 2h 重判"
    f.write_text(json.dumps(d, ensure_ascii=False, indent=1)); r.log("LONGRG", arm, tid, g.get("reward"), "timeout" if g.get("timeout") else "")
with ThreadPoolExecutor(int(sys.argv[1])) as ex: list(ex.map(one, jobs))

#!/usr/bin/env python3
"""10-23 汇总：每套题每种做法的 通过率(95% Wilson) / 总花费 / 每题花费中位 / 每题分钟中位，外加配对 McNemar。
取数：collected-<RES>/<机器>/<做法>/<题>.json；去占位；同 (做法,题) 多份时 优先补跑记录(rerun_reason)，再取开工早的；
剔除标准答案在所有机器上都判不过的题；价格一律按原价（Standard 档，不乘 fast 倍率）。"""
import json, math, statistics as st, sys
from pathlib import Path
B = Path(__file__).resolve().parent
PRICE = {"glm-5.3-flash": (0.15, 0.03, 0.50), "deepseek-flash": (0.15, 0.03, 0.60),
         "gpt-6.1-sol": (2.00, 0.10, 10.00), "gpt-6-astra": (10.00, 1.00, 50.00)}   # 美元/百万 token：未命中输入, 命中输入, 输出
ORDER = ["solo_dsf", "solo_glm", "solo_sol", "solo_solmax", "solo_astra", "dsf_t_dsf", "glm_t_dsf", "dsf_t_sol", "dsf_t_astra", "sol_t_sol"]
PAIRS = [("solo_sol", "solo_dsf"), ("solo_solmax", "solo_dsf"), ("solo_astra", "solo_dsf"), ("solo_solmax", "solo_sol"),
         ("solo_astra", "solo_sol"), ("sol_t_sol", "solo_sol"), ("dsf_t_sol", "solo_dsf"), ("dsf_t_astra", "solo_dsf"),
         ("dsf_t_dsf", "solo_dsf"), ("dsf_t_sol", "solo_sol"), ("dsf_t_astra", "solo_astra"), ("glm_t_dsf", "solo_glm"), ("solo_dsf", "solo_glm")]
def usd(u):
    if not u: return 0.0
    p = PRICE[u["model"]]; return ((u["input"] - u["cached"]) * p[0] + u["cached"] * p[1] + u["output"] * p[2]) / 1e6
def wilson(x, n, z=1.96):
    p = x / n; c = (p + z * z / 2 / n) / (1 + z * z / n); h = z * math.sqrt(p * (1 - p) / n + z * z / 4 / n / n) / (1 + z * z / n); return c - h, c + h
def mcnemar(b, c):
    n = b + c
    return 1.0 if n == 0 else min(1.0, 2 * sum(math.comb(n, i) for i in range(min(b, c) + 1)) / 2 ** n)
def load(res):
    best, gold = {}, {}
    for f in (B / f"collected-{res}").glob("*/*/*.json"):
        d = json.loads(f.read_text())
        if f.parent.name == "_gold":
            gold.setdefault(f.stem, []).append(d.get("reward")); continue
        if f.parent.name.startswith("_") or d.get("placeholder") or "id" not in d: continue
        k = (f.parent.name, d["id"]); rank = (0 if d.get("rerun_reason") else 1, float(d.get("started") or 0))
        if k not in best or rank < best[k][0]: best[k] = (rank, d, f)
    bad = {t for t, v in gold.items() if not any(v)}
    return {k: (d, f) for k, (_, d, f) in best.items() if k[1] not in bad}, bad
for res in ("results-hard", "results-deepswe"):
    R, bad = load(res)
    P = {k: bool(d.get("passed")) for k, (d, _) in R.items()}
    print(f"\n=== {res}   剔除坏题 {len(bad)}: {sorted(bad)}")
    print(f"{'做法':12s} {'通过':>8s} {'率':>6s} {'95%置信区间':>15s} {'总$':>8s} {'每题$中位':>9s} {'每题分中位':>9s} {'补跑':>4s} {'换线路':>4s}")
    arms = [a for a in ORDER if any(x == a for x, _ in R)] + sorted({x for x, _ in R} - set(ORDER))
    for a in arms:
        rs = [(k, d) for k, (d, _) in R.items() if k[0] == a]; n = len(rs); x = sum(P[k] for k, _ in rs)
        cost = [usd((d.get("usage") or {}).get("dev")) + usd((d.get("usage") or {}).get("rev")) for _, d in rs if (d.get("usage") or {}).get("dev")]
        mins = [d["elapsed_s"] / 60 for _, d in rs if d.get("elapsed_s")]
        lo, hi = wilson(x, n)
        print(f"{a:12s} {x:>3d}/{n:<4d} {x/n:6.1%} [{lo:5.1%},{hi:5.1%}] {sum(cost):8.2f} {st.median(cost) if cost else 0:9.3f} {st.median(mins) if mins else 0:9.1f} "
              f"{sum(bool(d.get('rerun_reason')) for _, d in rs):>4d} {sum(d.get('route') == 'acct' for _, d in rs):>4d}")
    print("配对 McNemar（b=只前者对, c=只后者对）")
    for a, b_ in PAIRS:
        ids = {t for x, t in P if x == a} & {t for x, t in P if x == b_}
        if not ids: continue
        b = sum(P[(a, t)] and not P[(b_, t)] for t in ids); c = sum(P[(b_, t)] and not P[(a, t)] for t in ids); p = mcnemar(b, c)
        print(f"  {a:12s} vs {b_:12s} n={len(ids):3d}  {b:2d}:{c:<2d}  p={p:.3f}{'  *' if p < 0.05 else ''}")

# 10-22 补跑：答题过程被模型服务错误（504/503/限流/断流）打断、或答题超时而失败的 (做法, 题)。
# 先把旧记录挪进 <RES>/_superseded/<arm>/（不删），再用独立工作目录和转发口重跑这一对。
# 用法：PB_BENCH=... RES=... python3 rerun_driver.py jobs.json <并发>
import json, os, sys, shutil, time
from concurrent.futures import ThreadPoolExecutor
os.environ.setdefault("PB_WORK", "work-rerun"); os.environ["PB_NO_CLAIM"] = "1"
import run_swepro as r
jobs = json.load(open(sys.argv[1])); par = int(sys.argv[2]) if len(sys.argv) > 2 else 4
for j in jobs:
    for ext in (".json", ".patch"):
        src = r.RES / j["arm"] / (j["id"] + ext)
        if src.exists():
            dst = r.RES / "_superseded" / j["arm"]; dst.mkdir(parents=True, exist_ok=True)
            shutil.move(str(src), str(dst / (j["id"] + ext)))
r.start_gw_relay()
def one(j):
    r.run_task(j["id"], [j["arm"]])
    f = r.RES / j["arm"] / (j["id"] + ".json")
    if f.exists():   # 记一笔这是补跑，原因留档
        d = json.load(open(f)); d["rerun_reason"] = j["why"]; f.write_text(json.dumps(d, ensure_ascii=False, indent=1))
with ThreadPoolExecutor(par) as ex: list(ex.map(one, jobs))
r.log("RERUN DONE", len(jobs))