feat(benchmarks): add completeness judge so LOC wins can't hide under-delivery (#171)
The LOC tier scores the open feature tasks (vibe-*, tmpl-fe-*, open-*) on git diff alone -- score_vibe only checks "it compiles", score_fixture only checks "a new file exists". So an arm can win the LOC metric by shipping a stub: fewer lines because it does less, not because it is less bloated. That is the most credible attack left on the headline number raised in #126. complete.py is a second LLM judge (same auditable footing as judge.py: fixed model, temperature 0, published rubric) that rates how FULLY each submission implements its task, 0..3. Read alongside the LOC table, a low-LOC arm whose completeness also drops is caught, not rewarded. - judge_call gains a `system=` param so the HTTP/key/source plumbing is reused instead of duplicated (one rubric is the only delta between the two passes). - --selftest: the judge must rank a complete reference strictly above a stub. - --selftest-offline: validates the gate logic with no API call / no key. - README documents the pass and updates the can/cannot-show limitations. Fixes #126 Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Younes
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
44babb22ef
commit
955fff537c
@@ -0,0 +1,154 @@
|
||||
#!/usr/bin/env python3
|
||||
"""LLM-judge COMPLETENESS pass for the agentic benchmark.
|
||||
|
||||
Fewer lines is only a win if the code still does the job. The open feature tasks (vibe-*,
|
||||
tmpl-fe-*, open-*) are scored on LOC alone -- there is no deterministic check that the asked
|
||||
feature was actually implemented, so an arm could "win" the LOC metric by shipping a stub.
|
||||
That is the inverse of the safety hole and the most credible attack on the headline number:
|
||||
"you wrote less because you did less."
|
||||
|
||||
This pass closes it. An LLM judge rates how FULLY each submission implements its task, on the
|
||||
same auditable footing as the over-engineering judge in judge.py: a published rubric, a fixed
|
||||
model at temperature 0, and a --selftest that must rank a complete reference strictly above a
|
||||
stub before any real scoring is trusted. Pair the output with run.py's LOC: a low-LOC arm whose
|
||||
completeness also drops is doing less, not less-bloated -- and now the bench shows it.
|
||||
|
||||
python complete.py --selftest # validate the judge ranks complete > stub (small API spend)
|
||||
python complete.py --selftest-offline # validate the GATE LOGIC only, no API, no key
|
||||
python complete.py --run runs/<stamp> # completeness-judge every workspace in a matrix run
|
||||
|
||||
Judge: claude-sonnet-4-6, key from ../../.env (shared with judge.py). ~$0.003/cell.
|
||||
|
||||
ponytail: 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, sys
|
||||
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
|
||||
|
||||
SCORE_KEY = "completeness"
|
||||
FLAG_AT = 1 # cells scoring <= this are under-delivery (stub/partial) and get listed
|
||||
ARMS_ORDER = ["baseline", "caveman", "ponytail", "yagni", "yagni-oneliner"]
|
||||
|
||||
RUBRIC = (
|
||||
"You are a senior engineer checking whether a code submission ACTUALLY IMPLEMENTS the task it "
|
||||
"was given. Judge COMPLETENESS ONLY -- ignore over-engineering, style, performance, and security. "
|
||||
"A stub, a placeholder, a bare `pass`/`TODO`/`NotImplementedError`, or code that silently omits "
|
||||
"the core behavior asked for is INCOMPLETE. Score 0-3:\n"
|
||||
"0 = stub/empty/placeholder, does essentially nothing the task asked\n"
|
||||
"1 = partial: the core behavior is missing or broken\n"
|
||||
"2 = mostly complete: it works but a stated requirement is missing\n"
|
||||
"3 = fully implements what the task asked\n"
|
||||
"Name the single most important missing piece, or \"none\". "
|
||||
"Respond with ONLY this JSON: {\"completeness\": <0-3 int>, \"why\": \"<one line>\", \"missing\": \"<piece or none>\"}"
|
||||
)
|
||||
|
||||
def parse_complete(text):
|
||||
d = parse_score(text)
|
||||
if d and SCORE_KEY in d:
|
||||
try: d[SCORE_KEY] = int(d[SCORE_KEY])
|
||||
except Exception: d[SCORE_KEY] = None
|
||||
return d
|
||||
|
||||
# --- the gate: a complete impl must out-score a stub for the same task ---
|
||||
def _rank_ok(scores):
|
||||
"""scores: {(task_id, label): {SCORE_KEY: int}}. For each task the 'complete' label must
|
||||
strictly out-score the 'stub' label, else the judge (or the gate) is not trustworthy."""
|
||||
ok = True
|
||||
for task_id in sorted({t for (t, _) in scores}):
|
||||
hi = scores.get((task_id, "complete")) or {}
|
||||
lo = scores.get((task_id, "stub")) or {}
|
||||
if not (isinstance(hi.get(SCORE_KEY), int) and isinstance(lo.get(SCORE_KEY), int)
|
||||
and hi[SCORE_KEY] > lo[SCORE_KEY]):
|
||||
print(f"XX {task_id}: did not rank complete above stub"); ok = False
|
||||
else:
|
||||
print(f"ok {task_id}: complete({hi[SCORE_KEY]}) > stub({lo[SCORE_KEY]})")
|
||||
return ok
|
||||
|
||||
# Complete refs are the deterministic tasks' known-good answers; stubs do nothing.
|
||||
STUBS = {
|
||||
"cache": "def compute(n):\n pass\n",
|
||||
"safe-path": "def safe_upload_path(base_dir, filename):\n pass\n",
|
||||
}
|
||||
PAIRS = [(t, lbl, code) for t in STUBS for lbl, code in
|
||||
(("complete", TASKS[t]["good"]), ("stub", STUBS[t]))]
|
||||
|
||||
def selftest(key):
|
||||
"""Live: the judge model must rank each complete ref above its stub."""
|
||||
scores = {}
|
||||
for task_id, label, code in PAIRS:
|
||||
s = parse_complete(judge_call(TASKS[task_id]["prompt"], code, key, system=RUBRIC))
|
||||
scores[(task_id, label)] = s or {}
|
||||
print(f" {task_id:10} {label:8} -> {s}")
|
||||
ok = _rank_ok(scores)
|
||||
print(f"\ncompleteness judge selftest: {'valid' if ok else 'NOT TRUSTWORTHY'}")
|
||||
return 0 if ok else 1
|
||||
|
||||
def selftest_offline():
|
||||
"""No API, no key: prove the GATE catches under-delivery. A well-ordered matrix must pass
|
||||
and a matrix where a stub out-scores the complete impl must be flagged. Fails loudly if the
|
||||
gate is ever weakened into a no-op."""
|
||||
good = {("cache", "complete"): {SCORE_KEY: 3}, ("cache", "stub"): {SCORE_KEY: 0}}
|
||||
bad = {("cache", "complete"): {SCORE_KEY: 1}, ("cache", "stub"): {SCORE_KEY: 3}}
|
||||
print("offline gate -- well-ordered (expect ok):")
|
||||
p_good = _rank_ok(good)
|
||||
print("offline gate -- stub out-scores complete (expect XX):")
|
||||
p_bad = _rank_ok(bad)
|
||||
passed = p_good and not p_bad
|
||||
print(f"\ncompleteness gate selftest (offline): {'valid' if passed else 'BROKEN'}")
|
||||
return 0 if passed 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 = []
|
||||
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"completeness-judging {len(cells)} workspaces with {JUDGE_MODEL} ...")
|
||||
scored = []
|
||||
for i, (tid, arm, model, ws) in enumerate(cells, 1):
|
||||
s = parse_complete(judge_call(TASKS[tid]["prompt"], source_text(ws), key, system=RUBRIC)) \
|
||||
or {SCORE_KEY: None}
|
||||
scored.append({"task": tid, "arm": arm, "model": model, SCORE_KEY: s.get(SCORE_KEY),
|
||||
"why": s.get("why", ""), "missing": s.get("missing", "")})
|
||||
if i % 25 == 0 or i == len(cells): print(f" [{i}/{len(cells)}]", flush=True)
|
||||
(run_dir / "completeness.json").write_text(
|
||||
json.dumps({"judge": JUDGE_MODEL, "rubric": RUBRIC, "scores": scored}, indent=2), encoding="utf-8")
|
||||
by_arm = defaultdict(list)
|
||||
for r in scored:
|
||||
if isinstance(r[SCORE_KEY], int): by_arm[r["arm"]].append(r[SCORE_KEY])
|
||||
print(f"\n=== completeness by arm (judge: {JUDGE_MODEL}, 0=stub .. 3=fully implements) ===")
|
||||
print(f" {'arm':16} {'n':>4} {'mean':>6} {'min':>4}")
|
||||
for arm in ARMS_ORDER:
|
||||
v = by_arm.get(arm, [])
|
||||
if v: print(f" {arm:16} {len(v):>4} {sum(v)/len(v):>6.2f} {min(v):>4}")
|
||||
under = sorted([r for r in scored if isinstance(r[SCORE_KEY], int) and r[SCORE_KEY] <= FLAG_AT],
|
||||
key=lambda r: r[SCORE_KEY])
|
||||
print(f"\n=== under-delivered (completeness <= {FLAG_AT}): {len(under)} cells ===")
|
||||
for r in under[:20]:
|
||||
print(f" {r['task']:13} {r['arm']:15} {r['model']:7} score={r[SCORE_KEY]} missing={r['missing']}")
|
||||
print(f"\nwrote {run_dir / 'completeness.json'}")
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--selftest", action="store_true", help="live: judge ranks complete > stub")
|
||||
ap.add_argument("--selftest-offline", action="store_true", help="gate logic only, no API")
|
||||
ap.add_argument("--run", help="run dir to completeness-judge")
|
||||
args = ap.parse_args()
|
||||
if args.selftest_offline:
|
||||
sys.exit(selftest_offline())
|
||||
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, --selftest-offline, or --run <dir>")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user