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
|
#!/usr/bin/env python3
"""Apply the frozen Oral-A-v3 causal-capture selector and gate."""
import argparse
import glob
import json
import math
import os
SPLIT_HASH = "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b"
ORACLE_EARLY = 0.02396580002561188
def load(path):
with open(path) as handle:
record = json.load(handle)
args = record["args"]
expected = {
"mode": "sdil", "depth": 20, "width": 16, "seed": 0,
"epochs": 0, "train_limit": 10000, "val_examples": 5000,
"a_warmup_steps": 400, "pert_directions": 1, "pert_every": 4,
"pert_sigma": 0.01, "perturb_seed": 1000,
"normalization": "batchnorm", "vectorizer_mode": "channel_gated",
"a_scale": 0.0, "alignment_probe": 64,
}
for key, value in expected.items():
if args.get(key) != value:
raise ValueError(
f"{path}: {key}={args.get(key)!r}, expected {value!r}")
if record["provenance"]["git_tracked_dirty"]:
raise ValueError(f"tracked-dirty result: {path}")
if record["split"]["validation_index_sha256"] != SPLIT_HASH:
raise ValueError(f"split drift: {path}")
mode = args["apical_calibration_mode"]
expected_space = {
"channel_subspace": "channel_basis_moments",
"vectorizer_subspace": "vectorizer_parameter_gradients",
}[mode]
if record.get("calibration_metric_space") != expected_space:
raise ValueError(f"metric-space drift: {path}")
diagnostics = record.get("diagnostics")
warmup = record.get("apical_warmup", {}).get("mean")
if diagnostics is None or warmup is None:
raise ValueError(f"missing diagnostics/warmup aggregate: {path}")
values = diagnostics["teaching_negative_gradient_cosine"]
early_count = max(1, len(values) // 3)
metrics = {
"early_third_alignment": sum(values[:early_count]) / early_count,
"all_layer_alignment": sum(values) / len(values),
"mean_calibration_mse": warmup["calibration_mse"],
"mean_target_power": warmup["target_power"],
"mean_prediction_target_cosine": warmup["prediction_target_cosine"],
"mean_parameter_update_rms": warmup["parameter_update_rms"],
}
finite = (record["final"]["finite"]
and all(math.isfinite(value) for value in metrics.values()))
return {
"path": path, "source_commit": record["provenance"]["git_commit"],
"calibration_mode": mode, "eta_A": float(args["eta_A"]),
"metric_space": expected_space, "metrics": metrics, "finite": finite,
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--input", default="results/oral_a_v3_calibration")
parser.add_argument("--out", default="results/oral_a_v3_calibration_gate.json")
args = parser.parse_args()
rows = [load(path) for path in sorted(glob.glob(
os.path.join(args.input, "*.json")))]
observed = {(row["calibration_mode"], row["eta_A"]) for row in rows}
expected = {("channel_subspace", 0.01)} | {
("vectorizer_subspace", rate) for rate in (0.01, 0.1, 1.0)}
if observed != expected or len(rows) != len(expected):
raise ValueError(
f"incomplete v3 grid: missing={expected-observed}, extra={observed-expected}")
if len({row["source_commit"] for row in rows}) != 1:
raise ValueError("v3 calibration source commits differ")
reference = next(row for row in rows
if row["calibration_mode"] == "channel_subspace")
candidates = [row for row in rows
if row["calibration_mode"] == "vectorizer_subspace"
and row["finite"]]
candidates.sort(key=lambda row: (
-row["metrics"]["early_third_alignment"],
-row["metrics"]["all_layer_alignment"], row["eta_A"]))
selected = candidates[0] if candidates else None
checks = {
"all_four_records_finite": all(row["finite"] for row in rows),
"v3_candidate_selected": selected is not None,
}
if selected is not None:
metrics = selected["metrics"]
checks.update({
"v3_early_third_at_least_0.01": (
metrics["early_third_alignment"] >= 0.01),
"v3_all_layer_at_least_0.05": (
metrics["all_layer_alignment"] >= 0.05),
"v3_early_gain_over_v2_at_least_0.01": (
metrics["early_third_alignment"]
- reference["metrics"]["early_third_alignment"] >= 0.01),
"v3_reaches_60pct_of_family_oracle": (
metrics["early_third_alignment"] >= 0.60 * ORACLE_EARLY),
})
else:
checks.update({
"v3_early_third_at_least_0.01": False,
"v3_all_layer_at_least_0.05": False,
"v3_early_gain_over_v2_at_least_0.01": False,
"v3_reaches_60pct_of_family_oracle": False,
})
passed = all(checks.values())
output = {
"protocol": "oral_a_v3_vectorizer_causal_capture_v1",
"status": "passed" if passed else "failed",
"checks": checks, "rows": rows, "matched_v2_reference": reference,
"selected_v3": selected, "family_oracle_early_third": ORACLE_EARLY,
"confirmation_test_seeds_touched": False,
"review_score_before": 5, "review_score_after": 5,
"score_change_rule": "mechanics/calibration alone cannot raise score",
}
os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
with open(args.out, "w") as handle:
json.dump(output, handle, indent=2, sort_keys=True)
handle.write("\n")
print(json.dumps({
"status": output["status"], "checks": checks,
"reference": reference, "selected_v3": selected,
}, indent=2))
if __name__ == "__main__":
main()
|
