protocol: implement frozen mirror short accuracy gate

1 files changed, 133 insertions, 0 deletions

diff --git a/experiments/analyze_mirror_short.py b/experiments/analyze_mirror_short.py
new file mode 100644
index 0000000..a79ab8e
--- /dev/null
+++ b/experiments/analyze_mirror_short.py
@@ -0,0 +1,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()
+