#!/usr/bin/env python3 """Leg H's scorer: collapse the orders, pool over cases, bootstrap over the 36 clusters. No rating. python3 harness/pairwise_score.py --plan python3 harness/pairwise_score.py --score --out results/legH/pairwise.json THE FENCE THIS FILE ENFORCES, BY NAME ------------------------------------ PREREG §9 forbids *"Bradley-Terry or any pairwise-derived rating"* on this instrument, and PLAN §2's second fence says the ONLY comparative object published is a per-case preference count with its denominator beside it and a cluster interval over cases. **There is no Bradley-Terry here and there is no rating function to add one to.** What there is: 1. **COLLAPSE FIRST.** The two orders of one (judge, case) comparison become ONE observation — the mean of the two, with A-preferred = 1, tie = 0.5, B-preferred = 0 — before anything is computed. Both orders in one bucket would double-count the case and shrink every interval. 2. **POOL OVER CASES, not over calls.** The case is the independent unit (PLAN §2). The pooled rate is the mean over cases of the mean over judges. 3. **A CLUSTER BOOTSTRAP OVER THE 36 CASES** — 2,000 percentile resamples, resampling CASES (the cluster), never judged cells. The t(35) note prints beside it because 36 ≥ 30 is what makes the interval reportable at all, and it is barely so. 4. **THE ORDER-FLIP RATE IS ITS OWN MECHANICAL COLUMN** — how often a judge's verdict changed when the same two answers changed places. It is a property of the panel, not of the arms. 5. **THE NULL PARAGRAPH IS WRITTEN FROM THE NUMBERS, and its text was fixed before the run** (PREREG §5). If the interval covers 0.5 the page says the panel did not separate the arms at this sample size. No number in the paragraph is typed by hand. The recusal join runs and is asserted EMPTY: the local seat is not in Leg H, so no seat shares a family with either arm here (PREREG §7; A3 ANSWERED (8)). """ from __future__ import annotations import sys sys.dont_write_bytecode = True import argparse import json import math import random import statistics from collections import defaultdict from pathlib import Path import judge_seats as JS import prereg_integers as P import sample_record as sr BASE = Path(__file__).resolve().parent.parent OUT_DIR = BASE / "results" / "legH" #: A-preferred / tie / B-preferred as numbers. The direction is stated once, here, and the report #: prints which arm "A" was for each sheet from `key.json` — the letter itself means nothing. VERDICT_VALUE = {"A": 1.0, "tie": 0.5, "B": 0.0} #: PREREG §5, written before the run. `{}`-formatted from the scorer's own numbers, never typed. #: #: THE THREE NUMBERS THAT WERE TYPED (audit part 2, B1 and B3). "36-case" was in the null paragraph #: at any n — 35 collected cases still read "the 36-case interval" — and "seven-family" was in the #: separation sentence however many families actually carried. Both are §9-forbidden (*"every number #: in the prose is written by the scorer, never typed"*), and both now format from the measurement. NULL_PARAGRAPH = ( "The panel's reading, over {n} cases, gives an interval of [{lo}, {hi}] that covers 0.5: the " "panel did not separate the two arms on the rules desk at this sample size; the per-case table " "shows where each was preferred and the page draws no ordering." ) SEPARATION_SENTENCE = ( "preferred by a blind panel of {families} families in {favoured} of {n} cases " "(interval {lo}–{hi}); this is a one-day reading and carries no ordering past its window." ) #: §9 again: *"intervals only where the unit N ≥ 30; below it, counts with denominators and no #: interval"*. There was no sentence for that case, and `SEPARATION_SENTENCE.format` raised KeyError #: on `lo` — the scorer CRASHED at 29 cases (audit part 2, B5). SEPARATION_SENTENCE_NO_INTERVAL = ( "preferred by a blind panel of {families} families in {favoured} of {n} cases (no interval: the " "independent unit N is {n}, below the registered N ≥ {min_n}, so this is a count with its " "denominator); this is a one-day reading and carries no ordering past its window." ) def load_key(out_dir: Path | None = None) -> dict: p = (out_dir or OUT_DIR) / "key.json" if not p.is_file(): raise SystemExit(f"no key at {p}; pairwise_build.py --build writes it") return json.loads(p.read_text(encoding="utf-8"))["key"] def load_judge_rows(out_dir: Path | None = None) -> list[dict]: d = (out_dir or OUT_DIR) / "judge" if not d.is_dir(): raise SystemExit(f"no judged rows under {d}") rows = [] for path in sorted(d.glob("*.jsonl")): for line in path.read_text(encoding="utf-8").splitlines(): if line.strip(): rows.append(json.loads(line)) return rows def load_manifest(out_dir: Path | None = None) -> dict: """The sheet manifest — the only file that knows how many cases A11 appended.""" p = (out_dir or OUT_DIR) / "manifest.json" if not p.is_file(): raise SystemExit(f"no sheet manifest at {p}; pairwise_build.py --build writes it") return json.loads(p.read_text(encoding="utf-8")) def seat_census(rows: list[dict], key: dict) -> dict: """PREREG A16 (4) — the head-to-head's collection PER SEAT. The panel-wide census says 55 rows did not carry; it does not say that all 55 are one seat, and the pooled rate is over the cases that HAVE observations, so a reader who cannot see the per-seat split cannot see which seat is missing from which case. Every seat's row therefore carries: the rows it was sent, how many carried, how many did not, the cases it actually contributes to, the sheet it was lost at, and the runner's own recorded reason. The four state buckets partition the seat's rows exactly (COLLECTED · NOT-CARRIED · the ctx refusal · everything else, which is transport), so ``rows`` is always their sum — a state this round has not met cannot hide inside a subtraction. """ census: dict[str, dict] = {} for row in rows: seat = census.setdefault(row["seat"], { "family": row["family"], "rows": 0, "collected": 0, "not_carried": 0, "refused_ctx": 0, "transport": 0, "cases_carried": 0, "lost_at_sheet": None, "detail": None, "_cases": set()}) seat["rows"] += 1 state = row.get("collection_state") if state == JS.NOT_CARRIED: seat["not_carried"] += 1 if seat["lost_at_sheet"] is None: # the FIRST sheet this seat was lost at, and the reason as the runner wrote it seat["lost_at_sheet"] = row.get("sheet_id") seat["detail"] = row.get("collection_detail") elif state == JS.REFUSED_CTX: seat["refused_ctx"] += 1 elif state != sr.COLLECTED or not row.get("verdict"): seat["transport"] += 1 else: seat["collected"] += 1 joined = key.get(row["sheet_id"]) if not joined: raise SystemExit(f"sheet {row['sheet_id']!r} is not in key.json; refusing to guess") seat["_cases"].add(joined["case_id"]) for seat in census.values(): seat["cases_carried"] = len(seat.pop("_cases")) return {name: census[name] for name in sorted(census)} def collapse(rows: list[dict], key: dict, *, arm_a: str) -> tuple[dict, dict]: """(observations[(seat, case)] -> value in [0,1], census). The collapse of both orders. ``arm_a`` names the arm the RATE is about, so a reader never has to guess whether 0.62 favours the arm they were thinking of. The per-order verdict is re-expressed as "did this judge prefer ``arm_a``", which is what makes the two orders the same quantity. """ per_pair: dict[tuple[str, str], dict[int, float]] = defaultdict(dict) census = {"rows": len(rows), "collected": 0, "not_carried": 0, "refused_ctx": 0, "transport": 0, "recognised_cells": 0, "orders_missing_a_half": 0, "recognised_flag_rule": sr.RECOGNISED_FLAG_RULE} for row in rows: state = row.get("collection_state") if state == JS.NOT_CARRIED: census["not_carried"] += 1 continue if state == JS.REFUSED_CTX: census["refused_ctx"] += 1 continue if state != sr.COLLECTED or not row.get("verdict"): census["transport"] += 1 continue census["collected"] += 1 if sr.recognised_flag(row): census["recognised_cells"] += 1 k = key.get(row["sheet_id"]) if not k: raise SystemExit(f"sheet {row['sheet_id']!r} is not in key.json; refusing to guess") value = VERDICT_VALUE[row["verdict"]] # Re-express as "preferred arm_a": if arm_a sat in position B, flip. if k["position_A_arm"] != arm_a: value = 1.0 - value per_pair[(row["seat"], k["case_id"])][int(k["order"])] = value observations: dict[tuple[str, str], float] = {} for pair, orders in per_pair.items(): if len(orders) < P.LEG_H_ORDERS: census["orders_missing_a_half"] += 1 observations[pair] = statistics.fmean(orders.values()) return observations, census def order_flip_rate(rows: list[dict], key: dict, *, arm_a: str) -> dict: """How often a judge's verdict changed when the same two answers changed places. Both orders are re-expressed as "preference for ``arm_a``" first — the raw letter cannot be compared across orders, because the letters mean different arms in each. A pair counts as flipped when the two orders disagree after that translation (0 vs 1, or either vs a 0.5 tie). """ per: dict[tuple[str, str], dict[int, float]] = defaultdict(dict) for row in rows: if row.get("collection_state") != sr.COLLECTED or not row.get("verdict"): continue k = key[row["sheet_id"]] value = VERDICT_VALUE[row["verdict"]] if k["position_A_arm"] != arm_a: value = 1.0 - value per[(row["seat"], k["case_id"])][int(k["order"])] = value flips, both, tie_involved = 0, 0, 0 for orders in per.values(): if len(orders) < P.LEG_H_ORDERS: continue both += 1 values = sorted(orders.values()) if len({round(v, 3) for v in values}) > 1: flips += 1 if 0.5 in {round(v, 3) for v in values}: tie_involved += 1 return {"pairs_with_both_orders": both, "flipped": flips, "flips_where_one_order_was_a_tie": tie_involved, "rate": round(flips / both, 3) if both else None, "denominator": f"{both} (judge, case) comparisons with both orders collected", "what_it_is": "a property of the PANEL, not of the arms: the fraction of (judge, case) " "comparisons whose preference changed when the two answers swapped places"} def cluster_bootstrap(per_case: dict[str, float], *, resamples: int = P.BOOTSTRAP_RESAMPLES, seed: int = 0) -> dict: """Percentile interval, resampling CASES (the cluster). Never judged cells.""" cases = sorted(per_case) n = len(cases) if n < P.INTERVAL_MIN_N: return {"state": "NO-INTERVAL", "why": f"the independent unit N is {n}; PLAN §2 permits an interval only at " f"N ≥ {P.INTERVAL_MIN_N}. The counts print with their denominator instead."} rng = random.Random(seed) means = [] for _ in range(resamples): draw = [per_case[cases[rng.randrange(n)]] for _ in range(n)] means.append(statistics.fmean(draw)) means.sort() lo = means[int(0.025 * resamples)] hi = means[min(resamples - 1, int(0.975 * resamples))] sd = statistics.pstdev(list(per_case.values())) if n > 1 else 0.0 return {"state": "PERCENTILE", "resamples": resamples, "clusters": n, "lo": round(lo, 3), "hi": round(hi, 3), "t_note": (f"t({n - 1}) — 36 clusters is barely above the N ≥ " f"{P.INTERVAL_MIN_N} line this round registered, so the interval is " "reportable and wide, and the per-case table is printed beside it"), "cluster_sd": round(sd, 3), "seed": seed} def guarded_rate(values: list[float], family: str | None) -> dict: """One judge's cell: the count and its denominator always, the proportion only at N ≥ 30.""" n = len(values) cell = {"family": family, "cases": n, "favoured_sum": round(sum(values), 2)} if n >= P.INTERVAL_MIN_N: cell["rate"] = round(statistics.fmean(values), 3) else: cell["rate"] = None cell["state"] = (f"NO-RATE: N is {n}, below the registered N ≥ {P.INTERVAL_MIN_N} " "(PREREG §9 — counts with denominators, no percentages)") return cell def verdict_paragraph(*, pooled: float | None, covers_half: bool, boot: dict, n: int, favoured: int, other: int, families: int, floor: dict) -> str: """The one paragraph the page may print about the head-to-head, in the state it is actually in. Four states, and every number in it comes from the arguments: the panel floor (nothing else may be said), no data, an interval that covers 0.5 (the registered null paragraph), and a separation — which itself splits on whether §9 permits an interval at this n (audit part 2, B1/B3/B5). """ if not floor.get("pass"): return (f"{floor.get('state') or 'NOT-COLLECTED — PANEL-FLOOR'} — " f"{floor.get('why') or ''}".strip(" —") + ". No preference rate, no interval and no ordering may be printed for this " "comparison (PREREG §7's panel floor).") if pooled is None or n == 0: return ("NOT-RUN — no collapsed observation was collected, so the head-to-head has no " "reading to publish.") if boot.get("state") != "PERCENTILE": return SEPARATION_SENTENCE_NO_INTERVAL.format( families=families, favoured=favoured if pooled > 0.5 else other, n=n, min_n=P.INTERVAL_MIN_N) if covers_half: return NULL_PARAGRAPH.format(n=n, lo=boot["lo"], hi=boot["hi"]) return SEPARATION_SENTENCE.format( families=families, favoured=favoured if pooled > 0.5 else other, n=n, lo=boot["lo"], hi=boot["hi"]) def score(*, out_dir: Path | None = None, arm_a: str | None = None, bootstrap_seed: int = 0) -> dict: import pairwise_build as PB key = load_key(out_dir) rows = load_judge_rows(out_dir) manifest = load_manifest(out_dir) arm_a = arm_a or PB.FRONTIER_ARMS[0] other = [a for a in PB.FRONTIER_ARMS if a != arm_a][0] observations, census = collapse(rows, key, arm_a=arm_a) per_case: dict[str, list[float]] = defaultdict(list) per_seat: dict[str, list[float]] = defaultdict(list) seat_family = {r["seat"]: r["family"] for r in rows} for (seat, case), value in observations.items(): per_case[case].append(value) per_seat[seat].append(value) case_means = {c: statistics.fmean(v) for c, v in per_case.items()} P.refuse_unless(len(case_means) <= P.LEG_H_COMPARISONS, got=len(case_means), want=f"≤ {P.LEG_H_COMPARISONS}", section="PREREG §5", what="cases with at least one collapsed observation") pooled = statistics.fmean(case_means.values()) if case_means else None boot = cluster_bootstrap(case_means, seed=bootstrap_seed) if case_means else {"state": "NO-DATA"} favoured_cases = sum(1 for v in case_means.values() if v > 0.5) tied_cases = sum(1 for v in case_means.values() if v == 0.5) other_cases = sum(1 for v in case_means.values() if v < 0.5) floor = JS.panel_floor_verdict(rows) # The panel word is MEASURED: the families that actually carried a collected verdict here. families_carrying = sorted({r["family"] for r in rows if r.get("collection_state") == sr.COLLECTED and r.get("verdict")}) # THE RECUSAL JOIN. Expected empty, and asserted so rather than assumed (PREREG §7). recused = sorted({seat for seat, fam in seat_family.items() if fam in {"anthropic", "openai"}}) covers_half = (boot.get("state") == "PERCENTILE" and boot["lo"] <= 0.5 <= boot["hi"]) out = { "round": P.ROUND_ID, "leg": "H", "arms": {"rate_is_about": arm_a, "the_other": other}, "registered_shape": {"comparisons": P.LEG_H_COMPARISONS, "orders": P.LEG_H_ORDERS, "seats": P.PANEL_SEATS, "judge_calls": P.LEG_H_JUDGE_CALLS, "judge_reps": P.JUDGE_REPS}, "collection_census": census, # PREREG A16 (4): the panel census, split by seat. 55 rows that did not carry are one seat's # 55 rows, and the page says which. "seat_census": seat_census(rows, key), # PREREG A16 (4) / A11: copied from the sheet manifest, which is the only file that knows. # A manifest built before the A11 append carries no such cases, and zero is then the truth. "cases_added_under_A11": manifest.get("cases_added_under_A11", 0), "cases_added_under_A11_ids": manifest.get("cases_added_under_A11_ids", []), "collapse_rule": ("the two orders of each (judge, case) comparison are COLLAPSED to one " "observation (mean; A-preferred 1, tie 0.5, B-preferred 0) before " "anything is computed — PREREG §5"), "cases_with_observations": len(case_means), "observations": len(observations), "pooled_preference_rate": (round(pooled, 3) if pooled is not None and len(case_means) >= P.INTERVAL_MIN_N else None), "pooled_preference_state": (None if len(case_means) >= P.INTERVAL_MIN_N else f"NO-RATE: the independent unit N is {len(case_means)}, below " f"the registered N ≥ {P.INTERVAL_MIN_N}; the counts print with " "their denominator (PREREG §9)"), "pooled_preference_mean_unrounded": pooled, "pooled_denominator": f"{len(case_means)} cases", "panel_families_carrying": {"count": len(families_carrying), "families": families_carrying}, "per_case_counts": {"favoured_" + arm_a: favoured_cases, "tied": tied_cases, "favoured_" + other: other_cases, "denominator": len(case_means)}, "cluster_bootstrap": boot, # audit part 2, MAJOR 4: a bare proportion at N < 30 is §9-forbidden. The rate is guarded at # the source — the count and its denominator are always there, the proportion only when the # unit N clears the registered line. "per_judge_rates": {seat: guarded_rate(v, seat_family.get(seat)) for seat, v in sorted(per_seat.items())}, "order_flip": order_flip_rate(rows, key, arm_a=arm_a), "per_case_table": [{"case_id": c, "rate": round(v, 3), "judges": len(per_case[c])} for c, v in sorted(case_means.items())], #: PREREG A19 (4): the shape of the per-case votes, so the page prints counts and never an #: adjective. A collapsed rate of exactly 0 or 1 means every carrying judge, in both orders, #: took the same side. "per_case_shape": { "within_0_1_of_half": sum(1 for v in case_means.values() if abs(v - 0.5) <= 0.1), "unanimous_for_" + arm_a: sum(1 for v in case_means.values() if v == 1.0), "unanimous_for_" + other: sum(1 for v in case_means.values() if v == 0.0), "denominator": len(case_means), "rule": ("over the collapsed per-case rates: within_0_1_of_half counts |rate − 0.5| ≤ 0.1; " "unanimous counts a rate of exactly 1.0 (every carrying judge, both orders, for the " "A arm) or exactly 0.0 (for the other)")}, "panel_floor": floor, "recusal_join": {"recused_seats": recused, "expected": [], "note": "no seat shares a family with either frontier arm, and the local " "seat is not in Leg H (PREREG §7; A3 ANSWERED (8))"}, "sensitivity_cut": _sensitivity(rows, key, arm_a=arm_a, bootstrap_seed=bootstrap_seed), "forbidden_by_registration": ["Bradley-Terry", "any pairwise-derived rating", "a ranked table", "a crown card", "best / beats / wins / loses to"], "verdict_paragraph": verdict_paragraph( pooled=pooled, covers_half=covers_half, boot=boot, n=len(case_means), favoured=favoured_cases, other=other_cases, families=len(families_carrying), floor=floor), "interval_covers_half": covers_half, "stamped_utc": sr.utc_stamp(), } if recused != []: raise SystemExit( f"the recusal join is not empty: {recused}. PREREG §7 expects zero recused cells in " "Leg H; a seat from either arm's family sitting here is a design breach, not a cell to " "drop quietly." ) return out def _sensitivity(rows: list[dict], key: dict, *, arm_a: str, bootstrap_seed: int) -> dict: """The same computation with every cell whose judge said it RECOGNISED a system dropped.""" kept = [r for r in rows if not sr.recognised_flag(r)] dropped = len(rows) - len(kept) obs, _ = collapse(kept, key, arm_a=arm_a) per_case: dict[str, list[float]] = defaultdict(list) for (_seat, case), value in obs.items(): per_case[case].append(value) case_means = {c: statistics.fmean(v) for c, v in per_case.items()} return { "dropped_rows": dropped, "cases": len(case_means), "pooled_preference_rate": (round(statistics.fmean(case_means.values()), 3) if case_means else None), "cluster_bootstrap": (cluster_bootstrap(case_means, seed=bootstrap_seed) if case_means else {"state": "NO-DATA"}), "what_it_is": "the headline recomputed with every cell a judge said it recognised removed " "(PREREG §7's sensitivity cut); it publishes beside the headline, not instead", } def plan_lines() -> list[str]: return [ "LEG H SCORER — plan only; a scorer never makes a call", f" collapse {P.LEG_H_ORDERS} orders → 1 observation per (judge, case)", f" pool over {P.LEG_H_COMPARISONS} cases · cluster bootstrap {P.BOOTSTRAP_RESAMPLES} " f"resamples, percentile, t({P.LEG_H_COMPARISONS - 1}) noted", f" per-judge rates · order-flip rate · per-case table · sensitivity cut · recusal join", " NO Bradley-Terry, no rating, no ranked table (PREREG §9 forbids them by name)", f"TOTALS H {P.LEG_H_JUDGE_CALLS} judge calls scored · 0 calls made by this file", ] def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--plan", action="store_true") ap.add_argument("--score", action="store_true") ap.add_argument("--dir", default=None) ap.add_argument("--arm-a", default=None) ap.add_argument("--bootstrap-seed", type=int, default=0) ap.add_argument("--out", default=None) args = ap.parse_args(argv) if args.plan or not args.score: print("\n".join(plan_lines())) return 0 out = score(out_dir=Path(args.dir) if args.dir else None, arm_a=args.arm_a, bootstrap_seed=args.bootstrap_seed) text = json.dumps(out, indent=1, ensure_ascii=False) if args.out: Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text(text + "\n", encoding="utf-8") print(f"wrote {args.out}") print(text[:3000]) return 0 if __name__ == "__main__": raise SystemExit(main())