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
|
#!/usr/bin/env python3
"""Audit and gate the frozen WM-2 short accuracy screen."""
import argparse
import glob
import json
import math
import os
RATES = (0.03, 0.1)
BP_ACCURACY = 0.7494
def load(path):
with open(path) as handle:
record = json.load(handle)
args = record["args"]
expected = {
"mode": "wm", "depth": 20, "width": 16, "seed": 0,
"loader_seed": 0, "epochs": 20, "train_limit": 10000,
"val_examples": 5000, "split_seed": 2027,
"eval_split": "validation", "eval_every": 0,
"augment_train": 1, "lr_schedule": "cosine", "warmup_epochs": 0,
"momentum": 0.9, "weight_decay": 1e-4,
"normalization": "batchnorm", "output_lr": 0.1,
"a_scale": 1.0, "alignment_probe": 32,
"mirror_warmup_steps": 20, "mirror_every": 16,
"mirror_batch_size": 1, "mirror_eta": 0.1,
"mirror_noise_std": 1.0, "mirror_seed": 3000,
}
for key, value in expected.items():
if args.get(key) != value:
raise ValueError(f"{path}: {key} drift")
rate = float(args["lr"])
if rate not in RATES:
raise ValueError(f"{path}: unregistered hidden rate")
if record["provenance"]["git_tracked_dirty"]:
raise ValueError(f"tracked-dirty result: {path}")
if record["evaluation_protocol"]["test_evaluations"]:
raise ValueError(f"test touched: {path}")
if record.get("calibration_metric_space") != "local_parent_child_response":
raise ValueError(f"mirror metric-space drift: {path}")
accuracy = float(record["final"]["accuracy"])
loss = float(record["final"]["loss"])
diagnostics = record["diagnostics"]
early = float(diagnostics["early_third_mean"])
values = diagnostics["teaching_negative_gradient_cosine"]
all_layer = sum(values) / len(values)
finite = (bool(record["final"]["finite"])
and math.isfinite(accuracy + loss + early + all_layer))
return {
"path": path, "lr": rate, "accuracy": accuracy, "loss": loss,
"early_third_alignment": early, "all_layer_alignment": all_layer,
"mean_feedback_forward_cosine": sum(
diagnostics["feedback_forward_cosine"])
/ len(diagnostics["feedback_forward_cosine"]),
"finite": finite,
"total_macs": int(record["work"]["total_macs_estimate"]),
"logical_batch_loss_queries": int(
record["work"]["logical_batch_loss_queries"]),
"peak_memory_allocated_bytes": int(
record["hardware"]["peak_memory_allocated_bytes"]),
"wall_s": float(record["timing"]["total_timed_wall_s"]),
"mirror_events": int(record["counters"]["mirror_events"]),
"source_commit": record["provenance"]["git_commit"],
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--input", default="results/mirror_short")
parser.add_argument("--bp", default="results/oral_a_short/bp_lr0.1.json")
parser.add_argument("--out", default="results/mirror_short_gate.json")
args = parser.parse_args()
rows = [load(path) for path in sorted(
glob.glob(os.path.join(args.input, "*.json")))]
if len(rows) != len(RATES) or {row["lr"] for row in rows} != set(RATES):
raise ValueError("incomplete WM-2 rate grid")
if len({row["source_commit"] for row in rows}) != 1:
raise ValueError("WM-2 source commits differ")
with open(args.bp) as handle:
bp = json.load(handle)
if float(bp["final"]["accuracy"]) != BP_ACCURACY:
raise ValueError("matched BP endpoint drift")
bp_macs = int(bp["work"]["total_macs_estimate"])
finite = [row for row in rows if row["finite"]]
finite.sort(key=lambda row: (-row["accuracy"], row["total_macs"], row["lr"]))
selected = finite[0] if finite else None
checks = {
"both_records_finite": all(row["finite"] for row in rows),
"candidate_selected": selected is not None,
}
if selected is None:
checks.update({
"accuracy_at_least_0.65": False,
"within_10_points_of_bp": False,
"early_alignment_at_least_0.30": False,
"zero_task_loss_queries": False,
"macs_at_most_1.15x_bp": False,
})
else:
checks.update({
"accuracy_at_least_0.65": selected["accuracy"] >= 0.65,
"within_10_points_of_bp": (
selected["accuracy"] >= BP_ACCURACY - 0.10),
"early_alignment_at_least_0.30": (
selected["early_third_alignment"] >= 0.30),
"zero_task_loss_queries": all(
row["logical_batch_loss_queries"] == 0 for row in rows),
"macs_at_most_1.15x_bp": selected["total_macs"] <= 1.15 * bp_macs,
})
output = {
"protocol": "normalized_response_mirror_short_v1",
"status": "passed" if all(checks.values()) else "failed",
"checks": checks, "rows": rows, "selected_wm": selected,
"matched_bp": {"accuracy": BP_ACCURACY, "total_macs": bp_macs},
"confirmation_test_seeds_touched": False,
"review_score_before": 5, "review_score_after": 5,
"score_change_rule": "inherited short baseline 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,
"rows": rows, "selected_wm": selected,
}, indent=2))
if __name__ == "__main__":
main()
|
