path: root/experiments/analyze_oral_a_dynamic_scaling.py

#!/usr/bin/env python3
"""Audit the frozen standard-depth dynamic-innovation recovery panel."""
import argparse
import glob
import hashlib
import json
import math
import os
import statistics


DEPTHS = (20, 32, 56)
SEEDS = tuple(range(10, 15))
METHODS = ("bp", "dfa", "clean_kp", "dynamic")
T_CRITICAL_ONE_SIDED_95_DF4 = 2.131846786
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PROTOCOL_PATH = os.path.join(ROOT, "ORAL_A_RECOVERY.md")
RUNNER_PATH = os.path.join(ROOT, "experiments", "oral_a_dynamic_scaling.py")


def require(condition, message):
    if not condition:
        raise ValueError(message)


def sha256(path):
    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def numeric_leaves(value):
    if isinstance(value, bool) or value is None:
        return
    if isinstance(value, (int, float)):
        yield float(value)
    elif isinstance(value, dict):
        for child in value.values():
            yield from numeric_leaves(child)
    elif isinstance(value, (list, tuple)):
        for child in value:
            yield from numeric_leaves(child)


def upper_confidence_bound(values):
    return (statistics.mean(values)
            + T_CRITICAL_ONE_SIDED_95_DF4
            * statistics.stdev(values) / math.sqrt(len(values)))


def lower_confidence_bound(values):
    return (statistics.mean(values)
            - T_CRITICAL_ONE_SIDED_95_DF4
            * statistics.stdev(values) / math.sqrt(len(values)))


def expected_args(method, depth, seed):
    common = {
        "depth": depth, "width": 16, "seed": seed, "loader_seed": seed,
        "batch_size": 128, "epochs": 200, "train_limit": 0,
        "val_examples": 0, "split_seed": 2027, "eval_split": "test",
        "eval_every": 0, "augment_train": 1, "lr_schedule": "step",
        "lr_milestones": "100,150", "lr_gamma": 0.1,
        "warmup_epochs": 0, "momentum": 0.9, "weight_decay": 1e-4,
        "normalization": "batchnorm", "a_scale": 1.0,
    }
    if method == "bp":
        return {**common, "mode": "bp", "lr": 0.1, "output_lr": None,
                "alignment_probe": 0}
    if method == "dfa":
        return {**common, "mode": "dfa", "lr": 0.03, "output_lr": 0.1,
                "vectorizer_mode": "spatial_template", "alignment_probe": 32}
    if method == "clean_kp":
        return {**common, "mode": "kp", "lr": 0.1, "output_lr": 0.1,
                "alignment_probe": 32}
    return {
        **common, "mode": "kp_traffic", "lr": 0.1, "output_lr": 0.1,
        "alignment_probe": 32, "traffic_rule": "innovation",
        "predictor_mode": "closed_form", "neutral_projection": 1,
        "traffic_seed": 5000 + seed, "traffic_ratio": 4.0,
        "traffic_calibration_examples": 64, "learn_P": 1, "eta_P": 0.1,
        "predictor_warmup_steps": 1, "predictor_every": 0,
    }


def validate_record(record, path, method, depth, seed):
    for key, value in expected_args(method, depth, seed).items():
        require(record.get("args", {}).get(key) == value,
                f"{path}: argument drift {key}")
    require(record.get("provenance", {}).get("git_tracked_dirty") is False,
            f"{path}: dirty source")
    split = record.get("split", {})
    require(split.get("train_examples") == 50_000
            and split.get("validation_examples") == 0
            and split.get("validation_index_sha256") is None
            and split.get("test_examples") == 10_000,
            f"{path}: split drift")
    evaluation = record.get("evaluation_protocol", {})
    require(evaluation.get("validation_evaluations") == 0
            and evaluation.get("test_evaluations") == 1
            and evaluation.get("test_used_for_selection") is False,
            f"{path}: evaluation drift")
    hardware = record.get("hardware", {})
    require(hardware.get("cuda_device_name") == "NVIDIA GeForce GTX 1080"
            and hardware.get("cuda_visible_devices") in ("5", "7"),
            f"{path}: unauthorized hardware")
    epochs = record.get("epochs", [])
    require(len(epochs) == 200
            and all(row.get("epoch") == index + 1
                    for index, row in enumerate(epochs)),
            f"{path}: trajectory drift")
    values = list(numeric_leaves({
        "final": record.get("final"), "epochs": epochs,
        "diagnostics": record.get("diagnostics"),
        "work": record.get("work"), "hardware": record.get("hardware"),
    }))
    require(record.get("final", {}).get("finite") is True
            and all(math.isfinite(value) for value in values),
            f"{path}: nonfinite")
    if method == "bp":
        require(record.get("diagnostics") is None, f"{path}: BP diagnostic")
    else:
        require(record.get("diagnostics") is not None,
                f"{path}: missing local diagnostic")
    return {
        "accuracy": float(record["final"]["accuracy"]),
        "loss": float(record["final"]["loss"]),
        "early_alignment": (None if method == "bp" else float(
            record["diagnostics"]["early_third_mean"])),
        "total_macs": int(record["work"]["total_macs_estimate"]),
        "elementwise_operations": int(record["work"][
            "elementwise_operations_estimate"]),
        "logical_queries": int(record["work"]["logical_batch_loss_queries"]),
        "peak_memory": int(hardware[
            "peak_memory_allocated_bytes"]),
        "source_commit": record["provenance"]["git_commit"],
    }


def dynamic_invariants(record, depth, seed, failures):
    epochs = record["epochs"]
    projection = [row.get("neutral_projection") for row in epochs]
    mixed = [row.get("mixed_apical") for row in epochs]
    tracking = [row.get("feedback_tracking") for row in epochs]
    tag = f"dynamic_d{depth}_s{seed}"
    if any(value is None for value in projection + mixed + tracking):
        failures.append(f"{tag}:mechanism_trajectory")
        return {}
    signal_error = max(abs(
        float(values["teaching_rms"])
        / max(float(values["instruction_rms"]), 1e-30) - 1.0)
        for values in mixed)
    post_ratio = max(float(value[
        "maximum_post_projection_traffic_rms_ratio"]) for value in projection)
    post_slope = max(float(value[
        "maximum_absolute_post_projection_soma_slope"]) for value in projection)
    instruction_observations = sum(int(
        value["instruction_observations"]) for value in projection)
    early = float(record["diagnostics"]["early_third_mean"])
    final_feedback = float(record["diagnostics"][
        "mean_feedback_forward_cosine"])
    late_feedback = statistics.mean(float(value[
        "mean_feedback_forward_cosine"]) for value in tracking[150:])
    counters = record["counters"]
    work = record["work"]
    warmup = record.get("predictor_warmup", {})
    if not (warmup.get("mode") == "closed_form"
            and warmup.get("steps") == 1
            and warmup.get("examples") == 64
            and warmup.get("instruction_present") is False
            and warmup.get("task_loader_state_restored") is True
            and warmup.get("reuses_traffic_calibration_forward") is True):
        failures.append(f"{tag}:warmup")
    if signal_error > 1e-4:
        failures.append(f"{tag}:instruction_ratio")
    if post_ratio > 1e-5:
        failures.append(f"{tag}:post_ratio")
    if post_slope > 1e-5:
        failures.append(f"{tag}:post_slope")
    if instruction_observations != 0:
        failures.append(f"{tag}:instruction_leak")
    if not (counters["predictor_update_examples"] == 64
            and counters["predictor_warmup_examples"] == 0
            and counters["neutral_projection_examples"]
            == counters["ordinary_examples"] == 10_000_000):
        failures.append(f"{tag}:observation_counts")
    if final_feedback < 0.98 or late_feedback < 0.97:
        failures.append(f"{tag}:feedback_tracking")
    if work["logical_batch_loss_queries"] != 0:
        failures.append(f"{tag}:queries")
    if not (work["elementwise_operations_estimate"] > 0
            and work["neutral_projection_observations"] == 10_000_000):
        failures.append(f"{tag}:elementwise_work")
    return {
        "instruction_ratio_error": signal_error,
        "post_projection_traffic_ratio": post_ratio,
        "post_projection_soma_slope": post_slope,
        "early_alignment": early,
        "final_feedback_cosine": final_feedback,
        "last50_feedback_cosine": late_feedback,
    }


def oral_a_checks(accuracies, alignments, failures):
    """Apply the frozen score gate to already validated paired arrays."""
    bp_deficits = {depth: [bp - dynamic for bp, dynamic in zip(
        accuracies["bp"][depth], accuracies["dynamic"][depth])]
        for depth in DEPTHS}
    kp_deficits = {depth: [kp - dynamic for kp, dynamic in zip(
        accuracies["clean_kp"][depth], accuracies["dynamic"][depth])]
        for depth in DEPTHS}
    d56_dfa_advantage = [dynamic - dfa for dynamic, dfa in zip(
        accuracies["dynamic"][56], accuracies["dfa"][56])]
    dynamic_depth_gain = [deep - shallow for shallow, deep in zip(
        accuracies["dynamic"][20], accuracies["dynamic"][56])]
    bp_depth_gain = [deep - shallow for shallow, deep in zip(
        accuracies["bp"][20], accuracies["bp"][56])]
    checks = {
        "all_60_records_and_audited_values_finite": not failures,
        "bp_mean_accuracy_at_least_0p90_each_depth": all(
            statistics.mean(accuracies["bp"][depth]) >= 0.90
            for depth in DEPTHS),
        "every_dynamic_accuracy_at_least_0p87": min(
            value for depth in DEPTHS
            for value in accuracies["dynamic"][depth]) >= 0.87,
        "dynamic_mean_within_2pt_bp_each_depth": all(
            statistics.mean(bp_deficits[depth]) <= 0.02 for depth in DEPTHS),
        "dynamic_bp_deficit_upper_bound_at_most_3pt_each_depth": all(
            upper_confidence_bound(bp_deficits[depth]) <= 0.03
            for depth in DEPTHS),
        "dynamic_mean_within_1p5pt_kp_each_depth": all(
            statistics.mean(kp_deficits[depth]) <= 0.015 for depth in DEPTHS),
        "dynamic_kp_deficit_upper_bound_at_most_2p5pt_each_depth": all(
            upper_confidence_bound(kp_deficits[depth]) <= 0.025
            for depth in DEPTHS),
        "d56_dynamic_mean_advantage_over_dfa_at_least_2pt":
            statistics.mean(d56_dfa_advantage) >= 0.02,
        "d56_dynamic_dfa_advantage_lower_bound_at_least_1pt":
            lower_confidence_bound(d56_dfa_advantage) >= 0.01,
        "dynamic_mean_d20_to_d56_gain_at_least_0p5pt":
            statistics.mean(dynamic_depth_gain) >= 0.005,
        "dynamic_depth_gain_lower_bound_nonnegative":
            lower_confidence_bound(dynamic_depth_gain) >= 0.0,
        "at_least_four_dynamic_seeds_improve_with_depth":
            sum(value > 0 for value in dynamic_depth_gain) >= 4,
        "dynamic_depth_gain_within_1pt_of_bp_gain":
            statistics.mean(dynamic_depth_gain)
            >= statistics.mean(bp_depth_gain) - 0.01,
        "dynamic_mean_alignment_at_least_0p85_each_depth": all(
            statistics.mean(alignments[depth]) >= 0.85 for depth in DEPTHS),
        "every_d56_dynamic_alignment_at_least_0p80":
            min(alignments[56]) >= 0.80,
        "d56_mean_alignment_retains_90pct_of_d20":
            statistics.mean(alignments[56])
            >= 0.90 * statistics.mean(alignments[20]),
        "all_mechanism_query_cost_memory_invariants": not failures,
    }
    derived = {
        "bp_deficit": bp_deficits,
        "kp_deficit": kp_deficits,
        "d56_dfa_advantage": d56_dfa_advantage,
        "dynamic_d20_to_d56_gain": dynamic_depth_gain,
        "bp_d20_to_d56_gain": bp_depth_gain,
    }
    return checks, derived


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--d4_results", default="results/kp_dynamic_projection_confirmation")
    parser.add_argument(
        "--new_results", default="results/oral_a_dynamic_scaling")
    parser.add_argument(
        "--d4_gate",
        default="results/kp_dynamic_projection_confirmation_gate.json")
    parser.add_argument(
        "--r2_gate", default="results/bci_td_confirmation_gate.json")
    parser.add_argument(
        "--out", default="results/oral_a_dynamic_scaling_gate.json")
    args = parser.parse_args()
    with open(args.d4_gate) as handle:
        d4 = json.load(handle)
    with open(args.r2_gate) as handle:
        r2 = json.load(handle)
    d4_digest = sha256(args.d4_gate)
    require(d4.get("protocol") ==
            "kp_dynamic_neutral_projection_confirmation_v1"
            and d4.get("status") == "passed"
            and d4.get("review_score_after") == 7, "D4 gate")
    require(r2.get("protocol") == "oral_b_td_confirmation_v1"
            and r2.get("status") == "passed"
            and r2.get("oral_b_plasticity_innovation_established") is True
            and r2.get("review_score_after") == 8
            and r2.get("d4_gate_sha256") == d4_digest, "R2 gate")

    expected_d4 = {
        f"seed{seed}_{condition}.json"
        for seed in SEEDS for condition in ("clean_kp", "dynamic")
    }
    observed_d4 = {os.path.basename(path) for path in glob.glob(
        os.path.join(args.d4_results, "*.json"))}
    require(observed_d4 == expected_d4,
            f"D4 record drift: missing={sorted(expected_d4-observed_d4)}, "
            f"extra={sorted(observed_d4-expected_d4)}")
    expected_new = {
        f"{method}_d{depth}_s{seed}.json"
        for depth in DEPTHS for seed in SEEDS for method in METHODS
        if not (depth == 20 and method in ("clean_kp", "dynamic"))
    }
    observed_new = {os.path.basename(path) for path in glob.glob(
        os.path.join(args.new_results, "*.json"))}
    require(observed_new == expected_new,
            f"new record drift: missing={sorted(expected_new-observed_new)}, "
            f"extra={sorted(observed_new-expected_new)}")

    records = {}
    rows = {}
    source_sha256 = {}
    d4_commits = set()
    new_commits = set()
    for seed in SEEDS:
        for method, condition in (("clean_kp", "clean_kp"),
                                  ("dynamic", "dynamic")):
            path = os.path.join(args.d4_results, f"seed{seed}_{condition}.json")
            with open(path) as handle:
                record = json.load(handle)
            records[(method, 20, seed)] = record
            rows[(method, 20, seed)] = validate_record(
                record, path, method, 20, seed)
            d4_commits.add(rows[(method, 20, seed)]["source_commit"])
            source_sha256[path] = sha256(path)
    for depth in DEPTHS:
        for seed in SEEDS:
            for method in METHODS:
                if depth == 20 and method in ("clean_kp", "dynamic"):
                    continue
                path = os.path.join(
                    args.new_results, f"{method}_d{depth}_s{seed}.json")
                with open(path) as handle:
                    record = json.load(handle)
                records[(method, depth, seed)] = record
                rows[(method, depth, seed)] = validate_record(
                    record, path, method, depth, seed)
                new_commits.add(rows[(method, depth, seed)]["source_commit"])
                source_sha256[path] = sha256(path)
    require(len(rows) == 60, "oral-A grid must contain exactly 60 records")
    require(len(d4_commits) == 1 and len(new_commits) == 1,
            "source revision drift")
    require(next(iter(d4_commits)) == d4["metrics"]["source_commit"],
            "D4 source does not match its gate")

    failures = []
    mechanism = {}
    for depth in DEPTHS:
        for seed in SEEDS:
            dynamic = records[("dynamic", depth, seed)]
            mechanism[f"d{depth}_s{seed}"] = dynamic_invariants(
                dynamic, depth, seed, failures)
            bp_macs = rows[("bp", depth, seed)]["total_macs"]
            for method in ("clean_kp", "dynamic"):
                if rows[(method, depth, seed)]["total_macs"] > 1.34 * bp_macs:
                    failures.append(f"{method}_d{depth}_s{seed}:mac_cost")
            if rows[("dynamic", depth, seed)]["peak_memory"] > 8 * 1024 ** 3:
                failures.append(f"dynamic_d{depth}_s{seed}:peak_memory")
            for method in ("dfa", "clean_kp", "dynamic"):
                if rows[(method, depth, seed)]["logical_queries"] != 0:
                    failures.append(f"{method}_d{depth}_s{seed}:queries")

    accuracies = {
        method: {depth: [rows[(method, depth, seed)]["accuracy"]
                         for seed in SEEDS]
                 for depth in DEPTHS}
        for method in METHODS
    }
    alignments = {
        depth: [rows[("dynamic", depth, seed)]["early_alignment"]
                for seed in SEEDS]
        for depth in DEPTHS
    }
    checks, derived = oral_a_checks(accuracies, alignments, failures)
    passed = all(checks.values())
    output = {
        "protocol": "oral_a_dynamic_innovation_scaling_v1",
        "status": "passed" if passed else "failed",
        "complete_grid": True,
        "checks": checks,
        "metrics": {
            "accuracy_by_method_depth_seed": accuracies,
            "mean_accuracy": {method: {
                str(depth): statistics.mean(values)
                for depth, values in by_depth.items()}
                for method, by_depth in accuracies.items()},
            **derived,
            "dynamic_alignment": alignments,
            "mechanism": mechanism,
            "invariant_failures": failures,
        },
        "d4_source_commit": next(iter(d4_commits)),
        "new_source_commit": next(iter(new_commits)),
        "source_sha256": source_sha256,
        "protocol_sha256": sha256(PROTOCOL_PATH),
        "runner_sha256": sha256(RUNNER_PATH),
        "d4_gate_sha256": d4_digest,
        "r2_gate_sha256": sha256(args.r2_gate),
        "test_evaluations": 60,
        "standard_depth_scaling_established": passed,
        "review_score_before": 8,
        "review_score_after": 9 if passed else 8,
        "score_change_rule": (
            "only the complete R2-gated 60-cell standard-ResNet panel can "
            "establish oral-A dynamic-innovation scaling"),
    }
    os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
    if os.path.exists(args.out):
        with open(args.out) as handle:
            existing = json.load(handle)
        require(existing == output,
                "existing oral-A gate differs from deterministic re-audit")
        print(json.dumps(output, indent=2))
        return
    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()