"""Audit and select the frozen validation-only FashionMNIST C1 recovery.""" import glob import json import math import os ROOT = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "results") PREFIX = "c1_fmnist_val_v1_" TRAFFIC_SEEDS = (1234, 5678) SIGNALS = ("raw", "matched", "residual", "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 sum(finite) / len(finite) 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 audit_row(path, row): args = row["args"] required = { "mode": "sdil", "dataset": "fmnist", "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": 5000, "split_seed": 2027, "eval_split": "validation", "eval_every": 0, "diagnostics": "alignment", "diagnostics_schedule": "final", "probe_bs": 512, "seed": 0, } 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") != "validation": raise RuntimeError(f"non-validation development 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}") split = row.get("split", {}) if (not split.get("split_from_training_only") or split.get("validation_examples") != 5000 or split.get("evaluation_split") != "validation"): raise RuntimeError(f"invalid validation split {path}: {split}") 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}") def main(): paths = sorted(glob.glob(os.path.join(ROOT, PREFIX + "*.json"))) rows = [] commits = set() split_hashes = set() for path in paths: with open(path) as handle: row = json.load(handle) audit_row(path, row) rows.append(row) commits.add(row["provenance"]["git_commit"]) split_hashes.add(row["split"]["validation_index_sha256"]) if len(commits) != 1 or len(split_hashes) != 1: raise RuntimeError(f"expected one source and split, got {commits}, {split_hashes}") groups = {} for row in rows: args = row["args"] key = (args["traffic_mode"], args["nuis_rho"], args["traffic_seed"], signal_name(args)) if key in groups: raise RuntimeError(f"duplicate validation row: {key}") groups[key] = row expected = {("none", 0.0, 1234, signal) for signal in SIGNALS[:3]} expected |= {("soma", 0.5, traffic_seed, signal) for traffic_seed in TRAFFIC_SEEDS for signal in SIGNALS} expected |= {("topdown", rho, traffic_seed, signal) for rho in (0.2, 0.5) for traffic_seed in TRAFFIC_SEEDS for signal in SIGNALS} if set(groups) != expected or len(rows) != 27: raise RuntimeError(f"incomplete/unexpected panel: rows={len(rows)}, " f"missing={expected - set(groups)}, extra={set(groups) - expected}") print(f"commit={next(iter(commits))} split={next(iter(split_hashes))} rows={len(rows)}") print("| traffic | rho | projection seed | signal | validation (%) | cos(r,-g) | R2 |") print("|:---|---:|---:|:---|---:|---:|---:|") order = [("none", 0.0, 1234)] order += [("soma", 0.5, seed) for seed in TRAFFIC_SEEDS] order += [("topdown", rho, seed) for rho in (0.2, 0.5) for seed in TRAFFIC_SEEDS] for family, rho, traffic_seed in order: signals = SIGNALS[:3] if family == "none" else SIGNALS for signal in signals: row = groups[(family, rho, traffic_seed, signal)] final = row["final"] r2 = "--" if family == "none" else f"{finite_mean(final['traffic_r2']):.3f}" print(f"| {family} | {rho:g} | {traffic_seed} | {signal} | " f"{100 * final['val_acc']:.3f} | {finite_mean(final['cos_r_negg']):+.3f} | {r2} |") no_key = ("none", 0.0, 1234) no_acc = 100 * groups[no_key + ("residual",)]["final"]["val_acc"] no_changes = { signal: no_acc - 100 * groups[no_key + (signal,)]["final"]["val_acc"] for signal in ("raw", "matched") } no_ok = min(no_changes.values()) >= -0.5 print(f"no-traffic residual-minus controls={no_changes}: {'PASS' if no_ok else 'FAIL'}") checks = {} for family, rho in (("soma", 0.5), ("topdown", 0.2), ("topdown", 0.5)): realization_checks = [] gains = [] for traffic_seed in TRAFFIC_SEEDS: prefix = (family, rho, traffic_seed) residual = groups[prefix + ("residual",)] matched = groups[prefix + ("matched",)] gain = 100 * (residual["final"]["val_acc"] - matched["final"]["val_acc"]) residual_acc = 100 * residual["final"]["val_acc"] r2 = finite_mean(residual["final"]["traffic_r2"]) alignment = finite_mean(residual["final"]["cos_r_negg"]) passed = (gain >= 2.0 and residual_acc >= no_acc - 3.0 and r2 >= 0.5 and alignment >= 0.1) gains.append(gain) realization_checks.append(passed) print(f"{prefix}: gain={gain:+.3f}, residual={residual_acc:.3f}, " f"no-traffic gap={residual_acc - no_acc:+.3f}, R2={r2:.3f}, " f"alignment={alignment:+.3f}: {'PASS' if passed else 'FAIL'}") checks[(family, rho)] = (all(realization_checks), min(gains)) soma_ok = checks[("soma", 0.5)][0] eligible = [(rho, checks[("topdown", rho)][1]) for rho in (0.2, 0.5) if checks[("topdown", rho)][0]] if soma_ok and no_ok and eligible: selected_rho, worst_gain = max(eligible, key=lambda item: (item[1], -item[0])) print(f"C1 FashionMNIST validation gate: PASS; selected topdown rho={selected_rho:g} " f"by highest worst-projection gain ({worst_gain:+.3f} points)") return print("C1 FashionMNIST validation gate: FAIL; do not evaluate the FashionMNIST test set") raise SystemExit(1) if __name__ == "__main__": main()