experiments: freeze oral A development gates

1 files changed, 72 insertions, 0 deletions

diff --git a/experiments/analyze_oral_a_bp.py b/experiments/analyze_oral_a_bp.py
new file mode 100755
index 0000000..d309273
--- /dev/null
+++ b/experiments/analyze_oral_a_bp.py
@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Apply the frozen A1 exact-BP viability gate."""
+import argparse
+import json
+import math
+import os
+
+
+def read(path, variant):
+    with open(path) as handle:
+        record = json.load(handle)
+    args = record["args"]
+    expected = {
+        "mode": "bp", "depth": 20, "width": 16, "seed": 0,
+        "epochs": 200, "val_examples": 5000, "eval_split": "validation",
+        "normalization": "batchnorm",
+    }
+    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"] != (
+            "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b"):
+        raise ValueError(f"split drift: {path}")
+    accuracy = float(record["final"]["accuracy"])
+    loss = float(record["final"]["loss"])
+    return {
+        "variant": variant, "path": path,
+        "source_commit": record["provenance"]["git_commit"],
+        "accuracy": accuracy, "loss": loss,
+        "finite": record["final"]["finite"] and math.isfinite(accuracy + loss),
+        "test_evaluations": record["evaluation_protocol"]["test_evaluations"],
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--primary", default="results/oral_a_dev/bp_reference_primary.json")
+    parser.add_argument("--recovery", default="results/oral_a_dev/bp_reference_recovery.json")
+    parser.add_argument("--out", default="results/oral_a_bp_selection.json")
+    args = parser.parse_args()
+    primary = read(args.primary, "primary")
+    rows = [primary]
+    if primary["finite"] and primary["accuracy"] >= 0.90:
+        status = "passed_primary"
+        selected = primary
+    elif not os.path.exists(args.recovery):
+        status = "recovery_required"
+        selected = None
+    else:
+        recovery = read(args.recovery, "recovery")
+        rows.append(recovery)
+        selected = max(rows, key=lambda row: row["accuracy"] if row["finite"] else -1.0)
+        status = ("passed_recovery" if selected["finite"] and selected["accuracy"] >= 0.90
+                  else "failed_no_viable_reference")
+    if any(row["test_evaluations"] for row in rows):
+        raise ValueError("A1 must not evaluate test")
+    output = {
+        "protocol": "oral_a_A1_v1", "status": status,
+        "rows": rows, "selected": selected,
+        "confirmation_test_seeds_touched": False,
+    }
+    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(output, indent=2))
+
+
+if __name__ == "__main__":
+    main()