summaryrefslogtreecommitdiff
path: root/experiments/analyze_oral_a_v5_calibration.py
blob: 2844743b42f2ccd95120489083ad4f181bd6876d (plain)
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
#!/usr/bin/env python3
"""Validate and gate the frozen no-KP layerwise causal-bootstrap screen."""
import argparse
import json
import math
import os


SPLIT_HASH = "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b"


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input", default="results/oral_a_v5_calibration/result.json")
    parser.add_argument(
        "--out", default="results/oral_a_v5_calibration_gate.json")
    args = parser.parse_args()
    with open(args.input) as handle:
        record = json.load(handle)
    if record.get("protocol") != "oral_a_v5_layerwise_causal_bootstrap_capture_v1":
        raise ValueError("unexpected V5 protocol")
    expected = {
        "depth": 20, "width": 16, "seed": 0, "loader_seed": 0,
        "batch_size": 128, "train_limit": 10000,
        "val_examples": 5000, "split_seed": 2027,
        "normalization": "batchnorm", "residual_scale": 1.0,
        "feedback_scale": 1.0, "sigma": 0.01, "eta_A": 0.1,
        "perturb_seed": 5000, "sweeps": 20, "alignment_probe": 64,
        "calibration_augmentation": False,
    }
    if record.get("settings") != expected:
        raise ValueError("V5 settings drift")
    if record["provenance"]["git_tracked_dirty"]:
        raise ValueError("V5 result came from a tracked-dirty tree")
    if record["split"]["validation_index_sha256"] != SPLIT_HASH:
        raise ValueError("V5 split drift")
    if record["test_examples_touched"] or record["validation_endpoints_observed"]:
        raise ValueError("V5 touched a held-out endpoint")
    work = record["work"]
    audit = record["method_audit"]
    fixed = record["fixed_hfa"]
    learned = record["learned_lcb"]
    finite_metrics = [
        fixed["early_third_alignment"], fixed["all_layer_alignment"],
        learned["early_third_alignment"], learned["all_layer_alignment"],
        learned["min_feedback_forward_norm_ratio"],
        learned["max_feedback_forward_norm_ratio"],
    ]
    checks = {
        "finite": bool(record["finite"])
                  and all(math.isfinite(value) for value in finite_metrics),
        "exactly_380_edge_events": work["edge_events"] == 380,
        "exactly_760_batch_loss_queries": (
            work["logical_batch_loss_queries"] == 760),
        "exactly_48640_per_example_observations": (
            work["per_example_causal_observations"] == 48640),
        "forward_state_bitwise_fixed": (
            audit["forward_state_max_absolute_difference"] == 0.0),
        "zero_forward_weight_reads_in_update": (
            audit["forward_weight_reads_in_feedback_update"] == 0),
        "zero_reverse_mode_learning_operations": (
            audit["reverse_mode_learning_operations"] == 0),
        "early_third_at_least_0.10": (
            learned["early_third_alignment"] >= 0.10),
        "all_layer_at_least_0.20": (
            learned["all_layer_alignment"] >= 0.20),
        "early_gain_over_fixed_hfa_at_least_0.08": (
            learned["early_third_alignment"]
            - fixed["early_third_alignment"] >= 0.08),
        "feedback_norm_ratios_in_0.1_to_3": (
            learned["min_feedback_forward_norm_ratio"] >= 0.1
            and learned["max_feedback_forward_norm_ratio"] <= 3.0),
    }
    output = {
        "protocol": "oral_a_v5_layerwise_causal_bootstrap_gate_v1",
        "status": "passed" if all(checks.values()) else "failed",
        "checks": checks,
        "fixed_hfa": fixed,
        "learned_lcb": learned,
        "work": work,
        "source_commit": record["provenance"]["git_commit"],
        "source_result": args.input,
        "conditional_short_task_gate_open": all(checks.values()),
        "confirmation_test_seeds_touched": False,
        "review_score_before": 5,
        "review_score_after": 5,
        "score_change_rule": "causal capture 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,
        "fixed_hfa": fixed, "learned_lcb": learned,
    }, indent=2))


if __name__ == "__main__":
    main()