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
|
"""Audit the frozen 120-run C2 local-rule validation 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 = "c2_local_val_v1_"
METHODS = ("bp", "fa", "dfa", "sdil")
DEPTHS = (1, 4)
TASK_SEEDS = (0, 1, 2)
MODEL_SEEDS = (0, 1, 2, 3, 4)
def audit_row(path, row):
args = row["args"]
required = {
"dataset": "tentmap", "width": 8, "act": "relu", "residual": 1,
"epochs": 80, "batch_size": 256, "eta": 0.03, "momentum": 0.9,
"eta_A": 0.02, "eta_P": 0.002,
"pert_sigma": 0.01, "pert_every": 4, "pert_ndirs": 1,
"pert_mode": "simultaneous", "traffic_mode": "none", "nuis_rho": 0.0,
"use_residual": 1, "learn_A": 1, "learn_P": 1, "p_neutral": 1,
"task_train_examples": 10000, "task_test_examples": 5000,
"task_levels": 2, "task_n_in": 1,
"val_examples": 2000, "split_seed": 2027,
"eval_split": "validation", "eval_every": 0,
"diagnostics": "alignment", "diagnostics_schedule": "final", "probe_bs": 512,
}
mismatches = {key: (args.get(key), expected) for key, expected in required.items()
if args.get(key) != expected}
expected_lesion = 1.0 / 3.0 if args["depth"] == 4 else 0.0
if abs(args.get("residual_lesion_fraction", 0.0) - expected_lesion) > 1e-12:
mismatches["residual_lesion_fraction"] = (
args.get("residual_lesion_fraction"), expected_lesion)
if mismatches:
raise RuntimeError(f"protocol mismatch {path}: {mismatches}")
if row["final"].get("eval_split") != "validation":
raise RuntimeError(f"test-contaminated validation 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") != 2000
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}")
if args["depth"] == 4:
lesion = row["final"].get("residual_lesion")
if not lesion or lesion.get("lesioned_layers") != [3]:
raise RuntimeError(f"incorrect d4 lesion: {path}")
def mean_sd(values):
return statistics.mean(values), statistics.stdev(values)
def main():
paths = sorted(glob.glob(os.path.join(ROOT, PREFIX + "*.json")))
rows = {}
commits = set()
split_hashes = {}
for path in paths:
with open(path) as handle:
row = json.load(handle)
audit_row(path, row)
args = row["args"]
key = (args["mode"], args["depth"], args["task_seed"], args["seed"])
if key in rows:
raise RuntimeError(f"duplicate row: {key}")
rows[key] = row
commits.add(row["provenance"]["git_commit"])
split_hashes.setdefault(args["task_seed"], set()).add(
row["split"]["validation_index_sha256"])
expected = {(method, depth, task_seed, model_seed)
for method in METHODS for depth in DEPTHS
for task_seed in TASK_SEEDS for model_seed in MODEL_SEEDS}
if set(rows) != expected or len(commits) != 1:
raise RuntimeError(f"incomplete/mixed panel: rows={len(rows)}, "
f"missing={expected - set(rows)}, extra={set(rows) - expected}, "
f"commits={commits}")
if any(len(hashes) != 1 for hashes in split_hashes.values()):
raise RuntimeError(f"methods did not share splits within tasks: {split_hashes}")
print(f"commit={next(iter(commits))} rows={len(rows)}")
print("| method | depth | validation (%) | depth gain (points) | lesion drop (points) |")
print("|:---|---:|---:|---:|---:|")
method_accs = {}
method_gains = {}
for method in METHODS:
for depth in DEPTHS:
values = [100 * rows[(method, depth, task_seed, model_seed)]["final"]["val_acc"]
for task_seed in TASK_SEEDS for model_seed in MODEL_SEEDS]
method_accs[(method, depth)] = values
gains = [deep - shallow for shallow, deep in zip(
method_accs[(method, 1)], method_accs[(method, 4)])]
method_gains[method] = gains
for depth in DEPTHS:
acc_mean, acc_sd = mean_sd(method_accs[(method, depth)])
gain_text = "--" if depth == 1 else f"{statistics.mean(gains):+.3f}"
if depth == 1:
lesion_text = "--"
else:
drops = [100 * rows[(method, 4, task_seed, model_seed)]["final"]
["residual_lesion"]["lesion_acc_drop"]
for task_seed in TASK_SEEDS for model_seed in MODEL_SEEDS]
lesion_text = f"{statistics.mean(drops):+.3f}"
print(f"| {method} | {depth} | {acc_mean:.3f} +/- {acc_sd:.3f} | "
f"{gain_text} | {lesion_text} |")
bp_gain = statistics.mean(method_gains["bp"])
sdil_gain = statistics.mean(method_gains["sdil"])
recovery = sdil_gain / bp_gain if bp_gain > 0 else -math.inf
competitor_recoveries = {
method: statistics.mean(method_gains[method]) / bp_gain
for method in ("fa", "dfa")
}
strongest_recovery = max(competitor_recoveries.values())
strongest_deep = max(statistics.mean(method_accs[(method, 4)])
for method in ("fa", "dfa"))
sdil_deep_advantage = statistics.mean(method_accs[("sdil", 4)]) - strongest_deep
sdil_lesions = [100 * rows[("sdil", 4, task_seed, model_seed)]["final"]
["residual_lesion"]["lesion_acc_drop"]
for task_seed in TASK_SEEDS for model_seed in MODEL_SEEDS]
lesion_mean = statistics.mean(sdil_lesions)
lesion_positive = sum(value > 0 for value in sdil_lesions)
comparator_ok = strongest_recovery <= 0.5 or sdil_deep_advantage >= 2.0
passed = (bp_gain >= 5.0 and recovery >= 0.7 and comparator_ok
and lesion_mean >= 2.0 and lesion_positive >= 10)
print(f"BP mean gain={bp_gain:+.3f}; SDIL mean gain={sdil_gain:+.3f}; "
f"recovery={100 * recovery:.1f}%")
print(f"competitor recoveries={competitor_recoveries}; strongest={100 * strongest_recovery:.1f}%")
print(f"SDIL d4 advantage over strongest FA/DFA endpoint={sdil_deep_advantage:+.3f} points")
print(f"SDIL d4 lesion mean={lesion_mean:+.3f}; positive={lesion_positive}/15")
print(f"C2 local validation gate: {'PASS' if passed else 'FAIL'}")
if not passed:
raise SystemExit(1)
if __name__ == "__main__":
main()
|
