Blame for src/miniweb/scripts/grader_audit_report.py

"""Score the verifier and the VLM judge against the grader-audit labels (src/miniweb/scripts/build_grader_audit.py).

A human task label is pass when every primitive is labeled pass, fail when any is labeled fail, unsure otherwise;
unsure labels are left out. Task-level accuracy is estimated for the full population of judged RQ1 episodes by
weighting each stratum (both pass, both fail, judge only, verifier only) by its size there, with a stratified
bootstrap 95% interval. Primitive-level accuracy is reported per family, unweighted. Cases with two labelers give
the inter-labeler agreement and Cohen's kappa.

    python src/miniweb/scripts/grader_audit_report.py --url https://<deployment>   # token: MINIWEB_RECOVERY_TOKEN
    python src/miniweb/scripts/grader_audit_report.py --labels labels.json [--primary <annotator>]
    python src/miniweb/scripts/grader_audit_report.py --url ... --pull-only     # back up the labels without scoring them
"""
import argparse, json, os, random, sys, urllib.request
from collections import defaultdict
from pathlib import Path

ROOT = Path(__file__).resolve().parents[3]   # the repository
sys.path.insert(0, str(ROOT / "src"))
from miniwebgen.viewer import FAMILIES  # noqa: E402

OUT = ROOT / "data/grader_audit"
FAM = {m: f for f, ms in FAMILIES.items() for m in ms}


def fetch(url, token):
    req = urllib.request.Request(url.rstrip("/") + "/recovery/grader_audit/labels", headers={"X-Recovery-Token": token})
    with urllib.request.urlopen(req, timeout=60) as r:
        return json.loads(r.read().decode())


def task_label(prims):
    vals = set(prims.values())
    return "fail" if "fail" in vals else "pass" if vals == {"pass"} else "unsure"


def kappa(a, b):
    n = len(a)
    if not n:
        return float("nan"), float("nan")
    po = sum(x == y for x, y in zip(a, b)) / n
    pa, pb = sum(a) / n, sum(b) / n
    pe = pa * pb + (1 - pa) * (1 - pb)
    return po, (po - pe) / (1 - pe) if pe < 1 else float("nan")


def weighted_acc(rows, pop, grader):
    """Population accuracy of a grader: per-stratum accuracy weighted by the stratum's population share."""
    by = defaultdict(list)
    for r in rows:
        by[r["stratum"]].append(r[grader] == r["human"])
    tot = sum(pop[s] for s in by)
    return sum(pop[s] / tot * sum(v) / len(v) for s, v in by.items()), {s: (sum(v), len(v)) for s, v in by.items()}


def bootstrap(rows, pop, grader, reps=2000, seed=0):
    rng, by = random.Random(seed), defaultdict(list)
    for r in rows:
        by[r["stratum"]].append(r)
    accs = []
    for _ in range(reps):
        sample = [x for s in by.values() for x in (rng.choice(s) for _ in s)]
        accs.append(weighted_acc(sample, pop, grader)[0])
    accs.sort()
    return accs[int(0.025 * reps)], accs[int(0.975 * reps) - 1]


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--url", help="deployment to fetch labels from (token in MINIWEB_RECOVERY_TOKEN)")
    ap.add_argument("--labels", help="a labels JSON list instead (as /annotate/api/grader_audit/labels.json gives)")
    ap.add_argument("--primary", help="labeler whose labels are scored (default: the one with most labels)")
    ap.add_argument("--pull-only", action="store_true",
                    help="only save the labels to data/grader_audit/labels.json (no scores, so labeling stays blind)")
    args = ap.parse_args()
    key = json.loads((OUT / "key.json").read_text())
    pop, cases = key["population"], key["cases"]
    labels = fetch(args.url, os.environ.get("MINIWEB_RECOVERY_TOKEN", "")) if args.url else json.load(open(args.labels))
    (OUT / "labels.json").write_text(json.dumps(labels, indent=1, ensure_ascii=False))
    if args.pull_only:
        print(f"{len(labels)} labels by {sorted({r['labeler'] for r in labels})} -> {OUT / 'labels.json'}")
        return
    by_labeler = defaultdict(dict)
    for r in labels:
        if r["case_id"] in cases:
            by_labeler[r["labeler"]][r["case_id"]] = r
    if not by_labeler:
        sys.exit("no labels yet")
    primary = args.primary or max(by_labeler, key=lambda k: len(by_labeler[k]))
    lab = by_labeler[primary]
    rep = {"primary": primary, "labelers": {k: len(v) for k, v in by_labeler.items()}, "n_cases": len(cases)}

    tasks, inst = [], []
    for cid, r in lab.items():
        k = cases[cid]
        h = task_label(r["primitives"])
        if h != "unsure":
            tasks.append({"stratum": k["stratum"], "human": h == "pass", "judge": k["judge"], "verifier": k["verifier"]})
        for i, v in r["primitives"].items():
            if v != "unsure":
                inst.append({"family": FAM.get(i.split("#")[0], "?"), "human": v == "pass",
                             "judge": k["judge_instances"].get(i), "verifier": k["verifier_instances"].get(i)})
    rep["tasks_scored"], rep["tasks_unsure"] = len(tasks), len(lab) - len(tasks)
    for g in ("judge", "verifier"):
        acc, per = weighted_acc(tasks, pop, g)
        lo, hi = bootstrap(tasks, pop, g)
        rep[f"{g}_task_accuracy"] = {"population_weighted": acc, "ci95": [lo, hi],
                                     "per_stratum": {s: f"{a}/{n}" for s, (a, n) in per.items()}}
        fam = defaultdict(lambda: [0, 0])
        for x in inst:
            if x[g] is not None:
                fam[x["family"]][0] += x[g] == x["human"]
                fam[x["family"]][1] += 1
        rep[f"{g}_primitive_accuracy"] = {f: f"{a}/{n} ({100 * a / n:.1f}%)" for f, (a, n) in sorted(fam.items())}
    # who is right on the disagreements: the share of each disagreement stratum the human sides with the judge
    for s in ("judge_only", "verifier_only"):
        rows = [t for t in tasks if t["stratum"] == s]
        rep[f"{s}_human_pass"] = f"{sum(t['human'] for t in rows)}/{len(rows)}"

    others = [k for k in by_labeler if k != primary]
    agree = {}
    for o in others:
        a, b = [], []
        for cid in set(lab) & set(by_labeler[o]):
            for i, v in lab[cid]["primitives"].items():
                w = by_labeler[o][cid]["primitives"].get(i)
                if "unsure" not in (v, w) and w is not None:
                    a.append(v == "pass")
                    b.append(w == "pass")
        po, kp = kappa(a, b)
        agree[o] = {"primitives": len(a), "agree": po, "kappa": kp}
    rep["inter_labeler"] = agree
    (OUT / "report.json").write_text(json.dumps(rep, indent=1))
    print(json.dumps(rep, indent=1))


if __name__ == "__main__":
    main()