"""Audit the frozen C1 mixed-apical-traffic confirmation panel.""" import glob import json import math import os import statistics ROOT = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "results") PREFIX = "c1_confirm_v1_" MODEL_SEEDS = tuple(range(5)) TRAFFIC_FAMILIES = (("soma", 0.5, 1234), ("soma", 0.5, 5678), ("topdown", 0.2, 1234), ("topdown", 0.2, 5678)) SIGNALS = ("raw", "matched", "residual", "taskfit") def mean_sd(values): return (statistics.mean(values), statistics.stdev(values) if len(values) > 1 else 0.0) def signal_name(args): if not args["use_residual"]: return ("matched" if args["raw_scale_control"] == "match_innovation_norm" else "raw") return "residual" if args["p_neutral"] else "taskfit" def finite_mean(values): finite = [value for value in values if value is not None and math.isfinite(value)] if not finite: raise RuntimeError("expected at least one finite diagnostic") return statistics.mean(finite) def audit_row(path, row): args = row["args"] required = { "mode": "sdil", "dataset": "mnist", "depth": 3, "width": 256, "residual": 1, "epochs": 15, "batch_size": 128, "eta": 0.05, "momentum": 0.9, "eta_A": 0.02, "eta_P": 0.05, "pert_sigma": 0.01, "pert_every": 4, "pert_ndirs": 1, "pert_mode": "simultaneous", "learn_A": 1, "learn_P": 1, "p_warmup_steps": 200, "p_warmup_eta": 0.05, "val_examples": 0, "eval_split": "test", "eval_every": 0, "diagnostics": "alignment", "diagnostics_schedule": "final", } mismatches = {key: (args.get(key), value) for key, value in required.items() if args.get(key) != value} if mismatches: raise RuntimeError(f"protocol mismatch {path}: {mismatches}") if row.get("final", {}).get("eval_split") != "test": raise RuntimeError(f"non-test confirmation row: {path}") if any("eval_acc" in step or "cos_r_negg" in step for step in row.get("steps", [])): raise RuntimeError(f"intermediate held-out metric/diagnostic: {path}") protocol = row.get("diagnostic_protocol", {}) if protocol != {"probe_source": "training_prefix", "probe_examples": 512, "schedule": "final"}: raise RuntimeError(f"diagnostic protocol mismatch {path}: {protocol}") if row.get("provenance", {}).get("git_dirty") is not False: raise RuntimeError(f"dirty or unknown source provenance: {path}") if "hardware" not in row or "training_forward_equivalent_examples" not in row.get("cost", {}): raise RuntimeError(f"missing cost/memory audit: {path}") def main(): paths = sorted(glob.glob(os.path.join(ROOT, PREFIX + "*.json"))) rows = [] commits = set() for path in paths: with open(path) as handle: row = json.load(handle) audit_row(path, row) commits.add(row["provenance"]["git_commit"]) rows.append(row) if len(commits) != 1: raise RuntimeError(f"expected one clean source commit, got {commits}") groups = {} for row in rows: args = row["args"] family = (args["traffic_mode"], args["nuis_rho"], args["traffic_seed"]) groups.setdefault((family, signal_name(args)), []).append(row) expected = {(("none", 0.0, 1234), signal) for signal in SIGNALS[:3]} expected |= {(family, signal) for family in TRAFFIC_FAMILIES for signal in SIGNALS} actual = set(groups) bad_counts = {key: len(value) for key, value in groups.items() if len(value) != 5} if actual != expected or bad_counts or len(rows) != 95: raise RuntimeError(f"incomplete/unexpected C1 panel: rows={len(rows)}, " f"missing={expected - actual}, extra={actual - expected}, " f"bad_counts={bad_counts}") for key, group in groups.items(): group.sort(key=lambda row: row["args"]["seed"]) seeds = tuple(row["args"]["seed"] for row in group) if seeds != MODEL_SEEDS: raise RuntimeError(f"unpaired model seeds for {key}: {seeds}") print(f"commit={next(iter(commits))} model_seeds={list(MODEL_SEEDS)} rows={len(rows)}") print("| traffic | rho | traffic seed | signal | test (%) | mean cos(r,-g) | traffic R2 |") print("|:---|---:|---:|:---|---:|---:|---:|") order = (("none", 0.0, 1234),) + TRAFFIC_FAMILIES for family in order: signals = SIGNALS[:3] if family[0] == "none" else SIGNALS for signal in signals: group = groups[(family, signal)] acc = mean_sd([100 * row["final"]["test_acc"] for row in group]) cosine = mean_sd([finite_mean(row["final"]["cos_r_negg"]) for row in group]) r2_values = [] if family[0] != "none": r2_values = [finite_mean(row["final"]["traffic_r2"]) for row in group] r2_text = "--" if not r2_values else f"{mean_sd(r2_values)[0]:.3f}" print(f"| {family[0]} | {family[1]:g} | {family[2]} | {signal} | " f"{acc[0]:.3f} +/- {acc[1]:.3f} | {cosine[0]:+.3f} | {r2_text} |") gate_checks = [] for family in TRAFFIC_FAMILIES: matched = groups[(family, "matched")] residual = groups[(family, "residual")] paired_gain = statistics.mean( 100 * (res["final"]["test_acc"] - raw["final"]["test_acc"]) for res, raw in zip(residual, matched)) mean_r2 = statistics.mean(finite_mean(row["final"]["traffic_r2"]) for row in residual) mean_alignment = statistics.mean(finite_mean(row["final"]["cos_r_negg"]) for row in residual) passed = paired_gain >= 2.0 and mean_r2 >= 0.5 and mean_alignment >= 0.1 gate_checks.append(passed) print(f"{family}: residual-matched={paired_gain:+.3f} points, " f"R2={mean_r2:.3f}, alignment={mean_alignment:+.3f}: " f"{'PASS' if passed else 'FAIL'}") no_family = ("none", 0.0, 1234) no_residual = groups[(no_family, "residual")] no_harms = {} for signal in ("raw", "matched"): comparator = groups[(no_family, signal)] no_harms[signal] = statistics.mean( 100 * (res["final"]["test_acc"] - other["final"]["test_acc"]) for res, other in zip(no_residual, comparator)) no_traffic_pass = min(no_harms.values()) >= -0.5 gate_checks.append(no_traffic_pass) print(f"no-traffic residual-minus controls={no_harms}: " f"{'PASS' if no_traffic_pass else 'FAIL'}") passed = all(gate_checks) print(f"C1 confirmation gate: {'PASS' if passed else 'FAIL'}") if not passed: raise SystemExit(1) if __name__ == "__main__": main()