1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
"""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()
|
