diff --git a/benchmarks/agentic/README.md b/benchmarks/agentic/README.md index c9d3ef640..2b2d11dea 100644 --- a/benchmarks/agentic/README.md +++ b/benchmarks/agentic/README.md @@ -112,7 +112,7 @@ less, not less-bloated. It judges only the tasks scored on size alone (`vibe-*`, `open-*`, `tmpl-*`). On the repo tasks the judge sees the agent's diff, not the whole seeded repo; an answer given only in chat is judged from -the chat; `puritanical:` markers are neutralized so the judge stays blind; runs killed at the timeout are +the chat; every arm's marker word (`puritanical:`, `ponytail:` and so on) is neutralized in all the judges so they stay blind; runs killed at the timeout are counted apart instead of scored 0. Without an API key it runs through the `claude` CLI login (no temperature setting there; the self-test gates it the same way). @@ -171,11 +171,35 @@ export PURITANICAL_PLUGIN_DIR=/abs/path/to/Puritanical export PONYTAIL_PLUGIN_DIR=/abs/path/to/ponytail export SUPERPOWERS_PLUGIN_DIR=/abs/path/to/superpowers export KARPATHY_CLAUDE_MD=/abs/path/to/karpathy/CLAUDE.md +export PURITANICAL_TMPL=/abs/path/to/full-stack-fastapi-template python run.py --selftest -python run.py --arms baseline,puritanical,ponytail,superpowers,karpathy --models opus --runs 5 --workers 6 +# pilot: 20 sessions, to check timeouts, cost and what each arm does without a human to ask +python run.py --task tmpl-be-count,vibe-todo,open-dataclass,csv-sum --arms baseline,puritanical,ponytail,superpowers,karpathy --models opus --workers 6 +# full run +python run.py --all --arms baseline,puritanical,ponytail,superpowers,karpathy --models opus --runs 5 --workers 6 +python complete.py --run runs/ +python judge.py --run runs/ +python compare.py runs/ --a puritanical --b baseline,ponytail,superpowers,karpathy --strict 5 ``` -`karpathy` is a `CLAUDE.md`, which the isolated cells switch off, so its text goes in through `--append-system-prompt`. Each run records every rival's remote and commit in `rivals.json`, and stops before any spend if a rival is not a git checkout; check them before trusting the run. Bash is off for every arm, and neither plugin arm has been checked for activation under that setup, so read the session transcript of one finished cell per arm to prove what was injected. Publish the result whichever way it falls, including any task where another arm wins. +`karpathy` is a `CLAUDE.md`, which the isolated cells switch off, so its text goes in through `--append-system-prompt`. + +Before any spend, a run stops if Puritanical, a rival or the template is unset or not a clean git checkout, or if the template is not at its pinned commit. It then starts one free session per arm, using a model name that does not exist: the session fails with a cost of 0, but only after the hooks have run and it has listed what it loaded. A plugin arm must load only its own plugin, whose `SessionStart` hook injected something, and any other arm must load none. Every cell repeats that check, and keeps its full log in `_claude.stream.jsonl`. `sources.json` records each commit, the Claude Code version and the planned cells. + +A session that fails with an error (a rate limit, say) retries with backoff and never counts against its arm; the summary's `failed` column counts any that still fail, and 5 in a row stop the run. Finish a stopped run with the same command plus `--resume runs/`, which refuses to continue if any source or planned cell changed. A cell killed at `--timeout` (300 seconds by default) is the arm's own failure, whatever it had written, and shows in the `killed` column; it reports no cost, so that arm's `$/pass` for the task is blank. + +Publish the result whichever way it falls, including any task where another arm wins. + +### Analysis plan + +Fixed before the first paid run. Any later change is published with the results, and says why. + +- **Comparisons:** `puritanical` against each other arm. +- **Primary metrics:** the share of cells that are both correct and safe, on the 20 surgical tasks; and, on the 27 open tasks, the share of cells the completeness judge scores 2 or higher, with an equal share decided by the fewer median lines of code in those cells. +- **Test:** a sign test over tasks, dropping ties, Bonferroni-adjusted across all primary tests: 8 with 4 other arms, or 6 if the pilot removes one. Run every comparison in one `compare.py` call, which applies this. A difference counts when the adjusted p is below 0.05. Everything else it prints is exploratory. +- **Claims:** each claim is for one metric against one arm, and needs a significant difference in Puritanical's favour. A non-significant result is reported as no detected difference, not as no harm. Better overall than an arm needs a significant win on both primary metrics. Best overall needs that against every arm. +- **Fixed design:** 5 runs per cell, with no early stopping and no task dropped after seeing results. Pilot cells do not count. +- **Exclusions:** the pilot can remove an arm before the full run only if, in at least 2 of its 3 open tasks (`tmpl-be-count`, `vibe-todo`, `open-dataclass`), the arm wrote no code and its final reply asked for a decision. The removal and its evidence are published with the results. ## What this can and cannot show diff --git a/benchmarks/agentic/compare.py b/benchmarks/agentic/compare.py index 819a6d2cc..482142be1 100644 --- a/benchmarks/agentic/compare.py +++ b/benchmarks/agentic/compare.py @@ -1,25 +1,46 @@ #!/usr/bin/env python3 """Paired arm comparison over tasks, with a sign test. No API calls. - python compare.py runs/ [runs/ ...] --a puritanical --b puritanical2 + python compare.py runs/ [runs/ ...] --a puritanical --b baseline,ponytail,karpathy -Every run dir's results.json (and answer.json, if answer.py ran) is merged. Each task is one -pair: the per-task median (or rate) of arm A against arm B. Ties are dropped and the sign test -asks whether B wins on more tasks than chance allows. Pair a candidate's quick run with the -full run's reference arms, so iterating on the ruleset only pays for the candidate's cells. +Every run dir's results.json (and answer.json, completeness.json and judge.json, where those passes +ran) is merged. Each task is one pair: the per-task median (or rate) of arm A against arm B. Ties +are dropped and the sign test asks whether B wins on more tasks than chance allows. Pair a +candidate's quick run with the full run's reference arms, so iterating on the ruleset only pays for +the candidate's cells. + +The primary metrics are the ones the README's analysis plan fixes in advance, each on its own tasks. +Their p values are Bonferroni-adjusted across every primary test in the call. All other p values +are raw and exploratory. shortcut: sign test over tasks, not a mixed model. Coarse, but honest with n=2-3 per cell. """ -import argparse, json, math, statistics +import argparse, json, math, statistics, sys, tempfile from collections import defaultdict from pathlib import Path +from tasks import TASKS + RUNS_DIR = Path(__file__).resolve().parent / "runs" +GATE_AT = 2 # completeness from which a cell's LOC counts: at least "mostly complete" + +def _pass(c): + if c.get("infra"): return None # a failed session says nothing about the arm + if "correct" in c: return int(bool(c["correct"] and c["safe"])) + return 0 if "error" in c else None # a crash fails; judge-only rows have no verdict + +def _loc_gated(c): + """LOC of a cell that did the job, so less code that also does less cannot win.""" + return c.get("total_loc") if (c.get("completeness") or 0) >= GATE_AT else None + # metric -> (how to read one cell, aggregate over a task's cells, True if higher is better) METRICS = { - "pass": (lambda c: int(bool(c["correct"] and c["safe"])) if "correct" in c else None, statistics.mean, True), + "pass": (_pass, statistics.mean, True), + "loc_gated": (_loc_gated, statistics.median, False), "loc": (lambda c: c.get("total_loc"), statistics.median, False), "gz_code": (lambda c: c.get("gz_code"), statistics.median, False), + "completeness": (lambda c: c.get("completeness"), statistics.mean, True), + "over_eng": (lambda c: c.get("over_engineering"), statistics.mean, False), "density": (lambda c: c.get("reply_density"), statistics.median, None), "cost": (lambda c: c.get("cost"), statistics.median, False), "out_tok": (lambda c: c.get("out_tokens"), statistics.median, False), @@ -30,18 +51,44 @@ "coverage": (lambda c: c.get("coverage"), statistics.mean, True), "clarity": (lambda c: c.get("clarity"), statistics.mean, True), } +# metric -> the tasks it decides on: hidden checks gate the surgical tasks, and only size and the +# completeness judge can tell anything on the open ones, where "correct" is close to always 1. +PRIMARY = {"pass": lambda t: not TASKS[t].get("open"), + "loc_gated": lambda t: bool(TASKS[t].get("open"))} def load(dirs): - """Cells from results.json plus answer-judge rows from answer.json, as one list of dicts. - Answer rows only carry the judge scores; per-task aggregation skips missing values.""" - rows = [] + """Cells from results.json, with each judge's score joined to its cell, and how many scores + matched no cell. Those stay as rows carrying only the scores; per-task aggregation skips missing values.""" + rows, unmatched = [], 0 for d in dirs: d = Path(d) if Path(d).exists() else RUNS_DIR / Path(d).name - rows += [c for c in json.loads((d / "results.json").read_text(encoding="utf-8"))["results"] + cells = [c for c in json.loads((d / "results.json").read_text(encoding="utf-8"))["results"] if "task" in c and "arm" in c] - if (d / "answer.json").exists(): - rows += json.loads((d / "answer.json").read_text(encoding="utf-8"))["scores"] - return rows + for c in cells: c["source"] = str(d.resolve()) # each run numbers its cells from zero + by_cell = {c["cell"]: c for c in cells if "cell" in c} + rows += cells + for name in ("completeness.json", "judge.json", "answer.json"): + if not (d / name).exists(): continue + for score in json.loads((d / name).read_text(encoding="utf-8"))["scores"]: + if score.get("cell") in by_cell: by_cell[score["cell"]].update({k: v for k, v in score.items() if k not in by_cell[score["cell"]]}) + else: + rows.append(score) + unmatched += 1 + return rows, unmatched + +def problems(by, arms, unmatched, runs): + """Why these results cannot support a published claim: missing cells or scores would otherwise + quietly change the primary metrics.""" + found = [f"{unmatched} judge scores match no cell"] if unmatched else [] + for arm in arms: + for t in TASKS: + cells = by[(t, arm)] + if len({(c.get("source"), c.get("cell")) for c in cells}) != runs or len(cells) != runs: # a duplicate must not hide a missing cell + found.append(f"{arm} / {t}: {len(cells)} cells, expected {runs} different ones") + if any(c.get("infra") for c in cells): found.append(f"{arm} / {t}: session errors left unresolved") + if TASKS[t].get("open") and any(c.get("completeness") is None and not c.get("infra") and not c.get("timed_out") for c in cells): + found.append(f"{arm} / {t}: a cell has no completeness score") + return found def sign_p(wins, losses): n = wins + losses @@ -49,45 +96,124 @@ def sign_p(wins, losses): k = min(wins, losses) return min(1.0, 2 * sum(math.comb(n, i) for i in range(k + 1)) / 2 ** n) +def agg(by, task, arm, metric): + get, fn, _ = METRICS[metric] + v = [x for x in (get(c) for c in by[(task, arm)]) if x is not None] + return fn(v) if v else None + +def _gated(by, task, arm): + """The share of an arm's cells that did the job, then fewer lines in those cells. Compared in + that order, so one tiny delivery out of five cannot beat five larger ones.""" + cells = [c for c in by[(task, arm)] if not c.get("infra")] + done = [v for v in map(_loc_gated, cells) if v is not None] + return (len(done) / len(cells), -statistics.median(done)) if done else (0, 0) + +def tally(by, tasks, a, b, metric): + """B's wins, losses and ties on one metric, with the median B/A ratio.""" + higher = METRICS[metric][2] + w = l = ties = 0; ratios = [] + for t in tasks: + if metric == "loc_gated": + x, y = _gated(by, t, a), _gated(by, t, b) + if x[1] and y[1]: ratios.append(y[1] / x[1]) + w += y > x; l += y < x; ties += y == x + continue + x, y = agg(by, t, a, metric), agg(by, t, b, metric) + if x is None or y is None: continue + if x: ratios.append(y / x) + if higher is None or x == y: ties += 1; continue + better = y > x if higher else y < x + w += better; l += not better + return {"wins": w, "losses": l, "ties": ties, "ratio": statistics.median(ratios) if ratios else None} + +def compare(by, a, b, tasks_filter, family): + tasks = sorted({t for t, x in by if x == b} & {t for t, x in by if x == a}) + if tasks_filter: tasks = [t for t in tasks if t in tasks_filter] + print(f"\n{a} (A) vs {b} (B) on {len(tasks)} tasks") + columns = ("pass", "loc", "completeness", "cost", "coverage", "clarity", "words") + print(f" {'task':22} " + " ".join(f"{m:>15}" for m in columns)) + for t in tasks: + row = [] + for m in columns: + x, y = agg(by, t, a, m), agg(by, t, b, m) + f = lambda v: "-" if v is None else (f"{v:.3g}") + row.append(f"{f(x):>7}>{f(y):<7}") + print(f" {t:22} " + " ".join(row)) + print(f"\n {'metric':12} {'B wins':>7} {'B loses':>8} {'ties':>5} {'p':>7} {'median B/A':>10}") + for m, (_, _, higher) in METRICS.items(): + primary = PRIMARY.get(m) + r = tally(by, [t for t in tasks if t in TASKS and primary(t)] if primary else tasks, a, b, m) + p = "-" if higher is None else f"{min(1.0, sign_p(r['wins'], r['losses']) * (family if primary else 1)):.3f}" + ratio = f"{r['ratio']:.2f}" if r["ratio"] is not None else "-" + print(f" {m:12} {r['wins']:>7} {r['losses']:>8} {r['ties']:>5} {p:>7}{' *' if primary else ' '} {ratio:>10}") + print(f" * primary: p adjusted across {family} tests. Other p values are raw and exploratory.") + +def selftest(): + """No API. The primary metrics must count a crash as a failure, drop a failed session, and give + a task to the arm whose code did the job more often before counting lines.""" + by = defaultdict(list) + def cell(t, arm, **kw): by[(t, arm)].append({"task": t, "arm": arm, **kw}) + cell("a", "x", correct=1, safe=1); cell("a", "y", error="boom") + cell("b", "x", correct=1, safe=1); cell("b", "y", correct=1, safe=1); cell("b", "y", infra=True) + cell("c", "x", total_loc=50, completeness=3); cell("c", "y", total_loc=5, completeness=0) + cell("d", "x", total_loc=50, completeness=3); cell("d", "y", total_loc=40, completeness=2) + for _ in range(5): cell("e", "x", total_loc=40, completeness=3) + cell("e", "y", total_loc=2, completeness=3) + for _ in range(4): cell("e", "y", total_loc=1, completeness=0) + passes, gated = tally(by, ["a", "b"], "x", "y", "pass"), tally(by, ["c", "d"], "x", "y", "loc_gated") + tiny = tally(by, ["e"], "x", "y", "loc_gated") + complete = defaultdict(list) + t0, open_task = next(iter(TASKS)), next(t for t in TASKS if TASKS[t].get("open")) + for t in TASKS: complete[(t, "x")] = [{"completeness": 3}] + gaps = defaultdict(list, complete) + gaps[(t0, "x")] = [] + gaps[(open_task, "x")] = [{}] + found = problems(gaps, ["x"], 1, 1) + pairs = defaultdict(list, {(t, "x"): [{"cell": f"{t}{r}", "completeness": 3} for r in (0, 1)] for t in TASKS}) + with tempfile.TemporaryDirectory() as tmp: + runs = [Path(tmp) / n for n in ("one", "two")] + for d in runs: + d.mkdir() + (d / "results.json").write_text(json.dumps({"results": [ + {"task": t, "arm": "x", "cell": t, "completeness": 3} for t in TASKS]}), encoding="utf-8") + def merged(*dirs): + rows, unmatched = load(dirs) + grouped = defaultdict(list) + for c in rows: grouped[(c["task"], c["arm"])].append(c) + return problems(grouped, ["x"], unmatched, 2) + separate, same_twice = merged(*runs), merged(runs[0], runs[0]) + checks = [("a crash fails and a failed session is dropped", (passes["wins"], passes["losses"], passes["ties"]) == (0, 1, 1)), + ("LOC counts only where the job was done", (gated["wins"], gated["losses"]) == (1, 1)), + ("one tiny delivery out of five loses to five larger ones", (tiny["wins"], tiny["losses"]) == (0, 1)), + ("complete results have no problems", problems(complete, ["x"], 0, 1) == []), + ("a missing cell, a missing score and an unmatched score are problems", len(found) == 3), + ("a duplicate cell is a problem", problems(pairs, ["x"], 0, 2) == [] and + len(problems(defaultdict(list, {**pairs, (t0, "x"): [{"cell": "c", "completeness": 3}] * 2}), ["x"], 0, 2)) == 1), + ("separate runs may reuse cell names, but a run counts once", separate == [] and len(same_twice) == len(TASKS)), + ("sign test", sign_p(9, 0) < 0.01 and math.isclose(sign_p(5, 5), 1.0)), + ("primary metrics split the tasks", all(any(PRIMARY[m](t) for t in TASKS) for m in PRIMARY) + and not any(PRIMARY["pass"](t) and PRIMARY["loc_gated"](t) for t in TASKS))] + for name, ok in checks: print(f"{'ok ' if ok else 'XX '} compare {name}") + return sum(not ok for _, ok in checks) + def main(): ap = argparse.ArgumentParser() ap.add_argument("runs", nargs="+") ap.add_argument("--a", default="puritanical") - ap.add_argument("--b", default="puritanical2") + ap.add_argument("--b", default="puritanical2", help="comma list: A is compared with each") ap.add_argument("--tasks", help="comma list; default: every task both arms ran") + ap.add_argument("--strict", type=int, metavar="RUNS", + help="exit unless every arm has all tasks at RUNS cells each, with every score joined") args = ap.parse_args() by = defaultdict(list) - for c in load(args.runs): by[(c["task"], c["arm"])].append(c) - tasks = sorted({t for t, a in by if a == args.b} & {t for t, a in by if a == args.a}) - if args.tasks: tasks = [t for t in tasks if t in args.tasks.split(",")] - - def agg(task, arm, m): - get, fn, _ = METRICS[m] - v = [x for x in (get(c) for c in by[(task, arm)]) if x is not None] - return fn(v) if v else None - - print(f"{args.a} (A) vs {args.b} (B) on {len(tasks)} tasks") - print(f" {'task':22} " + " ".join(f"{m:>15}" for m in ("pass", "loc", "cost", "coverage", "clarity", "words"))) - for t in tasks: - row = [] - for m in ("pass", "loc", "cost", "coverage", "clarity", "words"): - a, b = agg(t, args.a, m), agg(t, args.b, m) - f = lambda x: "-" if x is None else (f"{x:.3g}") - row.append(f"{f(a):>7}>{f(b):<7}") - print(f" {t:22} " + " ".join(row)) - print(f"\n {'metric':10} {'B wins':>7} {'B loses':>8} {'ties':>5} {'p (sign)':>9} {'median B/A':>11}") - for m, (_, _, higher) in METRICS.items(): - w = l = ties = 0; ratios = [] - for t in tasks: - a, b = agg(t, args.a, m), agg(t, args.b, m) - if a is None or b is None: continue - if a: ratios.append(b / a) - if higher is None or a == b: ties += 1; continue - better = b > a if higher else b < a - w += better; l += not better - r = f"{statistics.median(ratios):.2f}" if ratios else "-" - p = f"{sign_p(w, l):.3f}" if higher is not None else "-" - print(f" {m:10} {w:>7} {l:>8} {ties:>5} {p:>9} {r:>11}") + rows, unmatched = load(args.runs) + for c in rows: by[(c["task"], c["arm"])].append(c) + arms = args.b.split(",") + if args.strict: + found = problems(by, [args.a, *arms], unmatched, args.strict) + if found: sys.exit("not ready to publish:\n " + "\n ".join(found)) + for b in arms: + compare(by, args.a, b, args.tasks and args.tasks.split(","), len(arms) * len(PRIMARY)) if __name__ == "__main__": main() diff --git a/benchmarks/agentic/complete.py b/benchmarks/agentic/complete.py index e8c73c6c7..b08d18aa3 100644 --- a/benchmarks/agentic/complete.py +++ b/benchmarks/agentic/complete.py @@ -23,13 +23,13 @@ shortcut: reuses judge.py's HTTP/key/source plumbing instead of duplicating it -- one rubric param is the only delta between the two passes. """ -import argparse, json, re, subprocess, sys +import argparse, json, sys from concurrent.futures import ThreadPoolExecutor from collections import defaultdict from pathlib import Path from tasks import TASKS -from judge import load_key, source_text, judge_call, parse_score, RUNS_DIR, JUDGE_MODEL +from judge import load_key, submission, judge_call, parse_score, RUNS_DIR, JUDGE_MODEL SCORE_KEY = "completeness" FLAG_AT = 1 # cells scoring <= this are under-delivery (stub/partial) and get listed @@ -105,20 +105,6 @@ def selftest_offline(): print(f"\ncompleteness gate selftest (offline): {'valid' if passed else 'BROKEN'}") return 0 if passed else 1 -def submission(ws, tid): - """What the agent delivered. Fixture tasks: only its diff against the seeded repo, not the whole - repo. Open tasks may answer in chat with no file. Arm-revealing marker words are neutralized so - the judge stays blind.""" - if TASKS[tid].get("fixture"): - text = subprocess.run(["git", "-C", str(ws), "diff", "--cached", "-U10", "HEAD", "--", ".", ":(exclude)_*"], - capture_output=True, text=True).stdout - else: - text = source_text(ws) - if not text.strip() and (ws / "_claude.json").exists(): - try: text = json.loads((ws / "_claude.json").read_text(encoding="utf-8")).get("result", "") or "" - except ValueError: text = "" # the agent timed out: nothing delivered - return re.sub(r"\bpuritanical:", "note:", text, flags=re.I) - def run(run_dir, key, model=JUDGE_MODEL, arms=None, workers=6): run_dir = Path(run_dir) if not run_dir.exists(): run_dir = RUNS_DIR / run_dir.name diff --git a/benchmarks/agentic/judge.py b/benchmarks/agentic/judge.py index 7d05316e5..4a0f2469b 100644 --- a/benchmarks/agentic/judge.py +++ b/benchmarks/agentic/judge.py @@ -61,6 +61,22 @@ def source_text(workdir: Path): except Exception: continue return "\n\n".join(out) +def submission(ws, tid): + """What the agent delivered. Fixture tasks: only its diff against the seeded repo, not the whole + repo. Open tasks may answer in chat with no file.""" + if TASKS[tid].get("fixture"): + return subprocess.run(["git", "-C", str(ws), "diff", "--cached", "-U10", "HEAD", "--", ".", ":(exclude)_*"], + capture_output=True, text=True).stdout + text = source_text(ws) + if not text.strip() and (ws / "_claude.json").exists(): + try: text = json.loads((ws / "_claude.json").read_text(encoding="utf-8")).get("result", "") or "" + except ValueError: text = "" # the agent timed out: nothing delivered + return text + +def blind(text): + """Neutralise each arm's marker word, so no judge can tell which arm wrote what it reads.""" + return re.sub(r"\b(puritanical2?|ponytail|superpowers|karpathy|caveman):", "note:", text, flags=re.I) + def _cli_call(system, user, model): """No API key: the same judge through the headless claude CLI and the user's login. No tools, no plugins, no CLAUDE.md, empty cwd. shortcut: the CLI has no temperature flag, so CLI judging @@ -74,7 +90,7 @@ def _cli_call(system, user, model): return json.loads(r.stdout)["result"] def judge_call(task_prompt, files, key, retries=3, system=RUBRIC, model=JUDGE_MODEL, user=None, max_tokens=300): - user = user or f"TASK GIVEN TO THE AUTHOR:\n{task_prompt}\n\nFILES THEY WROTE:\n{files}" + user = blind(user or f"TASK GIVEN TO THE AUTHOR:\n{task_prompt}\n\nFILES THEY WROTE:\n{files}") body = json.dumps({"model": model, "max_tokens": max_tokens, "temperature": 0, "system": system, "messages": [{"role": "user", "content": user}]}).encode() for attempt in range(retries): @@ -159,8 +175,8 @@ def run(run_dir, key): cells.append((parts[0], parts[1], parts[2], ws)) print(f"judging {len(cells)} workspaces with {JUDGE_MODEL} ...") for i, (tid, arm, model, ws) in enumerate(cells, 1): - s = parse_score(judge_call(TASKS[tid]["prompt"], source_text(ws), key)) or {"over_engineering": None} - rec = {"task": tid, "arm": arm, "model": model, "over_engineering": s.get("over_engineering"), + s = parse_score(judge_call(TASKS[tid]["prompt"], submission(ws, tid), key)) or {"over_engineering": None} + rec = {"task": tid, "arm": arm, "model": model, "cell": ws.name, "over_engineering": s.get("over_engineering"), "why": s.get("why", ""), "cite": s.get("cite", "")} scored.append(rec) if i % 25 == 0 or i == len(cells): print(f" [{i}/{len(cells)}]", flush=True) @@ -186,8 +202,7 @@ def main(): ap.add_argument("--selftest", action="store_true") ap.add_argument("--run", help="run dir to judge") args = ap.parse_args() - key = load_key() - if not key: sys.exit("no ANTHROPIC_API_KEY (.env or env)") + key = load_key() # none: judge_call uses the claude CLI login if args.selftest: sys.exit(selftest(key)) if args.run: if selftest(key): sys.exit("judge not trustworthy; refusing to judge the matrix") diff --git a/benchmarks/agentic/run.py b/benchmarks/agentic/run.py index 8112cb0e8..ca858ad7e 100644 --- a/benchmarks/agentic/run.py +++ b/benchmarks/agentic/run.py @@ -19,13 +19,17 @@ Recompute metrics + aggregate from kept workspaces. No API. Use after changing a metric or scorer so you never pay the API twice for a measurement tweak. + python run.py --all --models opus --runs 5 --resume runs/ + Finish a stopped run with the same arguments: only unfinished or failed cells run. + shortcut: the claude CLI is the harness (already installed, we run inside it). No SDK dependency. The CLI's JSON output already carries cost/tokens/duration/permission_denials. """ -import argparse, concurrent.futures, datetime, json, os, re, shutil, signal, statistics, subprocess, sys, tempfile, zlib +import argparse, concurrent.futures, datetime, json, os, re, shutil, signal, statistics, subprocess, sys, tempfile, time, zlib from collections import defaultdict from pathlib import Path +import compare from tasks import QUICK, TASKS ROOT = Path(__file__).resolve().parents[2] @@ -59,6 +63,7 @@ def _rules_file(env): # user's globally-enabled plugins (--setting-sources project,local) and load one plugin from its # cache dir (--plugin-dir). The smoke test verifies activation by output style. PLUGIN_ARMS = ("puritanical", "caveman", "puritanical2", "ponytail", "superpowers") # arms activated via --plugin-dir (vs raw --append prompts) +PLUGIN_NAMES = {"puritanical2": "puritanical"} # arms whose plugin is not named after them PLUGIN_CACHE = Path.home() / ".claude" / "plugins" / "cache" def _plugin_dir(name): @@ -74,25 +79,48 @@ def _plugin_dir(name): sys.exit(f"{name} plugin dir not found under {base}; install the plugin or set {name.upper()}_PLUGIN_DIR") return str(versions[-1]) # latest version dir; not pinned to one machine's hash -RIVALS = {"ponytail": "PONYTAIL_PLUGIN_DIR", "superpowers": "SUPERPOWERS_PLUGIN_DIR", "karpathy": "KARPATHY_CLAUDE_MD"} - -def _rival_sources(arms): - """Record which remote and commit each rival arm runs, so a result can be traced and a wrong - path shows up. Stops before any spend when a rival is unset or not a git checkout.""" +PINNED = {"puritanical": "PURITANICAL_PLUGIN_DIR", "ponytail": "PONYTAIL_PLUGIN_DIR", + "superpowers": "SUPERPOWERS_PLUGIN_DIR", "karpathy": "KARPATHY_CLAUDE_MD"} +TMPL_COMMIT = "cd83fc1" + +def _checkout(label, path): + """The remote and commit of a clean git checkout. Exits otherwise: a result has to be traceable.""" + checkout = Path(path) if Path(path).is_dir() else Path(path).parent + def git(*args): + r = subprocess.run(["git", "-C", str(checkout), *args], capture_output=True, text=True) + if r.returncode: sys.exit(f"{label}: {checkout} is not a git checkout with an origin") + return r.stdout.strip() + source = {"path": str(path), "remote": git("remote", "get-url", "origin"), "commit": git("rev-parse", "HEAD")} + if git("status", "--porcelain"): sys.exit(f"{label}: {checkout} has uncommitted changes") + return source + +def _fixture_path(name): + """A fixture is an absolute path or a name under fixtures/.""" + fx = Path(name) + return fx if fx.is_absolute() else Path(__file__).resolve().parent / "fixtures" / name + +def _sources(arms, task_ids): + """Record what each pinned arm and the template run from. Stops before any spend when one is + unset, not a clean git checkout, or (the template) not at its pinned commit.""" sources = {} for arm in arms: - if arm not in RIVALS: continue - path = os.environ.get(RIVALS[arm]) - if not path: sys.exit(f"set {RIVALS[arm]} for the {arm} arm (see Head to head in the README)") - checkout = Path(path) if Path(path).is_dir() else Path(path).parent - def git(*args): - r = subprocess.run(["git", "-C", str(checkout), *args], capture_output=True, text=True) - if r.returncode: sys.exit(f"{RIVALS[arm]}: {checkout} is not a git checkout with an origin") - return r.stdout.strip() - sources[arm] = {"path": path, "remote": git("remote", "get-url", "origin"), "commit": git("rev-parse", "HEAD")} + if arm not in PINNED: continue + path = os.environ.get(PINNED[arm]) + if not path: sys.exit(f"set {PINNED[arm]} for the {arm} arm (see Head to head in the README)") + sources[arm] = _checkout(PINNED[arm], path) + for name in {TASKS[t]["fixture"] for t in task_ids if TASKS[t].get("fixture")}: + path = _fixture_path(name) + if not path.is_dir(): sys.exit(f"template not found at {path}; see Reproduce in the README") + sources["template"] = _checkout("template", path) + if not sources["template"]["commit"].startswith(TMPL_COMMIT): + sys.exit(f"template: {path} is not at the pinned commit {TMPL_COMMIT}") return sources CELL_TIMEOUT = 300 # seconds per cell; a hung agent is force-killed (process tree) so the pool can't freeze +SCORE_TIMEOUT = 120 # seconds to score a cell; produced code runs during scoring and can hang +RETRY_WAITS = (30, 120, 300) # seconds before each retry of a session that failed with an error +MAX_FAILED_IN_A_ROW = 5 # cells still failing after their retries: stop rather than burn the queue +PREFLIGHT_MODEL = "claude-preflight-no-such-model" # Added to every arm's system prompt, identically. We measure code PRODUCTION, not execution: agents # write the implementation and stop. No live verification -- earlier attempts had agents open a browser, @@ -240,6 +268,8 @@ def selftest(): ref = task[kind] if isinstance(task[kind], dict) else {task["file"]: task[kind]} for fn, content in {**task.get("seed", {}), **ref}.items(): # seed, then the ref on top _write(Path(d) / fn, content) + for fn in ("_claude.json", "_claude.stream.jsonl", "_claude.stderr.txt", "_score.json"): + _write(Path(d) / fn, "") # what a live cell leaves behind r = task["score"](Path(d)) ok = (r["correct"] == 1 and r["safe"] == 1) if kind == "good" else (r[axis] == 0) print(f"{'ok ' if ok else 'XX '} {tid:12} {kind:4} correct={r['correct']} " @@ -249,11 +279,116 @@ def selftest(): ok = row["safe_rate"] == 0 and row["correct_rate"] == 0 print(f"{'ok ' if ok else 'XX '} aggregate a crashed cell counts as failed") failures += 0 if ok else 1 + failures += _selftest_aggregate() + failures += _selftest_session() + failures += _selftest_activation() + failures += _selftest_retries() + failures += _selftest_isolated_scoring() failures += _selftest_plugin_dir() failures += _selftest_kill() + failures += compare.selftest() print(f"\nselftest: {'all instruments valid' if not failures else str(failures) + ' BROKEN'}") return failures +def _report(name, ok): + print(f"{'ok ' if ok else 'XX '} {name}") + return 0 if ok else 1 + +def _selftest_aggregate(): + """A session error says nothing about the arm, so it leaves the rates. A timeout is the arm's + own failure, and its spend is unknown, so cost per pass must not look cheaper for it.""" + ok_cell = {"task": "t", "arm": "a", "model": "m", "correct": 1, "safe": 1, "cost": 1.0} + row = aggregate([ok_cell, {"task": "t", "arm": "a", "model": "m", "infra": True, "infra_reason": "limit"}])[0] + fails = _report("aggregate a session error leaves the rates and is counted apart", + row["n"] == 1 and row["correct_rate"] == 1 and row["infra"] == 1) + only = aggregate([{"task": "t", "arm": "a", "model": "m", "infra": True}])[0] + fails += _report("aggregate an arm with only session errors has no rates", only["n"] == 0 and only["correct_rate"] == 0) + row = aggregate([ok_cell, {"task": "t", "arm": "a", "model": "m", "correct": 0, "safe": 0, "timed_out": True}])[0] + return fails + _report("aggregate a timeout fails, and hides cost per pass", + row["n"] == 2 and abs(row["correct_rate"] - 0.5) < 1e-9 and row["timeouts"] == 1 and row["cost_per_pass"] is None) + +def _selftest_session(): + fails = 0 + with tempfile.TemporaryDirectory() as d: + ws, log = Path(d) / "ws", Path(d) / "log" + ws.mkdir() + log.write_text('{"type":"system"}\nnot json\n{"type":"result","is_error":true,"result":"limit reached"}\n', encoding="utf-8") + _keep_result(log, ws) + fails += _report("session an error result is a session error, with its reason", + _session_status(ws) == ("infra", "limit reached")) + log.write_text('{"type":"result","is_error":false,"result":"done"}\n', encoding="utf-8") + _keep_result(log, ws) + fails += _report("session a clean result is fine", _session_status(ws) == ("ok", "")) + log.write_text("", encoding="utf-8") + _keep_result(log, ws) + fails += _report("session no result is a session error", _session_status(ws)[0] == "infra") + (ws / "_claude.stderr.txt").write_text("[KILLED after 300s timeout]", encoding="utf-8") + fails += _report("session a killed session is a timeout", _session_status(ws)[0] == "timeout") + return fails + +def _selftest_activation(): + builtin = {"name": "cc-plugin-telemetry", "path": "builtin"} + inline = {"name": "ponytail", "path": "/x/ponytail"} + hook = {"subtype": "hook_response", "hook_event": "SessionStart", "exit_code": 0, "stdout": "the rules"} + cases = [("baseline", [{"subtype": "init", "plugins": [builtin]}], None), + ("baseline", [{"subtype": "init", "plugins": [builtin, inline]}, hook], "plugins"), + ("ponytail", [hook, {"subtype": "init", "plugins": [builtin, inline]}], None), + ("ponytail", [{"subtype": "init", "plugins": [builtin]}], "plugins"), + ("ponytail", [hook, {"subtype": "init", "plugins": [{"name": "puritanical", "path": "/x/p"}]}], "plugins"), + ("ponytail", [{**hook, "stdout": ""}, {"subtype": "init", "plugins": [inline]}], "hook"), + ("ponytail", [{**hook, "exit_code": 1}, {"subtype": "init", "plugins": [inline]}], "hook"), + ("concise", [{"subtype": "init", "plugins": [builtin], "output_style": "default"}], "style"), + ("concise", [{"subtype": "init", "plugins": [builtin], "output_style": "Concise"}], None), + ("baseline", [], "init")] + with tempfile.TemporaryDirectory() as d: + log = Path(d) / "log" + def verdict(arm, events): + log.write_text("\n".join(json.dumps(e) for e in events), encoding="utf-8") + return _activation_problem(arm, log) + ok = all((verdict(arm, events) is None) if want is None else (want in (verdict(arm, events) or "")) + for arm, events, want in cases) + return _report("activation each arm has its rules, and only its rules", ok) + +def _selftest_retries(): + calls, waits = [], [] + def flaky(): + calls.append(1) + return {"infra": len(calls) < 3} + ok = not _with_retries(flaky, waits.append).get("infra") and waits == list(RETRY_WAITS[:2]) + calls.clear(); waits.clear() + ok = ok and _with_retries(lambda: calls.append(1) or {"infra": True}, waits.append).get("infra") \ + and len(calls) == len(RETRY_WAITS) + 1 + return _report("retries a session error retries, then is reported", ok) + +def _selftest_isolated_scoring(): + """Scoring one cell must not see another cell's code, and produced code that hangs or exits + must fail that cell without taking the run with it.""" + fails = 0 + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as d: + finished = '{"is_error": false, "result": "done"}' + delivered, missing = Path(d) / "reuse-slug__a__m__0", Path(d) / "reuse-slug__b__m__0" + task = TASKS["reuse-slug"] + good = task["good"] if isinstance(task["good"], dict) else {task["file"]: task["good"]} + for fn, content in {**task.get("seed", {}), **good}.items(): + _write(delivered / fn, content) + _write(delivered / "_claude.json", finished) + _write(missing / "_claude.json", finished) + first = score_isolated("reuse-slug", "a", "m", delivered) + second = score_isolated("reuse-slug", "b", "m", missing) + fails += _report("scoring a cell cannot import another cell's code", + first["correct"] == 1 and second["correct"] == 0) + _write(delivered / "_claude.stderr.txt", "[KILLED after 300s timeout]") + killed = score_isolated("reuse-slug", "a", "m", delivered) + fails += _report("scoring a killed cell fails even with good code", + killed["correct"] == 0 and killed["safe"] == 0 and killed["timed_out"]) + for label, code in (("hangs", "import time\ntime.sleep(60)\n"), ("exits", "import os\nos._exit(3)\n")): + ws = Path(d) / f"cache__{label}__m__0" + _write(ws / TASKS["cache"]["file"], code) + _write(ws / "_claude.json", finished) + r = score_isolated("cache", label, "m", ws, timeout=5) + fails += _report(f"scoring produced code that {label} fails its cell", r["correct"] == 0 and r["safe"] == 0) + return fails + def _selftest_plugin_dir(): """Plugin-dir resolution must be portable: env override wins, and a missing install fails loudly (sys.exit) instead of silently passing a non-existent path to --plugin-dir.""" @@ -278,21 +413,38 @@ def _selftest_plugin_dir(): ok_miss = True print(f"{'ok ' if ok_miss else 'XX '} plugin_dir miss clear error (sys.exit)") fails += 0 if ok_miss else 1 - previous = os.environ.pop("PONYTAIL_PLUGIN_DIR", None) - try: - _rival_sources(["ponytail"]); ok_rival = False # reached only if it did NOT exit -> broken - except SystemExit: - ok_rival = True + def exits(fn): + try: + fn(); return False + except SystemExit: + return True + previous = {e: os.environ.pop(e, None) for e in ("PONYTAIL_PLUGIN_DIR", "PURITANICAL_PLUGIN_DIR")} + ok_rival = exits(lambda: _sources(["ponytail"], [])) and exits(lambda: _sources(["puritanical"], [])) with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as fixture: # not ROOT: a source archive has no git for args in (["init", "-q"], ["remote", "add", "origin", "https://example.invalid/rival.git"], ["-c", "user.name=t", "-c", "user.email=t@example.invalid", "commit", "-q", "--allow-empty", "-m", "t"]): subprocess.run(["git", "-C", fixture, *args], check=True, capture_output=True) os.environ["PONYTAIL_PLUGIN_DIR"] = fixture - ok_rival = ok_rival and len(_rival_sources(["ponytail"])["ponytail"]["commit"]) == 40 and _rival_sources(["baseline"]) == {} - if previous is None: del os.environ["PONYTAIL_PLUGIN_DIR"] - else: os.environ["PONYTAIL_PLUGIN_DIR"] = previous - print(f"{'ok ' if ok_rival else 'XX '} rival_sources unset exits, checkout records its commit") - return fails + (0 if ok_rival else 1) + ok_rival = ok_rival and len(_sources(["ponytail"], [])["ponytail"]["commit"]) == 40 and _sources(["baseline"], []) == {} + (Path(fixture) / "edit.txt").write_text("x", encoding="utf-8") + ok_dirty = exits(lambda: _sources(["ponytail"], [])) + for env, value in previous.items(): + os.environ.pop(env, None) + if value is not None: os.environ[env] = value + fails += _report("sources unset exits, checkout records its commit", ok_rival) + fails += _report("sources uncommitted changes exit", ok_dirty) + tmpl = TASKS["tmpl-be-count"]["fixture"] + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as fixture: + for args in (["init", "-q"], ["remote", "add", "origin", "https://example.invalid/template.git"], + ["-c", "user.name=t", "-c", "user.email=t@example.invalid", "commit", "-q", "--allow-empty", "-m", "t"]): + subprocess.run(["git", "-C", fixture, *args], check=True, capture_output=True) + TASKS["tmpl-be-count"]["fixture"] = fixture # a real checkout, at the wrong commit + ok_wrong = exits(lambda: _sources([], ["tmpl-be-count"])) + TASKS["tmpl-be-count"]["fixture"] = str(Path(tempfile.gettempdir()) / "puritanical-no-such-template") + ok_missing = exits(lambda: _sources([], ["tmpl-be-count"])) + TASKS["tmpl-be-count"]["fixture"] = tmpl + fails += _report("sources a missing or unpinned template exits", ok_wrong and ok_missing) + return fails def _tree_kill(proc): """Tree-kill one timed-out cell, never a blanket kill (that would also take down this @@ -326,23 +478,73 @@ def chat_code_loc(text): if not s.startswith(("#", "//", "*", "/*", "*/")): code += 1 return total, code +def _session_meta(workdir: Path): + """What the CLI reported about the session, and the reply. Empty when it reported nothing.""" + try: + j = json.loads((workdir / "_claude.json").read_text(encoding="utf-8")) + u = j.get("usage") or {} + result_text = j.get("result", "") + return {"cost": j.get("total_cost_usd"), "duration_ms": j.get("duration_ms"), + "turns": j.get("num_turns"), "denials": len(j.get("permission_denials") or []), + "out_tokens": u.get("output_tokens"), "in_tokens": u.get("input_tokens"), + "cache_tokens": (u.get("cache_read_input_tokens") or 0) + (u.get("cache_creation_input_tokens") or 0), + "reply_words": len(result_text.split()), + # share of the reply that is not redundancy: compressed / raw. Filler and repeated + # recaps compress well; dense information does not. Reported, not optimized blindly. + "reply_density": round(gz(result_text) / max(1, len(result_text.encode())), 3)}, result_text + except Exception: + return {}, "" + +def _keep_result(log: Path, workdir: Path): + """The stream's final result event becomes _claude.json, which the judges read. No result + leaves it empty, which they read as nothing delivered.""" + result = "" + for line in log.read_text(encoding="utf-8", errors="ignore").splitlines(): + try: event = json.loads(line) + except ValueError: continue + if event.get("type") == "result": result = line + (workdir / "_claude.json").write_text(result, encoding="utf-8") + +def _session_status(workdir: Path): + """ok, timeout (the arm ran out of time) or infra (the session failed, which says nothing about the arm).""" + stderr = workdir / "_claude.stderr.txt" + if stderr.exists() and "[KILLED after" in stderr.read_text(encoding="utf-8", errors="ignore"): + return "timeout", "" + try: j = json.loads((workdir / "_claude.json").read_text(encoding="utf-8")) + except (OSError, ValueError): return "infra", "the session reported no result" + if j.get("is_error") or "error" in j: + return "infra", str(j.get("result") or j.get("error"))[:200] + return "ok", "" + +def _activation_problem(arm, log: Path): + """What is wrong with the rules this session loaded, or None. A plugin arm must load only its own + plugin, whose SessionStart hook injected something; any other arm must load none.""" + events = [] + for line in log.read_text(encoding="utf-8", errors="ignore").splitlines(): + try: events.append(json.loads(line)) + except ValueError: continue + init = next((e for e in events if e.get("subtype") == "init"), None) + if not init: return "the session never started (no init event)" + plugins = [p.get("name") for p in init.get("plugins") or [] if p.get("path") != "builtin"] + expected = [PLUGIN_NAMES.get(arm, arm)] if arm in PLUGIN_ARMS else [] + if plugins != expected: return f"unexpected plugins loaded: {plugins}, expected {expected}" + if arm in PLUGIN_ARMS and not any(e.get("subtype") == "hook_response" and e.get("hook_event") == "SessionStart" + and e.get("exit_code") == 0 and (e.get("stdout") or "").strip() for e in events): + return f"the SessionStart hook of {plugins[0]} injected nothing" + if arm in SETTINGS_ARMS and init.get("output_style") != json.loads(SETTINGS_ARMS[arm])["outputStyle"]: + return f"output style is {init.get('output_style')}" + return None + def score_workspace(task_id, arm, model, workdir: Path): - meta, result_text = {}, "" - cj = workdir / "_claude.json" - if cj.exists(): - try: - j = json.loads(cj.read_text(encoding="utf-8")) - u = j.get("usage") or {} - meta = {"cost": j.get("total_cost_usd"), "duration_ms": j.get("duration_ms"), - "turns": j.get("num_turns"), "denials": len(j.get("permission_denials") or []), - "out_tokens": u.get("output_tokens"), "in_tokens": u.get("input_tokens"), - "cache_tokens": (u.get("cache_read_input_tokens") or 0) + (u.get("cache_creation_input_tokens") or 0)} - result_text = j.get("result", "") - meta["reply_words"] = len(result_text.split()) - # share of the reply that is not redundancy: compressed / raw. Filler and repeated - # recaps compress well; dense information does not. Reported, not optimized blindly. - meta["reply_density"] = round(gz(result_text) / max(1, len(result_text.encode())), 3) - except Exception: pass + meta, result_text = _session_meta(workdir) + cell = {"task": task_id, "arm": arm, "model": model, "cell": workdir.name} + status, reason = _session_status(workdir) + log = workdir / "_claude.stream.jsonl" + if status != "timeout" and log.exists(): # a kept run from before this log existed has nothing to check + problem = _activation_problem(arm, log) + if status == "ok" and problem: status, reason = "infra", problem + if status == "infra": return {**cell, "infra": True, "infra_reason": reason, **meta} + if status == "timeout": cell["timed_out"] = True surgical = not TASKS[task_id].get("open") and not TASKS[task_id].get("fixture") diffed = TASKS[task_id].get("fixture") or TASKS[task_id].get("diff") stats = git_diff_stats(workdir) if diffed else code_stats(workdir, selfcheck_as_test=surgical) @@ -355,28 +557,86 @@ def score_workspace(task_id, arm, model, workdir: Path): sc = {"correct": 1 if stats.get("total_loc", 0) > 0 else 0, "safe": 1, "reason": "git-diff"} else: sc = TASKS[task_id]["score"](workdir) - return {"task": task_id, "arm": arm, "model": model, **sc, **stats, **meta} + if status == "timeout": sc = {**sc, "correct": 0, "safe": 0} # partial files must not pass + return {**cell, **sc, **stats, **meta} + +def score_isolated(task_id, arm, model, workdir: Path, timeout=SCORE_TIMEOUT): + """Score in a child process. Scoring runs the agent's code: in-process, _import_pkg's sys.path + lets one cell import another's files, and code that hangs or exits stops the whole run.""" + workdir = Path(workdir).resolve() + out = workdir / "_score.json" + out.unlink(missing_ok=True) + proc = subprocess.Popen([sys.executable, str(Path(__file__).resolve()), "--score-cell", task_id, arm, model], + cwd=str(workdir), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + start_new_session=(os.name != "nt")) + try: + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + _tree_kill(proc) + try: proc.wait(timeout=15) + except Exception: pass + try: + return json.loads(out.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {"task": task_id, "arm": arm, "model": model, "cell": workdir.name, "correct": 0, "safe": 0, + "reason": "scoring hung or exited", **_session_meta(workdir)[0]} def _base_cmd(claude, prompt, model): return [claude, "-p", prompt, "--model", MODELS.get(model, model), - "--permission-mode", "bypassPermissions", "--output-format", "json", + "--permission-mode", "bypassPermissions", "--output-format", "stream-json", "--verbose", + "--include-hook-events", "--no-session-persistence", "--setting-sources", "project,local", "--strict-mcp-config", "--disallowedTools", "Bash"] -def warm_cache(model): - """One throwaway call before the pool. Without it the first parallel cells each pay the cold - prompt-cache write of the shared prefix (smoke run: ~$0.32 vs ~$0.10 per cell, in both arms), - which is ordering noise, not an arm effect. Cells running back to back keep the cache warm.""" - with tempfile.TemporaryDirectory() as d: - subprocess.run(_base_cmd(shutil.which("claude"), "Reply with: ok", model) + ["--append-system-prompt", NO_RUN], - cwd=d, capture_output=True, timeout=CELL_TIMEOUT, - env={**os.environ, "CLAUDE_CODE_DISABLE_CLAUDE_MDS": "1", "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1"}) +def _arm_cmd(arm, prompt, model): + # Skills are PLUGINS (SessionStart hook); --append of the SKILL text does NOT activate them. + # Exclude the user's globally-enabled plugins for every arm, then load exactly the one this arm + # needs from its cache dir. baseline loads none; yagni-oneliner is a raw prompt so it uses --append. + # No live verification (see NO_RUN): --strict-mcp-config drops all MCP servers so there is no browser + # tool, and --disallowedTools Bash blocks running a server/db/npm. An agent writes with + # Read/Write/Edit/Glob/Grep and stops -- no login wall, no browser thrash. We measure code, not execution. + claude = shutil.which("claude") + if not claude: sys.exit("claude CLI not found on PATH") + cmd = _base_cmd(claude, prompt, model) + append = NO_RUN # all arms get NO_RUN, identically + if arm in PLUGIN_ARMS: + cmd += ["--plugin-dir", _plugin_dir(arm)] # real activation of exactly one plugin + elif arm in SETTINGS_ARMS: + cmd += ["--settings", SETTINGS_ARMS[arm]] + else: + extra = ARMS[arm]() # baseline -> None; yagni-oneliner -> the prompt + if extra: append = extra + "\n\n" + NO_RUN + return cmd + ["--append-system-prompt", append] + +# Cells live under this repo, so without these every arm (baseline too) also loads the +# repo's AGENTS.md, which is the puritanical ruleset, plus the user's auto-memory. +CELL_ENV = {**os.environ, "CLAUDE_CODE_DISABLE_CLAUDE_MDS": "1", "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1"} -def run_cell(task_id, arm, model, workdir: Path): +def preflight(arms): + """Prove each arm activates, before any spend. An unknown model fails with a cost of 0, but only + after the hooks have run and the session has reported what it loaded.""" + for arm in arms: + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as d: + log = Path(d) / "preflight.jsonl" + with open(log, "wb") as so: + subprocess.run(_arm_cmd(arm, "ok", PREFLIGHT_MODEL), cwd=d, stdout=so, stderr=subprocess.DEVNULL, + env=CELL_ENV, timeout=120) + problem = _activation_problem(arm, log) + if problem: sys.exit(f"preflight: the {arm} arm did not activate: {problem}") + print(f"preflight: {arm} activates", flush=True) + +def warm_cache(arm, model): + """One throwaway call per arm before the pool. Without it the first parallel cells each pay the cold + prompt-cache write of that arm's prefix (smoke run: ~$0.32 vs ~$0.10 per cell), which is ordering + noise, not an arm effect. Cells running back to back keep the cache warm.""" + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as d: + subprocess.run(_arm_cmd(arm, "Reply with: ok", model), cwd=d, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, timeout=CELL_TIMEOUT, env=CELL_ENV) + +def run_session(task_id, arm, model, workdir: Path): task = TASKS[task_id] if task.get("fixture"): # copy a real repo in; record what was seeded - fx = Path(task["fixture"]) # absolute path, or a name under fixtures/ - if not fx.is_absolute(): fx = Path(__file__).resolve().parent / "fixtures" / task["fixture"] + fx = _fixture_path(task["fixture"]) shutil.copytree(fx, workdir, dirs_exist_ok=True, ignore=shutil.ignore_patterns("node_modules", ".git", "build", "dist", "dist-ssr", ".vite", "*.log", "__pycache__", @@ -391,59 +651,54 @@ def run_cell(task_id, arm, model, workdir: Path): for fn, content in task.get("seed", {}).items(): _write(workdir / fn, content) if task.get("fixture") or task.get("diff"): _git_snapshot(workdir) # baseline commit -> diff the agent's changes - claude = shutil.which("claude") - if not claude: sys.exit("claude CLI not found on PATH") - # Skills are PLUGINS (SessionStart hook); --append of the SKILL text does NOT activate them. - # Exclude the user's globally-enabled plugins for every arm, then load exactly the one this arm - # needs from its cache dir. baseline loads none; yagni-oneliner is a raw prompt so it uses --append. - # No live verification (see NO_RUN): --strict-mcp-config drops all MCP servers so there is no browser - # tool, and --disallowedTools Bash blocks running a server/db/npm. An agent writes with - # Read/Write/Edit/Glob/Grep and stops -- no login wall, no browser thrash. We measure code, not execution. - cmd = _base_cmd(claude, task["prompt"], model) - append = NO_RUN # all arms get NO_RUN, identically - if arm in PLUGIN_ARMS: - cmd += ["--plugin-dir", _plugin_dir(arm)] # real activation of exactly one plugin - elif arm in SETTINGS_ARMS: - cmd += ["--settings", SETTINGS_ARMS[arm]] - else: - extra = ARMS[arm]() # baseline -> None; yagni-oneliner -> the prompt - if extra: append = extra + "\n\n" + NO_RUN - cmd += ["--append-system-prompt", append] - out_path, err_path = workdir / "_claude.json", workdir / "_claude.stderr.txt" + cmd = _arm_cmd(arm, task["prompt"], model) + # Beside the workspace while the agent runs, so it cannot read its own log (with an arm's rules in it). + log, err = workdir.parent / f"{workdir.name}.stream.jsonl", workdir.parent / f"{workdir.name}.stderr.txt" # stdout -> file, never a PIPE: on Windows a hung agent's child processes can hold a stdout PIPE # open forever, so subprocess.run(timeout=) never fires and the worker freezes. Writing to a file # lets proc.wait(timeout) return reliably; on timeout _tree_kill ends ONLY this cell's process # tree -- never a blanket kill, which would also take down this Claude Code session. - try: - with open(out_path, "wb") as so, open(err_path, "wb") as se: - # Cells live under this repo, so without these every arm (baseline too) also loads the - # repo's AGENTS.md, which is the puritanical ruleset, plus the user's auto-memory. - env = {**os.environ, "CLAUDE_CODE_DISABLE_CLAUDE_MDS": "1", "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1"} - proc = subprocess.Popen(cmd, cwd=str(workdir), stdout=so, stderr=se, env=env, - start_new_session=(os.name != "nt")) - try: - proc.wait(timeout=CELL_TIMEOUT) - except subprocess.TimeoutExpired: - _tree_kill(proc) - try: proc.wait(timeout=15) - except Exception: pass - se.write(f"\n[KILLED after {CELL_TIMEOUT}s timeout]".encode()) - except Exception as e: - out_path.write_text(json.dumps({"error": str(e)[:300]}), encoding="utf-8") - return score_workspace(task_id, arm, model, workdir) + with open(log, "wb") as so, open(err, "wb") as se: + proc = subprocess.Popen(cmd, cwd=str(workdir), stdout=so, stderr=se, env=CELL_ENV, + start_new_session=(os.name != "nt")) + try: + proc.wait(timeout=CELL_TIMEOUT) + except subprocess.TimeoutExpired: + _tree_kill(proc) + try: proc.wait(timeout=15) + except Exception: pass + se.write(f"\n[KILLED after {CELL_TIMEOUT}s timeout]".encode()) + _keep_result(log, workdir) + shutil.move(str(log), str(workdir / "_claude.stream.jsonl")) + shutil.move(str(err), str(workdir / "_claude.stderr.txt")) + +def _with_retries(run, wait=time.sleep): + """Run a cell, retrying with backoff while its session fails: a rate limit is not the arm's fault.""" + for pause in (*RETRY_WAITS, None): + res = run() + if not res.get("infra") or pause is None: return res + wait(pause) def aggregate(results): groups = defaultdict(list) for r in results: groups[(r["task"], r["arm"], r["model"])].append(r) rows = [] - for (t, a, m), cells in sorted(groups.items()): + for (t, a, m), all_cells in sorted(groups.items()): + cells = [c for c in all_cells if not c.get("infra")] # a failed session says nothing about the arm n = len(cells) + if not n: + rows.append({"task": t, "arm": a, "model": m, "n": 0, "infra": len(all_cells), "timeouts": 0, "safe_rate": 0, + "correct_rate": 0, "wrote_file_rate": 0, "total_loc_median": 0, "src_loc_median": 0, "total_loc_max": 0, + "src_files_median": 0, "wrote_tests_rate": 0, "cost_mean": None, "cost_per_pass": None, + "reply_words_median": None, "out_tokens_mean": None, "total_tokens_mean": None, "time_s_mean": None}) + continue costs = [c["cost"] for c in cells if c.get("cost") is not None] passes = sum(1 for c in cells if c.get("correct") and c.get("safe")) words = [c["reply_words"] for c in cells if c.get("reply_words") is not None] loc_cells = [c for c in cells if c.get("total_loc", 0) > 0] # LOC only where code was delivered nl = len(loc_cells) rows.append({"task": t, "arm": a, "model": m, "n": n, + "infra": len(all_cells) - n, "timeouts": sum(1 for c in cells if c.get("timed_out")), # a crashed cell has only "error": count it as failed instead of crashing the summary "safe_rate": round(sum(c.get("safe") or 0 for c in cells) / n, 3), "correct_rate": round(sum(c.get("correct") or 0 for c in cells) / n, 3), @@ -454,8 +709,9 @@ def aggregate(results): "src_files_median": statistics.median(c["src_files"] for c in loc_cells) if nl else 0, "wrote_tests_rate": round(sum(1 for c in cells if c.get("test_files", 0) > 0) / n, 3), "cost_mean": round(statistics.mean(costs), 4) if costs else None, - # what a finished task costs: all spend over the cells that passed both gates - "cost_per_pass": round(sum(costs) / passes, 4) if costs and passes else None, + # what a finished task costs: all spend over the cells that passed both gates. + # A killed session reports no cost, so with one in the group the figure would flatter the arm. + "cost_per_pass": round(sum(costs) / passes, 4) if costs and passes and len(costs) == n else None, "reply_words_median": statistics.median(words) if words else None, "out_tokens_mean": (round(statistics.mean([c["out_tokens"] for c in cells if c.get("out_tokens") is not None])) if any(c.get("out_tokens") is not None for c in cells) else None), @@ -471,7 +727,7 @@ def print_table(rows): for r in rows: by[(r["task"], r["model"])].append(r) for (task, model), rs in sorted(by.items()): print(f"\n=== {task} ({model}, n={rs[0]['n']}) ===") - print(f" {'arm':16} {'wrote%':>7} {'correct':>8} {'safe':>5} {'LOC':>7} {'tot_tok':>9} {'$/run':>8} {'$/pass':>8} {'words':>6} {'time_s':>7}") + print(f" {'arm':16} {'wrote%':>7} {'correct':>8} {'safe':>5} {'LOC':>7} {'tot_tok':>9} {'$/run':>8} {'$/pass':>8} {'words':>6} {'time_s':>7} {'killed':>6} {'failed':>6}") for r in sorted(rs, key=lambda x: x["arm"]): c = ("$" + format(r["cost_mean"], ".4f")) if r["cost_mean"] is not None else "-" tt = r.get("total_tokens_mean"); t = r.get("time_s_mean") @@ -479,7 +735,7 @@ def print_table(rows): w = r.get("reply_words_median") print(f" {r['arm']:16} {r.get('wrote_file_rate', 1.0):>7} {r['correct_rate']:>8} {r['safe_rate']:>5} " f"{r['total_loc_median']:>7} {(tt if tt is not None else '-'):>9} {c:>8} {cp:>8} " - f"{(w if w is not None else '-'):>6} {(t if t is not None else '-'):>7}") + f"{(w if w is not None else '-'):>6} {(t if t is not None else '-'):>7} {r['timeouts']:>6} {r['infra']:>6}") def rescore(run_dir): run_dir = Path(run_dir) @@ -490,7 +746,7 @@ def rescore(run_dir): parts = ws.name.split("__") if len(parts) != 4 or parts[0] not in TASKS: continue tid, arm, model, _r = parts - results.append(score_workspace(tid, arm, model, ws)) + results.append(score_isolated(tid, arm, model, ws)) rows = aggregate(results) (run_dir / "results.json").write_text(json.dumps({"rescored": True, "results": results}, indent=2), encoding="utf-8") (run_dir / "summary.json").write_text(json.dumps(rows, indent=2), encoding="utf-8") @@ -502,9 +758,13 @@ def _claude_version(): except Exception: return "unknown" def main(): + global CELL_TIMEOUT ap = argparse.ArgumentParser() ap.add_argument("--selftest", action="store_true") + ap.add_argument("--score-cell", nargs=3, metavar=("TASK", "ARM", "MODEL"), help=argparse.SUPPRESS) # scores the cwd ap.add_argument("--rescore", help="recompute metrics from a kept run dir (no API)") + ap.add_argument("--resume", help="kept run dir: run only its unfinished or failed cells, with the same sources") + ap.add_argument("--timeout", type=int, default=CELL_TIMEOUT, help=f"seconds per cell (default {CELL_TIMEOUT})") ap.add_argument("--task", help="single task id") ap.add_argument("--all", action="store_true", help="all tasks") ap.add_argument("--arms", default=",".join(ARMS)) @@ -513,7 +773,12 @@ def main(): ap.add_argument("--runs", type=int, default=1) ap.add_argument("--workers", type=int, default=4, help="cells to run concurrently (default 4; cells are fully isolated)") args = ap.parse_args() + CELL_TIMEOUT = args.timeout + if args.score_cell: + tid, arm, model = args.score_cell + ws = Path.cwd() + return (ws / "_score.json").write_text(json.dumps(score_workspace(tid, arm, model, ws)), encoding="utf-8") if args.selftest: sys.exit(1 if selftest() else 0) if args.rescore: @@ -526,16 +791,43 @@ def main(): if not task_ids: sys.exit("give --task (comma list ok), --all, or --rescore ") arms = [a.strip() for a in args.arms.split(",")] models = [m.strip() for m in (args.model or args.models).split(",")] - rivals = _rival_sources(arms) # before any spend: a bad checkout stops the run here - stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") - out_dir = RUNS_DIR / stamp - out_dir.mkdir(parents=True, exist_ok=True) - if rivals: (out_dir / "rivals.json").write_text(json.dumps(rivals, indent=2), encoding="utf-8") - - cells =[(tid, arm, model, r) + claude_version = _claude_version() + # before any spend: a bad checkout or an arm that does not activate stops the run here + cells = [(tid, arm, model, r) for tid in task_ids for model in models for arm in arms for r in range(args.runs)] + sources = {**_sources(arms, task_ids), "claude": claude_version, + "cells": [f"{t}__{a}__{m}__{r}" for t, a, m, r in cells]} + preflight(arms) + results = [] + if args.resume: + out_dir = Path(args.resume) if Path(args.resume).exists() else RUNS_DIR / Path(args.resume).name + stamp = out_dir.name + if json.loads((out_dir / "sources.json").read_text(encoding="utf-8")) != sources: + sys.exit(f"the sources or cells differ from when {stamp} started, so it cannot resume; compare with sources.json") + kept = {p.name for p in out_dir.iterdir() if p.is_dir()} + remaining = [] + for spec in cells: + tid, arm, model, r = spec + res = None + if f"{tid}__{arm}__{model}__{r}" in kept: + res = score_isolated(tid, arm, model, out_dir / f"{tid}__{arm}__{model}__{r}") + if res and not res.get("infra"): results.append(res) + else: remaining.append(spec) + print(f"resuming {stamp}: {len(results)} cells kept", flush=True) + cells = remaining + else: + stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + out_dir = RUNS_DIR / stamp + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "sources.json").write_text(json.dumps(sources, indent=2), encoding="utf-8") total = len(cells) - results, done = [], 0 + done = failed_in_a_row = 0 + stopped = False + + def write_results(): + (out_dir / "results.json").write_text(json.dumps( + {"date": stamp, "models": {m: MODELS.get(m, m) for m in models}, "cell_timeout": CELL_TIMEOUT, + "claude": claude_version, "results": results}, indent=2), encoding="utf-8") # Cells run outside any git repo, then move into out_dir. Inside a repo, Claude Code shows the # model that repo's branch and recent commit messages, so this repo's own history leaked into @@ -546,13 +838,20 @@ def _one(spec): tid, arm, model, r = spec name = f"{tid}__{arm}__{model}__{r}" ws = scratch / name - ws.mkdir(parents=True, exist_ok=True) - try: - return run_cell(tid, arm, model, ws) - finally: - shutil.move(str(ws), str(out_dir / name)) - - for m in models: warm_cache(m) + def attempt(): + shutil.rmtree(ws, ignore_errors=True) + ws.mkdir(parents=True) + try: + run_session(tid, arm, model, ws) + finally: # kept whatever happens: the session is paid for + shutil.rmtree(out_dir / name, ignore_errors=True) # a failed cell from an earlier attempt + shutil.move(str(ws), str(out_dir / name)) + return score_isolated(tid, arm, model, out_dir / name) + return _with_retries(attempt) + + if cells: + for m in models: + for arm in arms: warm_cache(arm, m) print(f"running {total} cells, {args.workers} at a time", flush=True) # Cells are fully isolated (own copy + own claude context), so they parallelize safely. # To STOP a parallel run, kill the whole tree: taskkill /PID /T /F. Killing just the @@ -563,24 +862,33 @@ def _one(spec): tid, arm, model, r = futs[fut] try: res = fut.result() - except Exception as e: - res = {"task": tid, "arm": arm, "model": model, "error": str(e)[:200]} + except Exception as e: # the harness broke, not the arm + res = {"task": tid, "arm": arm, "model": model, "infra": True, "error": str(e)[:200]} results.append(res) done += 1 print(f" [{done}/{total}] {tid} / {arm} / {model} #{r} " f"LOC={res.get('total_loc')} " f"tok={(res.get('in_tokens') or 0) + (res.get('out_tokens') or 0) + (res.get('cache_tokens') or 0)} " f"cost=${res.get('cost')} time={round((res.get('duration_ms') or 0) / 1000, 1)}s " - f"correct={res.get('correct')}", flush=True) - (out_dir / "results.json").write_text(json.dumps( - {"date": stamp, "models": {m: MODELS.get(m, m) for m in models}, - "claude": _claude_version(), "results": results}, indent=2), encoding="utf-8") - + f"correct={res.get('correct')}" + + (f" SESSION FAILED: {res.get('infra_reason') or res.get('error')}" if res.get("infra") else ""), flush=True) + write_results() + failed_in_a_row = failed_in_a_row + 1 if res.get("infra") else 0 + if failed_in_a_row >= MAX_FAILED_IN_A_ROW: + stopped = True + ex.shutdown(wait=False, cancel_futures=True) + break + + write_results() shutil.rmtree(scratch, ignore_errors=True) rows = aggregate(results) (out_dir / "summary.json").write_text(json.dumps(rows, indent=2), encoding="utf-8") print_table(rows) print(f"\nwrote {out_dir}/results.json + summary.json ({len(results)} cells)") + failed = sum(1 for c in results if c.get("infra")) + if stopped or failed: + sys.exit(f"{'stopped early: ' if stopped else ''}{failed} cells failed with session errors; " + f"rerun the same command with --resume {out_dir} to finish them") if __name__ == "__main__": main() diff --git a/benchmarks/agentic/tasks.py b/benchmarks/agentic/tasks.py index 6140bea81..c8f9d5086 100644 --- a/benchmarks/agentic/tasks.py +++ b/benchmarks/agentic/tasks.py @@ -814,7 +814,7 @@ def _page_count(workdir): # Files run.py itself leaves in a workspace: the CLI output, its stderr, and (for "diff" tasks) the # baseline git repository. Importing pager.py can also leave a __pycache__ directory. -_HARNESS_FILES = {"pager.py", "_claude.json", "_claude.stderr.txt"} +_HARNESS_FILES = {"pager.py", "_claude.json", "_claude.stream.jsonl", "_claude.stderr.txt", "_score.json"} def score_abstain_fixed(workdir): # Unscored: whether the reply says the bug is already fixed. Scorers get only the workspace; a