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
|
"""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()
|
