小模型写、大模型审：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）；Opus 经同一服务的 Anthropic 接口，用 Claude Code 跑。
- 每道题一个镜像 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
BUDGET_S = int(os.environ.get("PB_BUDGET_S", BUDGET_S))   # 10-25：跑满时限的失败题放宽时限重跑时用
AGENT_GW = "http://127.0.0.1:18080"   # agent 容器断网，只能经这个回环端口 -> 宿主转发器 -> 模型服务
GW_PATH = os.environ.get("PB_GW_PATH", "/v1")  # local-pool proxy path override
ACCT = set(filter(None, os.environ.get("PB_ACCT_MODELS", "").split(",")))
CLAUDE_CODE = {"opus"}   # 10-24：这些模型改用 Claude Code 跑（容器里 /opt/pbcodex/claude-linux-x64[-musl]），其余仍用 Codex   # 10-23：这些模型走 /acct/v1（宿主上的路由转到另一条模型服务线路），其余走默认线路
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), "solo_opus": ("opus", None), "dsf_t_opus": ("dsf", "opus")}   # 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"), "opus": ("claude-opus-5-5", "medium")}   # 10-24 加：Opus 5.5，用 Claude Code 跑


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.path = "/acct/v1" if who in ACCT else GW_PATH
        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}
        if who in ACCT: self.usage["route"] = "acct"

    def turn(self, text):
        cfg = ["-c", "model_provider=pb",
               "-c", f"model_providers.pb={{ name='pb', base_url='{AGENT_GW}{self.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 {}
                # turn.completed 报的是整个会话到此为止的累计（resume 也接着累计），取最新值，别逐轮相加（10-26 修）
                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; self.usage["usage_cum_ok"] = True
            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:]})"



class DockerClaude:
    """容器里的一个 Claude Code 会话（10-24 加，Opus 用）：第一轮 claude -p，之后 --resume <session>。
    提示词走 stdin；令牌只按变量名传进容器；会话记录落在 /pbhome/claude-<role>。断网环境里关掉联网工具。"""

    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/claude-{role}", None
        self.usage = {"model": self.model, "input": 0, "cached": 0, "cache_write": 0, "output": 0, "turns": 0,
                      "harness": "claude-code", "cc_cost_usd": 0.0}
        os.environ["ANTHROPIC_AUTH_TOKEN"] = os.environ["PAIRBENCH_GATEWAY_TOKEN"]

    def turn(self, text):
        left = int(self.deadline - time.time())
        if left < 60: return "(no reply; budget exhausted)"
        m = self.model
        env = {"ANTHROPIC_BASE_URL": AGENT_GW, "CLAUDE_CONFIG_DIR": self.home, "IS_SANDBOX": "1",
               "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", "DISABLE_AUTOUPDATER": "1",
               # 模型服务上没有 Haiku：后台小活和子代理也用同一个模型
               "ANTHROPIC_DEFAULT_HAIKU_MODEL": m, "ANTHROPIC_DEFAULT_SONNET_MODEL": m, "ANTHROPIC_DEFAULT_OPUS_MODEL": m,
               "CLAUDE_CODE_SUBAGENT_MODEL": m, "BASH_DEFAULT_TIMEOUT_MS": "600000", "BASH_MAX_TIMEOUT_MS": "1800000"}
        args = ["-p", "--output-format", "stream-json", "--verbose", "--model", m,
                "--dangerously-skip-permissions", "--disallowedTools", "WebSearch,WebFetch"]
        if self.effort: args += ["--effort", self.effort]
        if self.thread: args += ["--resume", self.thread]
        pick = ('B=/opt/pbcodex/claude-linux-x64; ls /lib/ld-musl* >/dev/null 2>&1 && B=/opt/pbcodex/claude-linux-x64-musl; '
                f'mkdir -p {self.home} && exec timeout -k 20 {left} "$B" "$@"')
        cmd = ["docker", "exec", "-i", "-w", self.root, "-e", "ANTHROPIC_AUTH_TOKEN"]
        for k, v in env.items(): cmd += ["-e", f"{k}={v}"]
        cmd += [self.cname, "bash", "-c", f"export PATH=$PATH:/opt/pbcodex; {pick}", "_"] + args
        try:
            p = subprocess.run(cmd, input=text, 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/claude-linux || true")
        reply, err = "", ""
        for line in out.splitlines():
            try: j = json.loads(line)
            except Exception: continue
            t = j.get("type")
            if t == "system" and j.get("subtype") == "init" and not self.thread: self.thread = j.get("session_id")
            elif t == "result":
                self.thread = j.get("session_id") or self.thread
                if j.get("is_error"): err = json.dumps(j)[:400]
                else: reply = j.get("result") or ""
                # --resume 后 result 报的是整个会话到此为止的累计，取最新值，别逐轮相加（10-26 修）
                tot = {"input": 0, "cached": 0, "cache_write": 0, "output": 0}
                for u in (j.get("modelUsage") or {}).values():   # 每个模型一份：未命中输入 / 命中缓存 / 写缓存 / 输出
                    cr, cw = u.get("cacheReadInputTokens", 0), u.get("cacheCreationInputTokens", 0)
                    tot["input"] += u.get("inputTokens", 0) + cr + cw; tot["cached"] += cr
                    tot["cache_write"] += cw; tot["output"] += u.get("outputTokens", 0)
                self.usage.update(tot); self.usage["cc_cost_usd"] = float(j.get("total_cost_usd") or 0)
                self.usage["turns"] += 1; self.usage["usage_cum_ok"] = True
        if not reply and re.search(r"401|403|unauthorized|authentication_error|invalid.api.key", err + perr, re.I): raise GatewayDown(err or perr[-300:])
        if not reply: log("NOREPLY", self.cname, " ".join((err or perr[-300:]).split())[:300])
        return reply or f"(no reply; {err or perr[-300:]})"


def agent(who, *a):
    return (DockerClaude if who in CLAUDE_CODE else DockerCodex)(who, *a)


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 = agent(who, cname, root, "dev", deadline)
        if tester:
            card = TESTER_CARD.format(root=root, instruction=instr)
            rev = agent(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-26：Codex 用量按会话记录重算。turn.completed 报的是「会话累计」，旧代码逐轮相加 → 多轮的题多算。
# 对每个结果，找开工时间对得上的那次尝试（work*/<arm>/<id>/codex-<role>/sessions/**.jsonl），
# 每个会话文件取最后一条 total_token_usage（就是这个会话的真实合计），同一角色多个会话文件相加。
# 校验：只跑了 1 轮的角色，重算值必须等于结果文件里记的值。用法（容器里 root 跑）：python3 codex_cost.py <主机名>
import json, glob, os, sys
from datetime import datetime
host = sys.argv[1]
ROLE_DIRS = {"dev": ["dev"], "rev": ["tester", "advisor", "rev"]}
def sess_info(f):
    t0 = None; last = None
    for line in open(f, errors="ignore"):
        try: e = json.loads(line)
        except Exception: continue
        if t0 is None and e.get("timestamp"): t0 = datetime.fromisoformat(e["timestamp"].replace("Z", "+00:00")).timestamp()
        p = e.get("payload") or {}
        if p.get("type") == "token_count" and p.get("info"): last = p["info"]["total_token_usage"]
    return t0, last
dirs = {}   # (arm, tid) -> [attempt dir]
for d in glob.glob("work*/*/*"):
    if os.path.isdir(d): dirs.setdefault((d.split("/")[1], d.split("/")[2]), []).append(d)
out = {}
for b in ("deepswe", "hard", "deepswe-af", "hard-af"):
    for rf in glob.glob(f"results-{b}/*/*.json"):
        arm = rf.split("/")[1]
        if arm.startswith("_"): continue
        try: r = json.load(open(rf))
        except Exception: continue
        if "id" not in r or not r.get("usage"): continue
        st = float(r.get("started") or 0); key = f"{b}/{arm}/{r['id']}"
        roles = {k: v for k, v in r["usage"].items() if v and not str(v.get("model", "")).startswith("claude")}
        if not roles: continue
        best = None
        for d in dirs.get((arm, r["id"]), []):
            sess = {role: [sess_info(f) for rd in ROLE_DIRS[role] for f in glob.glob(f"{d}/codex-{rd}/sessions/**/*.jsonl", recursive=True)] for role in roles}
            ts = [t for v in sess.values() for t, _ in v if t]
            if not ts: continue
            gap = min(ts) - st
            if -120 < gap < 3600 and (best is None or abs(gap) < abs(best[1])): best = (d, gap, sess)
        rec = {"host": host, "recorded": {k: {x: v.get(x, 0) for x in ("input", "cached", "output", "turns")} for k, v in roles.items()}}
        if best:
            rec["dir"], rec["gap_s"] = best[0], round(best[1])
            for role, lst in best[2].items():
                lst = [u for _, u in lst if u]
                if lst: rec[role] = {"input": sum(u["input_tokens"] for u in lst), "cached": sum(u["cached_input_tokens"] for u in lst),
                                     "output": sum(u["output_tokens"] for u in lst), "sessions": len(lst)}
        out[key] = rec
json.dump(out, open(f"codex-usage-{host}.json", "w"), indent=1)
n = len(out); full = sum(all(k in v for k in v["recorded"]) for v in out.values())
bad1 = [k for k, v in out.items() for role, rv in v["recorded"].items() if rv["turns"] == 1 and role in v and v[role]["input"] != rv["input"]]
ok1 = sum(1 for v in out.values() for role, rv in v["recorded"].items() if rv["turns"] == 1 and role in v and v[role]["input"] == rv["input"])
print(host, "results", n, "fully recomputed", full, "| 1-turn check ok", ok1, "mismatch", len(bad1), bad1[:5])

# 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 倍率）。
--strict：DeepSWE 上「改了原有测试」的通过答案，按去掉那些改动后的重判结果算（strict-out-local/）。"""
import json, math, statistics, 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),
         "claude-opus-5-5": (4.00, 0.20, 20.00)}   # Opus 写缓存另按 1.25 倍输入价（5 分钟缓存），见 usd()   # 美元/百万 token：未命中输入, 命中输入, 输出
ORDER = ["solo_dsf", "solo_glm", "solo_sol", "solo_solmax", "solo_astra", "dsf_t_dsf", "glm_t_dsf", "dsf_t_sol", "dsf_a_sol", "dsf_t_astra", "sol_t_sol", "solo_opus", "dsf_t_opus", "dsf_a_opus"]
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"),
         ("solo_opus", "solo_dsf"), ("solo_opus", "solo_sol"), ("dsf_t_opus", "solo_dsf"), ("dsf_t_opus", "solo_opus"), ("dsf_t_opus", "dsf_t_sol"), ("dsf_t_sol", "solo_solmax"), ("dsf_t_sol", "solo_opus"), ("sol_t_sol", "solo_dsf"),
         ("dsf_a_sol", "dsf_t_sol"), ("dsf_a_sol", "solo_dsf"), ("dsf_a_sol", "solo_sol"),
         ("dsf_a_opus", "dsf_t_opus"), ("dsf_a_opus", "solo_opus"), ("dsf_a_opus", "dsf_a_sol"), ("dsf_a_opus", "solo_dsf"), ("dsf_a_opus", "dsf_t_sol")]   # 10-26 顾问模式（顾问 Sol / Opus）
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] + u.get("cache_write", 0) * p[0] * 0.25) / 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)
import os
ACCT_ARMS = set(filter(None, os.environ.get("PB_ACCT_ARMS", "").split(",")))   # 10-23：这些做法改用另一条线路重跑的那份（collected-<RES>-af），原线路那份不用
def load(res):
    best, gold = {}, {}
    files = [f for f in (B / f"collected-{res}").glob("*/*/*.json") if f.parent.name not in ACCT_ARMS]
    files += [f for f in (B / f"collected-{res}-af").glob("*/*/*.json") if f.parent.name in ACCT_ARMS]
    for f in files:
        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"]); st0 = float(d.get("started") or 0)
        rank = (0, -st0) if d.get("rerun_reason") else (1, st0)   # 补跑优先；补跑多次（跨机器）取最新那次，没补跑的取最早那次
        d["_src"] = str(f)   # 10-26：记下出处（机器、哪条线路），给 Codex 用量重算对号用
        if k not in best or rank < best[k][0]: best[k] = (rank, d, f)
    # PB_ACCT_PARTIAL：这些做法只有一部分题整组重跑干净（collected-<RES>-af/clean-<做法>.json 列出题号），
    # 这些题里，只在「重跑通过、原来没通过」时换成重跑结果（两次取好，等于这些题多给了一次机会），其余题仍用原来那份
    for arm in filter(None, os.environ.get("PB_ACCT_PARTIAL", "").split(",")):
        cf = B / f"collected-{res}-af" / f"clean-{arm}.json"
        if not cf.exists(): continue
        for t in json.loads(cf.read_text())["clean"]:
            # 10-25：同一题在两台机器上可能各有一份（被打断的旧份 + 重跑份），有重跑份就只看重跑份
            cands = [(json.loads(f.read_text()), f) for f in (B / f"collected-{res}-af").glob(f"*/{arm}/{t}.json")]
            cands = [x for x in cands if not x[0].get("placeholder") and "id" in x[0]]
            rr = [x for x in cands if x[0].get("rerun_reason")]
            for d, f in (rr or cands)[:1]:
                old = best.get((arm, t))
                d["_src"] = str(f)
                if not d.get("placeholder") and "id" in d and d.get("passed") and not (old and old[1].get("passed")):
                    best[(arm, t)] = ((-1, 0), 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
def strict_fail():
    out = set()
    for f in (B / "strict-out-local").glob("*.json"):
        j = json.loads(f.read_text())
        if not j["grade"].get("reward"): out.add((j["arm"], j["id"]))
    return out
# 10-25：Opus（Claude Code）用量。结果文件里只记了每轮 result 事件的用量：超时被掐的那一轮没有 result，记成 0；
# 会话记录（opus-usage-from-transcripts.json）补得上被掐那轮，但看不到后台小调用（起标题等，也走 Opus）。两者取大。
OPUS_FIX = json.loads((B / "opus-usage-from-transcripts.json").read_text()) if (B / "opus-usage-from-transcripts.json").exists() else {}
# 10-26：Codex 的 turn.completed 报的是「会话累计」，旧代码逐轮相加，多轮的题多算了。
# codex-usage-<机器>.json 是按会话记录重算的真实值（codex_cost.py）；对号时要求结果文件里记的值一致，防止对错尝试。
# 只跑 1 轮的角色本来就对，直接用；会话记录已不在的（工作目录已清），按同做法、同角色、同轮数重算过的题的「真实/记录」中位比例估算。
CODEX_FIX = {}
for _f in B.glob("codex-usage-*.json"):
    _h = _f.stem.split("codex-usage-")[1]
    for _k, _v in json.loads(_f.read_text()).items(): CODEX_FIX[(_h, _k)] = _v
_ratio = {}
for (_h, _k), _v in CODEX_FIX.items():
    for _role, _rv in _v["recorded"].items():
        _t = _v.get(_role)
        if not _t or _rv["turns"] <= 1 or not _rv["input"] or _t["input"] > _rv["input"]: continue
        _arm = _k.split("/")[1]
        for _key in ((_arm, _role, _rv["turns"]), (_role, _rv["turns"]), (_rv["turns"],)):
            _ratio.setdefault(_key, []).append({x: _t[x] / _rv[x] if _rv[x] else 1.0 for x in ("input", "cached", "output")})
CODEX_STAT = {"exact": 0, "est": 0, "one_turn": 0, "zero_est": 0}
LONG_MED = {}   # (res, arm, role) -> 跑了 40 分钟以上的题的用量中位数，给「被时限截断、记成 0」的记录估值
def codex_fixed(res, arm, d, role, u):
    if u.get("usage_cum_ok"): return u   # 修过之后记的，本来就是真实值
    if not u.get("input") and not u.get("turns") and (d.get("elapsed_s") or 0) >= 2400 and (res, arm, role) in LONG_MED:
        CODEX_STAT["zero_est"] += 1; return {**u, **LONG_MED[(res, arm, role)], "estimated": True}
    if (u.get("turns") or 0) <= 1 and u.get("input"): CODEX_STAT["one_turn"] += 1; return u
    src = d.get("_src", ""); parts = src.split("/")
    try:
        ci = [i for i, x in enumerate(parts) if x.startswith("collected-results-")][-1]
        bench = parts[ci].split("collected-results-")[1]; host = parts[ci + 1]
    except IndexError: bench, host = res.split("-", 1)[1], ""
    rec = CODEX_FIX.get((host, f"{bench}/{arm}/{d['id']}"))
    # 记录为 0 = 那一轮被时限截断、没报用量；会话记录里有就用会话记录（10-26 审查补）
    if rec and role in rec and rec["recorded"].get(role, {}).get("input") == u.get("input") and (rec[role]["input"] <= u["input"] or not u.get("input")):
        CODEX_STAT["exact"] += 1; return {**u, "input": rec[role]["input"], "cached": rec[role]["cached"], "output": rec[role]["output"]}
    for key in ((arm, role, u["turns"]), (role, u["turns"]), (u["turns"],)):
        rs = _ratio.get(key)
        if rs and len(rs) >= 5:
            CODEX_STAT["est"] += 1
            return {**u, **{x: round(u[x] * statistics.median(r[x] for r in rs)) for x in ("input", "cached", "output")}, "estimated": True}
    return u
def opus_fixed(res, arm, d, role):
    u = (d.get("usage") or {}).get(role)
    if u and not str(u.get("model", "")).startswith("claude"): return codex_fixed(res, arm, d, role, u)
    v = (OPUS_FIX.get(f"{res.split('-')[-1]}/{arm}/{d['id']}") or {}).get(role)
    # 10-26：Claude Code --resume 后 result 报的也是会话累计，旧代码逐轮相加 → 多轮的记录多算；多轮一律用会话记录重算值
    if v and u and u.get("model") == "claude-opus-5-5" and (u.get("turns") or 0) >= 2: return v
    if v and (not u or u.get("model") == "claude-opus-5-5") and usd(v) > usd(u): return v
    return u
strict = "--strict" in sys.argv; SF = strict_fail() if strict else set()
for res in ("results-hard", "results-deepswe"):
    R, bad = load(res)
    for (_a, _t), (_d, _) in R.items():   # 先算「长题」用量中位数（只用有用量的记录）
        for _role in ("dev", "rev"):
            _u = (_d.get("usage") or {}).get(_role)
            if _u and _u.get("input") and not str(_u.get("model", "")).startswith("claude") and (_d.get("elapsed_s") or 0) >= 2400:
                LONG_MED.setdefault((res, _a, _role), []).append(codex_fixed(res, _a, _d, _role, _u))
    for _k, _v in list(LONG_MED.items()):
        if isinstance(_v, list): LONG_MED[_k] = {x: round(statistics.median(z[x] for z in _v)) for x in ("input", "cached", "output")}
    P = {k: bool(d.get("passed")) and not (res.endswith("deepswe") and k in SF) for k, (d, _) in R.items()}
    print(f"\n=== {res}{'（严格：改测试的按重判）' if strict else ''}   剔除坏题 {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(opus_fixed(res, a, d, "dev")) + usd(opus_fixed(res, a, d, "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))

#!/usr/bin/env python3
"""10-25 新做法 dsf_a_opus：Opus 只动嘴（读题读代码 → 方案提示 + 文字验收用例），不写文件、不跑东西；
dsf 把用例写成测试、自己跑、自己修；最后 Opus 只看 dsf 的报告和 diff 再给意见（最多补 2 轮）。
不改 run_swepro.py（别的进程还在用它），只在本进程里把 run_job 换掉。
用法：source ~/.pairbench.env && PB_BENCH=deepswe RES=results-deepswe PB_WORK=work-adv PB_SOCK=$HOME/pb-swepro/sock-adv \
      PB_KEEP_IMAGE=1 python3 run_advice.py <题目清单.json> <并发>
"""
import json, os, sys, time, shutil, hashlib, subprocess, traceback
from concurrent.futures import ThreadPoolExecutor
os.environ["PB_NO_CLAIM"] = "1"
import run_swepro as r

# 10-25 加：顾问模式可换顾问模型（PB_ADV_MODEL=sol → 做法 dsf_a_sol）
ADV_MODEL = os.environ.get("PB_ADV_MODEL", "opus")
ARM = f"dsf_a_{ADV_MODEL}"
r.ARMS[ARM] = ("dsf", ADV_MODEL)
MAX_ADV_REVIEW = 2

ADVISOR = """You are the ADVISOR in a pair-programming session. Another agent (the developer) will implement the change in the repository at {root}; you only give advice.
STRICT RULES: do NOT create, edit or delete any file; do NOT build, install, run tests or run programs. Only READ: ls, cat, grep, find, sed -n, head, tail. Do not use the internet or git history.
Read the issue below and the relevant code, then reply with exactly two sections:

PLAN: the approach in a few bullets: which files/functions to change, the order of work, and the pitfalls or edge cases that are easy to get wrong.

ACCEPTANCE CASES: a numbered list of 8-20 concrete cases derived from the issue. Each case = setup/input -> exact expected result (return value, output, error type/message). Cover the main behaviour, every option the issue mentions, and edge cases (empty, nil/None, boundaries, invalid input, ordering). No test code, plain text only.

End with the line: ADVICE DONE

The issue:
{instruction}"""

DEV_ADVICE = """

An advisor (a stronger model) read the issue and the code and gave the advice below. Use the plan where it helps.
Turn EVERY acceptance case into a real test you write and run yourself. Put these tests under /tmp/pairtest/ (outside the repository) so they are not part of your change; you may also add normal tests to the repository's own test suite. Make all cases pass. If you think a case is wrong per the issue, say why instead of forcing it.
Remember to commit your work in the repository when you are done.

--- ADVICE ---
{advice}
--- END ADVICE ---"""

TO_A = """
Now write a report for the advisor: for EACH acceptance case number: the test you wrote (file and name), the command you ran, and PASS/FAIL. Then a short summary of what you changed. Paste the real test output tail as evidence. End with the line: TO ADVISOR"""

ADV_REVIEW = """Report from the developer:

{msg}

Still do NOT edit or run anything; you may read files (including /tmp/pairtest/ and `git diff`) to check the report.
Check: is every acceptance case covered by a real test that ran and passed? Does the change match the issue (read the diff)?
If good, reply with a one-line reason and VERDICT: APPROVED.
If not, list the concrete problems: missed or wrong cases, or up to 5 NEW concrete cases (input -> exact expected result) the change likely gets wrong. Then VERDICT: REVISE.
End with exactly one line: VERDICT: APPROVED  or  VERDICT: REVISE"""

def dirty(cname, root):
    return r.dexec(cname, f'cd "{root}" && git status --porcelain | head -20').stdout.strip()

def run_job(arm, tid, img):
    if arm != ARM: raise RuntimeError("run_advice only runs " + ARM)
    out = r.RES / arm / f"{tid}.json"
    tag = hashlib.md5((f"{arm}/{tid}" + os.environ.get("PB_REP", "")).encode()).hexdigest()[:10]
    home = r.WORK / arm / tid; shutil.rmtree(home, ignore_errors=True); home.mkdir(parents=True)
    os.chmod(home, 0o777)
    cname = r.start(f"pba-{tag}", img, home)
    rec = {"arm": arm, "id": tid, "started": time.time(), "reviews": [], "turns": 0, "advisor_violations": []}
    t0 = time.time(); dev = adv = None
    try:
        root = r.repo_root(cname)
        base0 = r.dexec(cname, f'cd "{root}" && git rev-parse HEAD').stdout.strip()   # 开局 HEAD，Hard 收卷以它为起点
        instr = (r.TASKS / tid / "instruction.md").read_text()
        deadline = t0 + r.BUDGET_S
        dev = r.agent("dsf", cname, root, "dev", deadline)
        adv = r.agent(ADV_MODEL, cname, root, "advisor", deadline)
        before = dirty(cname, root)
        advice = adv.turn(ADVISOR.format(root=root, instruction=instr))
        rec["advice"] = advice[:12000]; rec["advice_t"] = round(time.time() - t0)
        if dirty(cname, root) != before: rec["advisor_violations"].append("advice")   # 只记录，不回滚（回滚可能误伤）
        reply = dev.turn(r.GOAL.format(root=root, instruction=instr) + DEV_ADVICE.format(advice=advice)); rec["turns"] += 1
        for _ in range(r.MAX_CONT):
            if r.has_marker(reply, "FINISHED") or time.time() > deadline - 60: break
            reply = dev.turn(r.CONTINUE); rec["turns"] += 1
        msg = dev.turn(TO_A)
        for rnd in range(MAX_ADV_REVIEW):
            snap = dirty(cname, root)
            a = adv.turn(ADV_REVIEW.format(msg=msg))
            if dirty(cname, root) != snap: rec["advisor_violations"].append(f"review{rnd}")
            v = r.verdict(a)
            rec["reviews"].append({"t": round(time.time() - t0), "round": rnd, "verdict": v, "text": a[:3000]})
            if v == "APPROVED" or time.time() > deadline - 60: break
            msg = dev.turn(f"Advisor's reply:\n\n{a}\n\nTurn any new cases into tests under /tmp/pairtest/, fix what is wrong, re-run all your tests, commit." + TO_A)
        rec["elapsed_s"] = time.time() - t0
        cap = r.CAPTURE_DEEPSWE.format(root=root, base=r.base_commit(tid)) if r.BENCH == "deepswe" else r.CAPTURE.format(root=root).replace("git diff --cached --binary", f"git diff --cached --binary {base0}")
        rec["patch_bytes"] = int((r.dexec(cname, cap, timeout=300).stdout.strip() or "0").split()[-1])
    except r.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": adv.usage if adv else None}
        try: r.dexec(cname, "chmod -R a+rwX /pbhome", timeout=1800)
        except Exception as e: r.log("CHMOD SKIP", arm, tid[:60], repr(e)[:80])
        r.sh(["docker", "rm", "-f", cname])
    if rec.get("error"):
        r.log("NOT RECORDED", arm, tid[:60], rec["error"][:120]); return rec
    (r.RES / arm).mkdir(parents=True, exist_ok=True)
    shutil.copy(home / "model.patch", r.RES / arm / f"{tid}.patch")
    try: rec["grade"] = g = r.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))
    r.log(arm, tid[:70], "PASS" if rec["passed"] else "fail",
          f"{rec['elapsed_s']/60:.0f}min reviews={len(rec['reviews'])} viol={rec['advisor_violations']} patch={rec['patch_bytes']}B")
    return rec

r.run_job = run_job   # run_task 按模块全局名调用 run_job

if __name__ == "__main__":
    if not r.GW or not os.environ.get("PAIRBENCH_GATEWAY_TOKEN"): sys.exit("source ~/.pairbench.env first")
    ids = json.load(open(sys.argv[1])); par = int(sys.argv[2]) if len(sys.argv) > 2 else 4
    r.start_gw_relay()
    with ThreadPoolExecutor(par) as ex: list(ex.map(lambda t: r.run_task(t, [ARM]), ids))
    r.log("ADVICE DONE ALL", len(ids))

# 10-26 每题每个模型的 token 用量：输入（含命中缓存）、其中命中缓存、输出；写 tokens-<题集>.csv，再打印每种做法的每题均值/中位
# 取数口径和 report.py 一致（同一份 load：补跑优先、两次取好、剔除坏题；Opus 用会话记录重算的那份）
import csv, statistics as st, os
src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "report.py")).read()
exec(src[:src.index('strict = "--strict"')])
ROLE = {"dev": "写", "rev": "审"}
for res in ("results-deepswe", "results-hard"):
    R, _ = load(res); rows = []
    for (arm, tid), (d, _f) in sorted(R.items()):
        for role in ("dev", "rev"):
            u = opus_fixed(res, arm, d, role)
            if not u: continue
            rows.append({"arm": arm, "task": tid, "role": ROLE[role], "model": u["model"], "input": u["input"], "cached": u["cached"],
                         "output": u["output"], "usd": round(usd(u), 4), "passed": int(bool(d.get("passed")))})
    out = f"tokens-{res.split('-')[-1]}.csv"
    with open(out, "w", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=list(rows[0])); w.writeheader(); w.writerows(rows)
    print(f"\n=== {res}（{out}，{len(rows)} 行）  单位：千 token / 题")
    print("| 做法 | 角色 | 模型 | 题数 | 输入均值 | 输入中位 | 其中命中缓存均值 | 输出均值 | 输出中位 |")
    print("|---|---|---|---|---|---|---|---|---|")
    arms = [a for a in ORDER if any(r["arm"] == a for r in rows)] + sorted({r["arm"] for r in rows} - set(ORDER))
    for a in arms:
        for role in ("写", "审"):
            xs = [r for r in rows if r["arm"] == a and r["role"] == role]
            if not xs: continue
            k = lambda f: [r[f] / 1000 for r in xs]
            print(f"| `{a}` | {role} | {xs[0]['model']} | {len(xs)} | {st.mean(k('input')):,.0f} | {st.median(k('input')):,.0f} | "
                  f"{st.mean(k('cached')):,.0f} | {st.mean(k('output')):,.1f} | {st.median(k('output')):,.1f} |")