#!/usr/bin/env python3 """Arm A2-id — Implementation B (pure-Python aggregation; pyarrow used only to read parquet bytes; numpy used only for the P9 permutation null, per prereg §6). Written independently of impl_A.py from the preregistration text. Usage: python impl_B.py (run from the work/ directory)""" import json, sys import pyarrow.parquet as pq C3 = { "swe-bench": "Issue Resolution", "swe-bench-multimodal": "Frontend", "commit0": "Greenfield", "swt-bench": "Testing", "gaia": "Information Gathering", } def slug(cat): return cat.lower().replace(" ", "_") NOANT = (["sdk_version", "openness", "country", "supports_vision", "release_date", "average_runtime"] + [slug(c) + "_runtime" for c in C3.values()] + [slug(c) + "_logs_url" for c in C3.values()] + [slug(c) + "_visualization_url" for c in C3.values()]) CATS = ["agent_name", "agent_type", "language_model"] ST1, ST2, CT1, CT2 = 0.05, 0.5, 0.005, 0.05 SEED, NDRAWS = 20260725, 1000 def load(path): t = pq.read_table(path) d = t.to_pydict() cols = t.schema.names n = t.num_rows return cols, [{c: d[c][i] for c in cols} for i in range(n)] def med(xs): s = sorted(xs) n = len(s) if not n: return None return s[n // 2] if n % 2 else 0.5 * (s[n // 2 - 1] + s[n // 2]) def r9(x): return None if x is None else round(x, 9) def tnum(rec, pub, dp, t1, t2): if rec is None: return None if round(rec, dp) == pub: return 0 d = abs(rec - pub) return 1 if d <= t1 else (2 if d <= t2 else "div") def main(outp): tcols, trows = load("test.parquet") _, irows = load("instances.parquet") out = {"meta": {"impl": "B", "prereg": "PREREG_A2-id_2026-07-25.md"}} # join-key search def keyed(rows, fn): return [fn(r) for r in rows] KF = { "K1_id": lambda r: r["id"], "K2_language_model": lambda r: r["language_model"], "K3_agentname_lm": lambda r: r["agent_name"] + "\x1f" + r["language_model"], "K4_id_lower": lambda r: r["id"].lower(), } cands = {} for name, fn in KF.items(): tk = keyed(trows, fn) ik = set(keyed(irows, fn)) cands[name] = {"coverage": sum(1 for k in tk if k in ik) / len(tk), "test_key_unique": len(set(tk)) == len(tk), "n_instance_groups": len(ik)} order = ["K1_id", "K2_language_model", "K3_agentname_lm", "K4_id_lower"] ranked = sorted(order, key=lambda n: (-cands[n]["coverage"], not cands[n]["test_key_unique"], order.index(n))) winner, runner = ranked[0], ranked[1] out["join"] = {"candidates": cands, "winner": winner, "runner_up": runner} kf = KF[winner] for r in trows: r["_k"] = kf(r) for r in irows: r["_k"] = kf(r) # column-family mapping pairs = sorted({(r["benchmark"], r["category"]) for r in irows}) c1, amb = {}, [] for b, c in pairs: if b in c1 and c1[b] != c: amb.append(b) else: c1.setdefault(b, c) out["colmap"] = {"C1_pairs": [list(p) for p in pairs], "C1_ambiguous_benchmarks": amb, "agrees_with_C2_README_and_C3_code": {b: c1.get(b) == c for b, c in C3.items()}, "mapping_used": {b: slug(c) for b, c in c1.items() if c is not None}} igroups = {r["_k"] for r in irows} tkeys = {r["_k"] for r in trows} out["instances_side"] = { "n_rows": len(irows), "distinct_benchmarks": sorted({r["benchmark"] for r in irows}), "groups_without_test_row": sorted(igroups - tkeys), "test_rows_without_group": sorted(tkeys - igroups), "n_groups": len(igroups), "resolved_nulls": sum(1 for r in irows if r["resolved"] is None), "cost_nulls": sum(1 for r in irows if r["cost"] is None), } # group statistics GB, GM = {}, {} for r in irows: kb = (r["_k"], r["benchmark"]) g = GB.setdefault(kb, {"n_all": 0, "n_obs": 0, "r_true": 0, "costs": []}) g["n_all"] += 1 if r["resolved"] is not None: g["n_obs"] += 1 if r["resolved"] is True: g["r_true"] += 1 if r["cost"] is not None: g["costs"].append(float(r["cost"])) m = GM.setdefault(r["_k"], {"vals": {c: set() for c in CATS}, "counts": {c: {} for c in CATS}, "benchmarks": set(), "bench_obs": set()}) m["benchmarks"].add(r["benchmark"]) if r["resolved"] is not None: m["bench_obs"].add(r["benchmark"]) for c in CATS: v = r[c] if v is not None: m["vals"][c].add(v) m["counts"][c][v] = m["counts"][c].get(v, 0) + 1 columns, PORDER = {}, {} pub_of = {c: {r["id"]: r[c] for r in trows} for c in tcols} def emit(col, family, dp, t1, t2, polfn, pnames, primary): pols = {} for pn in pnames: per_row = {} cnt = {0: 0, 1: 0, 2: 0, "div": 0, "undef": 0} for r in trows: rec = polfn(r, pn) pub = pub_of[col][r["id"]] if family in ("count",): t = None if rec is None else (0 if rec == pub else "div") d = None if rec is None else r9(rec - pub) elif family == "categorical": if rec == "___MULTI___": per_row[r["id"]] = [None, polfn(r, pn + "#note"), "div"] cnt["div"] += 1 continue t = None if rec is None else (0 if rec == pub else "div") d = None else: t = tnum(rec, pub, dp, t1, t2) d = None if rec is None else r9(rec - pub) per_row[r["id"]] = [rec, d, t] cnt[t if t in (0, 1, 2, "div") else "undef"] += 1 pols[pn] = {"per_row": per_row, "t0": cnt[0], "t1": cnt[1], "t2": cnt[2], "div": cnt["div"], "undef": cnt["undef"]} vals = [pub_of[col][r["id"]] for r in trows] prim = pols[primary]["per_row"] deltas = {rid: abs(v[1]) for rid, v in prim.items() if v[1] is not None and not isinstance(v[1], str)} mx = max(deltas.values()) if deltas else None columns[col] = { "family": family, "dp": dp, "published_range": None if family == "categorical" else [float(min(vals)), float(max(vals))], "zero_variance": len(set(vals)) == 1, "policies": pols, "primary": primary, "max_abs_delta": mx, "max_abs_rows": sorted([rid for rid, d in deltas.items() if d == mx]) if mx is not None else [], } PORDER[col] = pnames inv = {slug(c): b for b, c in c1.items()} for cat in C3.values(): s = slug(cat) b = inv.get(s) def spol(r, pn, b=b): g = GB.get((r["_k"], b)) if g is None: return None if pn == "S1_pct_Dall": return 100.0 * g["r_true"] / g["n_all"] if g["n_all"] else None if pn == "S2_pct_Dobs": return 100.0 * g["r_true"] / g["n_obs"] if g["n_obs"] else None if pn == "S1_frac_Dall": return g["r_true"] / g["n_all"] if g["n_all"] else None if pn == "S2_frac_Dobs": return g["r_true"] / g["n_obs"] if g["n_obs"] else None emit(s + "_score", "score", 1, ST1, ST2, spol, ["S1_pct_Dall", "S2_pct_Dobs", "S1_frac_Dall", "S2_frac_Dobs"], "S1_pct_Dall") def cpol(r, pn, b=b): g = GB.get((r["_k"], b)) if g is None: return None c, na = g["costs"], g["n_all"] if pn == "C_mean_excl": return sum(c) / len(c) if c else None if pn == "C_mean_zero": return sum(c) / na if na else None if pn == "C_median_excl": return med(c) if pn == "C_median_zero": return med(c + [0.0] * (na - len(c))) if na else None if pn == "C_sum_over_resolved": return sum(c) / g["r_true"] if g["r_true"] else None emit(s + "_cost", "cost", 4, CT1, CT2, cpol, ["C_mean_excl", "C_mean_zero", "C_median_excl", "C_median_zero", "C_sum_over_resolved"], "C_mean_excl") def as_pol(r, pn): k = r["_k"] s1, s2, NA, NO, RT = [], [], 0, 0, 0 for b in C3: g = GB.get((k, b)) if g is None: continue if g["n_all"]: s1.append(100.0 * g["r_true"] / g["n_all"]) if g["n_obs"]: s2.append(100.0 * g["r_true"] / g["n_obs"]) NA += g["n_all"]; NO += g["n_obs"]; RT += g["r_true"] if pn == "AS_macro_raw_Dall": return sum(s1) / len(s1) if s1 else None if pn == "AS_macro_raw_Dobs": return sum(s2) / len(s2) if s2 else None if pn == "AS_macro_pub1dp": return sum(round(x, 1) for x in s1) / len(s1) if s1 else None if pn == "AS_micro_Dall": return 100.0 * RT / NA if NA else None if pn == "AS_micro_Dobs": return 100.0 * RT / NO if NO else None if pn == "AS_macro_all5_zero": return sum(s1) / 5.0 if s1 else None emit("average_score", "score", 2, ST1, ST2, as_pol, ["AS_macro_raw_Dall", "AS_macro_raw_Dobs", "AS_macro_pub1dp", "AS_micro_Dall", "AS_micro_Dobs", "AS_macro_all5_zero"], "AS_macro_raw_Dall") def ac_pol(r, pn): k = r["_k"] cm, allc, NA = [], [], 0 for b in C3: g = GB.get((k, b)) if g is None: continue if g["costs"]: cm.append(sum(g["costs"]) / len(g["costs"])) allc += g["costs"]; NA += g["n_all"] if pn == "AC_macro_raw": return sum(cm) / len(cm) if cm else None if pn == "AC_macro_4dp": return sum(round(x, 4) for x in cm) / len(cm) if cm else None if pn == "AC_micro_excl": return sum(allc) / len(allc) if allc else None if pn == "AC_micro_zero": return sum(allc) / NA if NA else None if pn == "AC_macro_all5_zero": return sum(cm) / 5.0 if cm else None emit("average_cost", "cost", 4, CT1, CT2, ac_pol, ["AC_macro_raw", "AC_macro_4dp", "AC_micro_excl", "AC_micro_zero", "AC_macro_all5_zero"], "AC_macro_raw") def cc_pol(r, pn): m = GM.get(r["_k"]) if m is None: return None if pn == "CC1_any_row": return len([b for b in m["benchmarks"] if b in C3]) if pn == "CC2_any_obs": return len([b for b in m["bench_obs"] if b in C3]) emit("categories_completed", "count", None, None, None, cc_pol, ["CC1_any_row", "CC2_any_obs"], "CC1_any_row") for cc in CATS: def upol(r, pn, cc=cc): m = GM.get(r["_k"]) if m is None: return None if pn == "U_unique": vs = sorted(m["vals"][cc]) return vs[0] if len(vs) == 1 else "___MULTI___" if pn == "U_unique#note": return f"multiplicity={len(m['vals'][cc])}" if pn == "U_major": cnts = m["counts"][cc] mxc = max(cnts.values()) return sorted([v for v, n in cnts.items() if n == mxc])[0] emit(cc, "categorical", None, None, None, upol, ["U_unique", "U_major"], "U_unique") columns["id"] = {"family": "join_key", "dp": None, "published_range": None, "zero_variance": False, "policies": {}, "primary": None, "max_abs_delta": None, "max_abs_rows": []} PORDER["id"] = [] # cell accounting by_col = {} totals = {"reconciled": 0, "divergent": 0, "no_antecedent": 0, "structural": 0} tiers = {0: 0, 1: 0, 2: 0} for col in tcols: cls = {} if col in NOANT: cls = {r["id"]: "no_antecedent" for r in trows} elif col == "id" and winner in ("K1_id", "K4_id_lower"): cls = {r["id"]: "structural" for r in trows} elif columns[col]["zero_variance"]: cls = {r["id"]: "structural" for r in trows} else: for r in trows: rid = r["id"] best, bestp = None, None for pn in PORDER[col]: t = columns[col]["policies"][pn]["per_row"][rid][2] if t in (0, 1, 2): if best is None or best == "div" or t < best: best, bestp = t, pn elif t == "div" and best is None: best, bestp = "div", pn if best in (0, 1, 2): cls[rid] = f"reconciled_T{best}:{bestp}" tiers[best] += 1 elif best == "div": cls[rid] = "divergent" else: cls[rid] = "no_antecedent(all_policies_undefined)" c = {"reconciled": 0, "divergent": 0, "no_antecedent": 0, "structural": 0} for v in cls.values(): key = ("reconciled" if v.startswith("reconciled") else "divergent" if v == "divergent" else "no_antecedent" if v.startswith("no_antecedent") else "structural") c[key] += 1 by_col[col] = {"counts": c, "cells": cls} for kk in totals: totals[kk] += c[kk] totals["total"] = sum(totals[k] for k in ["reconciled", "divergent", "no_antecedent", "structural"]) totals["reconciled_by_tier"] = {f"T{k}": v for k, v in tiers.items()} out["cells"] = {"totals": totals, "by_column": by_col} # permutation null (numpy permitted here only) import numpy as np rng = np.random.default_rng(SEED) perms = [rng.permutation(len(trows)) for _ in range(NDRAWS)] nulls = {} addressed = [c for c in tcols if c not in NOANT and c != "id"] for col in addressed: info = columns[col] if info["zero_variance"]: nulls[col] = {"structural_zero_variance": True, "note": "null distribution degenerate; floor structural, ratios undefined"} continue fam, dp = info["family"], info["dp"] t1 = ST1 if fam == "score" else CT1 t2 = ST2 if fam == "score" else CT2 pr = info["policies"][info["primary"]]["per_row"] pub = [pub_of[col][r["id"]] for r in trows] rec = [pr[r["id"]][0] for r in trows] rec = [None if v == "___MULTI___" else v for v in rec] def stat(perm): t0, devs = 0, [] for i in range(len(trows)): rv = rec[perm[i]] if rv is None: continue if fam in ("count", "categorical"): t0 += 1 if rv == pub[i] else 0 else: if round(rv, dp) == pub[i]: t0 += 1 devs.append(abs(rv - pub[i])) return t0, (sum(devs) / len(devs) if devs else None) ot0, omad = stat(list(range(len(trows)))) nt0, nmad = [], [] for p in perms: a, b = stat(p) nt0.append(a) if b is not None: nmad.append(b) nulls[col] = {"observed_t0": ot0, "observed_mad": omad, "t0_median": float(np.median(nt0)), "t0_p5": float(np.percentile(nt0, 5)), "t0_p95": float(np.percentile(nt0, 95)), "mad_median": float(np.median(nmad)) if nmad else None, "mad_p5": float(np.percentile(nmad, 5)) if nmad else None, "n_draws": NDRAWS, "seed": SEED} out["nulls"] = nulls # tie census def census(vals): n = len(vals) pairs = sum(1 for i in range(n) for j in range(i + 1, n) if vals[i] == vals[j]) s = sorted(vals, reverse=True) return {"n": n, "distinct": len(set(vals)), "tied_pairs": pairs, "adjacent_ties_in_desc_sort": sum(1 for i in range(n - 1) if s[i] == s[i + 1]), "strictly_decidable": pairs == 0} pub_as = [r["average_score"] for r in trows] pr = columns["average_score"]["policies"]["AS_macro_raw_Dall"]["per_row"] rec_as = [pr[r["id"]][0] for r in trows if pr[r["id"]][0] is not None] out["tie_census"] = {"published_average_score": census(pub_as), "recomputed_primary_raw": census(rec_as), "recomputed_primary_2dp": census([round(x, 2) for x in rec_as]), "published_sort_descending": all(pub_as[i] >= pub_as[i + 1] for i in range(len(pub_as) - 1))} # self-identification lm = [str(r["language_model"]).lower() for r in trows] a2x = [i for i, s in enumerate(lm) if s == "claude-fable-5"] a2 = a2x if a2x else [i for i, s in enumerate(lm) if "fable" in s] b2 = [i for i, s in enumerate(lm) if s.startswith("gemini")] srt = sorted(pub_as, reverse=True) def rinfo(i): return {"file_order_rank_1based": i + 1, "rank_by_average_score": srt.index(trows[i]["average_score"]) + 1, "published_row": {k: v for k, v in trows[i].items() if k != "_k"}} out["self_id"] = { "rule": "exact lower(language_model)=='claude-fable-5', fallback substring 'fable'; B2: startswith 'gemini' (pin unknown to this arm)", "A2_rows": {str(i): rinfo(i) for i in a2}, "A2_match_type": "exact" if a2x else ("substring" if a2 else "none"), "B2_candidate_rows": {str(i): rinfo(i) for i in b2}, } out["hash_check"] = {"executed": False, "note": "M-H runs in Impl A only per prereg addition A2"} out["columns"] = columns with open(outp, "w") as f: json.dump(out, f, sort_keys=True, indent=1) f.write("\n") tt = out["cells"]["totals"] print(f"impl_B wrote {outp}; cell totals: {tt}") if __name__ == "__main__": main(sys.argv[1])