feat(benchmarks): agentic LOC + safety benchmark answering #126
Rebuild the benchmark to the standard #126 asked for: real headless Claude Code sessions (not a bare model) editing a real public repo (tiangolo/full-stack-fastapi-template @ cd83fc1, MIT), fair arms (baseline, caveman, ponytail, and the "YAGNI + one-liners" prompt), n=4, Haiku 4.5. LOC is the git diff; the safety tasks execute the produced code against adversarial input. Results: ponytail -54% LOC mean (up to -94% on over-build features like the date/color picker), -22% tokens, -20% cost, -27% time, and never more than baseline; 100% safe vs the one-liner prompt's 95% (it dropped a path-traversal guard once). caveman writes less code but spends more tokens. Also fixes a baseline-contamination bug (the ponytail plugin's SessionStart hook fired on every arm; now isolated with --setting-sources project,local + per-arm --plugin-dir) and a Windows subprocess-timeout hang. Lead both READMEs with the agentic numbers; demote the single-shot 80-94% to a labelled "isolated generation" note; supersede the contaminated 2026-06-17 writeup. Dead react-app fixture left untracked. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Emeriko
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
45f7d2f83f
commit
75fea8f743
@@ -0,0 +1,185 @@
|
||||
#!/usr/bin/env python3
|
||||
"""LLM-judge over-engineering pass for the agentic benchmark.
|
||||
|
||||
Over-engineering is the one axis that resists a deterministic check, so it gets an LLM judge --
|
||||
but an auditable one: a published rubric, a fixed judge model at temperature 0, and every score
|
||||
must name the specific construct it considers unnecessary (or "none"). The judge is validated
|
||||
first by --selftest: it must rank a deliberately over-engineered reference strictly above a
|
||||
minimal one for the same task, or we do not trust it on real submissions.
|
||||
|
||||
python judge.py --selftest # validate the judge on reference pairs (small spend)
|
||||
python judge.py --run runs/<stamp> # judge every workspace's source in a matrix run
|
||||
|
||||
Judge: claude-sonnet-4-6 via the Anthropic Messages API (key from ../../.env). Scores the SOURCE
|
||||
files only (tests excluded -- a test is not over-engineering). Cost is ~$0.003/cell.
|
||||
|
||||
ponytail: stdlib urllib for the API call, no requests dependency.
|
||||
"""
|
||||
import argparse, json, os, re, sys, time, urllib.request
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
from tasks import TASKS
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
RUNS_DIR = Path(__file__).resolve().parent / "runs"
|
||||
JUDGE_MODEL = "claude-sonnet-4-6"
|
||||
|
||||
RUBRIC = (
|
||||
"You are a senior engineer reviewing a code submission for OVER-ENGINEERING ONLY. "
|
||||
"Ignore correctness, style, performance, and security. Over-engineering means structure "
|
||||
"beyond what the task needs: speculative abstraction, classes/factories/config/flexibility "
|
||||
"nobody asked for, a framework for a one-off job. Score 0-3:\n"
|
||||
"0 = minimal, appropriate to the task\n"
|
||||
"1 = slightly more structure than needed\n"
|
||||
"2 = noticeably over-built (an unneeded class/abstraction/config/flexibility)\n"
|
||||
"3 = clearly over-engineered (speculative generality, a framework for a one-off)\n"
|
||||
"Name the single most unnecessary construct, or \"none\". "
|
||||
"Respond with ONLY this JSON: {\"over_engineering\": <0-3 int>, \"why\": \"<one line>\", \"cite\": \"<construct or none>\"}"
|
||||
)
|
||||
|
||||
def load_key():
|
||||
try:
|
||||
for line in (ROOT / ".env").read_text(encoding="utf-8").splitlines():
|
||||
if line.startswith("ANTHROPIC_API_KEY=") and len(line) > 18:
|
||||
return line.split("=", 1)[1].strip()
|
||||
except Exception:
|
||||
pass
|
||||
return os.environ.get("ANTHROPIC_API_KEY")
|
||||
|
||||
def _is_test(name):
|
||||
n = name.lower()
|
||||
return n.startswith("test_") or n.endswith("_test.py") or n == "conftest.py"
|
||||
|
||||
def source_text(workdir: Path):
|
||||
"""Concatenate the agent's source files (tests + artifacts excluded), with name headers."""
|
||||
out = []
|
||||
for p in sorted(workdir.rglob("*")):
|
||||
if not p.is_file() or "__pycache__" in p.parts or p.suffix == ".pyc": continue
|
||||
if p.name.startswith((".", "_")) or _is_test(p.name): continue
|
||||
try: out.append(f"# === {p.relative_to(workdir)} ===\n{p.read_text(encoding='utf-8', errors='ignore')}")
|
||||
except Exception: continue
|
||||
return "\n\n".join(out)
|
||||
|
||||
def judge_call(task_prompt, files, key, retries=3):
|
||||
user = f"TASK GIVEN TO THE AUTHOR:\n{task_prompt}\n\nFILES THEY WROTE:\n{files}"
|
||||
body = json.dumps({"model": JUDGE_MODEL, "max_tokens": 300, "temperature": 0,
|
||||
"system": RUBRIC, "messages": [{"role": "user", "content": user}]}).encode()
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
req = urllib.request.Request("https://api.anthropic.com/v1/messages", data=body,
|
||||
headers={"x-api-key": key, "anthropic-version": "2023-06-01", "content-type": "application/json"})
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
j = json.loads(r.read())
|
||||
return j["content"][0]["text"]
|
||||
except Exception as e:
|
||||
if attempt == retries - 1: return f'{{"error": "{str(e)[:120]}"}}'
|
||||
time.sleep(2 * (attempt + 1))
|
||||
|
||||
def parse_score(text):
|
||||
m = re.search(r"\{.*\}", text or "", re.S)
|
||||
if not m: return None
|
||||
try:
|
||||
d = json.loads(m.group(0))
|
||||
if "over_engineering" in d: d["over_engineering"] = int(d["over_engineering"])
|
||||
return d
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
# --- selftest: the judge must rank over-engineered above minimal for the same task ---
|
||||
CACHE_OVER = (
|
||||
"import time\nfrom collections import OrderedDict\n"
|
||||
"class CacheEntry:\n def __init__(self, value, created_at):\n self.value = value\n self.created_at = created_at\n"
|
||||
"class ComputeCache:\n \"\"\"Configurable TTL cache with LRU eviction and hit/miss stats.\"\"\"\n"
|
||||
" def __init__(self, max_size=128, ttl_seconds=3600, enable_stats=True):\n"
|
||||
" self.max_size = max_size; self.ttl_seconds = ttl_seconds; self.enable_stats = enable_stats\n"
|
||||
" self._store = OrderedDict(); self._hits = 0; self._misses = 0\n"
|
||||
" def _evict(self):\n while len(self._store) > self.max_size: self._store.popitem(last=False)\n"
|
||||
" def get_or_compute(self, n, fn):\n now = time.time()\n"
|
||||
" if n in self._store and now - self._store[n].created_at < self.ttl_seconds:\n"
|
||||
" self._hits += 1; self._store.move_to_end(n); return self._store[n].value\n"
|
||||
" self._misses += 1; v = fn(n); self._store[n] = CacheEntry(v, now); self._evict(); return v\n"
|
||||
"_cache = ComputeCache()\n"
|
||||
"def compute(n):\n return _cache.get_or_compute(n, lambda m: sum(i*i for i in range(m)))\n"
|
||||
)
|
||||
SAFEPATH_OVER = (
|
||||
"import os\nclass PathPolicy:\n def __init__(self, allow_symlinks=False, max_depth=10, allowed_extensions=None):\n"
|
||||
" self.allow_symlinks = allow_symlinks; self.max_depth = max_depth\n self.allowed_extensions = allowed_extensions or []\n"
|
||||
"class PathSanitizer:\n \"\"\"Pluggable path sanitizer with configurable policy.\"\"\"\n def __init__(self, policy=None):\n self.policy = policy or PathPolicy()\n"
|
||||
" def sanitize(self, base_dir, filename):\n base = os.path.abspath(base_dir)\n target = os.path.abspath(os.path.join(base, filename))\n"
|
||||
" if os.path.commonpath([base, target]) != base: raise ValueError('traversal')\n return target\n"
|
||||
"_default = PathSanitizer()\ndef safe_upload_path(base_dir, filename):\n return _default.sanitize(base_dir, filename)\n"
|
||||
)
|
||||
SELFTEST_PAIRS = [
|
||||
("cache", "minimal", TASKS["cache"]["good"]),
|
||||
("cache", "over", CACHE_OVER),
|
||||
("safe-path", "minimal", TASKS["safe-path"]["good"]),
|
||||
("safe-path", "over", SAFEPATH_OVER),
|
||||
]
|
||||
|
||||
def selftest(key):
|
||||
scores = {}
|
||||
for task_id, label, code in SELFTEST_PAIRS:
|
||||
s = parse_score(judge_call(TASKS[task_id]["prompt"], code, key))
|
||||
scores[(task_id, label)] = s
|
||||
print(f" {task_id:10} {label:8} -> {s}")
|
||||
ok = True
|
||||
for task_id in ("cache", "safe-path"):
|
||||
lo = scores.get((task_id, "minimal"), {}) or {}
|
||||
hi = scores.get((task_id, "over"), {}) or {}
|
||||
if not (isinstance(hi.get("over_engineering"), int) and isinstance(lo.get("over_engineering"), int)
|
||||
and hi["over_engineering"] > lo["over_engineering"]):
|
||||
print(f"XX {task_id}: judge did not rank over-engineered above minimal")
|
||||
ok = False
|
||||
else:
|
||||
print(f"ok {task_id}: over({hi['over_engineering']}) > minimal({lo['over_engineering']})")
|
||||
print(f"\njudge selftest: {'valid' if ok else 'NOT TRUSTWORTHY'}")
|
||||
return 0 if ok else 1
|
||||
|
||||
def run(run_dir, key):
|
||||
run_dir = Path(run_dir)
|
||||
if not run_dir.exists(): run_dir = RUNS_DIR / run_dir.name
|
||||
cells, scored = [], []
|
||||
for ws in sorted(p for p in run_dir.iterdir() if p.is_dir()):
|
||||
parts = ws.name.split("__")
|
||||
if len(parts) != 4 or parts[0] not in TASKS: continue
|
||||
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"),
|
||||
"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)
|
||||
(run_dir / "judge.json").write_text(json.dumps({"judge": JUDGE_MODEL, "rubric": RUBRIC, "scores": scored}, indent=2), encoding="utf-8")
|
||||
# aggregate
|
||||
by_arm = defaultdict(list)
|
||||
for r in scored:
|
||||
if isinstance(r["over_engineering"], int): by_arm[r["arm"]].append(r["over_engineering"])
|
||||
print(f"\n=== over-engineering by arm (judge: {JUDGE_MODEL}, 0=minimal .. 3=over-built) ===")
|
||||
print(f" {'arm':16} {'n':>4} {'mean':>6} {'max':>4}")
|
||||
for arm in ["baseline", "caveman", "ponytail", "yagni", "yagni-oneliner"]:
|
||||
v = by_arm.get(arm, [])
|
||||
if v: print(f" {arm:16} {len(v):>4} {sum(v)/len(v):>6.2f} {max(v):>4}")
|
||||
worst = sorted([r for r in scored if isinstance(r["over_engineering"], int) and r["over_engineering"] >= 2],
|
||||
key=lambda r: -r["over_engineering"])
|
||||
print(f"\n=== flagged over-engineered (score >= 2): {len(worst)} cells ===")
|
||||
for r in worst[:20]:
|
||||
print(f" {r['task']:11} {r['arm']:15} {r['model']:7} score={r['over_engineering']} cite={r['cite']}")
|
||||
print(f"\nwrote {run_dir / 'judge.json'}")
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
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)")
|
||||
if args.selftest: sys.exit(selftest(key))
|
||||
if args.run:
|
||||
if selftest(key): sys.exit("judge not trustworthy; refusing to judge the matrix")
|
||||
return run(args.run, key)
|
||||
sys.exit("give --selftest or --run <dir>")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user