path: root/experiments/analyze_kp_dynamic_projection_confirmation.py

#!/usr/bin/env python3
"""Audit the frozen D4 paired five-seed test confirmation."""
import argparse
import glob
import json
import math
import os
import statistics


CONDITIONS = ("clean_kp", "dynamic")
SEEDS = tuple(range(10, 15))
T_CRITICAL_ONE_SIDED_95_DF4 = 2.131846786


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 main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input_dir", default="results/kp_dynamic_projection_confirmation")
    parser.add_argument(
        "--full_gate", default="results/kp_dynamic_projection_full_gate.json")
    parser.add_argument(
        "--out", default="results/kp_dynamic_projection_confirmation_gate.json")
    args = parser.parse_args()
    with open(args.full_gate) as handle:
        full_gate = json.load(handle)
    if (full_gate.get("protocol") != "kp_dynamic_neutral_projection_full_v1"
            or full_gate.get("status") != "passed"
            or full_gate.get("independent_confirmation_opened") is not True):
        raise ValueError("D4 requires the audited D3 pass")

    common = {
        "depth": 20, "width": 16, "batch_size": 128, "epochs": 200,
        "train_limit": 0, "val_examples": 0, "split_seed": 2027,
        "eval_split": "test", "eval_every": 0, "augment_train": 1,
        "lr": 0.1, "output_lr": 0.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,
        "alignment_probe": 32,
    }
    dynamic_expected = {
        "traffic_rule": "innovation", "predictor_mode": "closed_form",
        "neutral_projection": 1, "traffic_ratio": 4.0,
        "traffic_calibration_examples": 64, "learn_P": 1,
        "eta_P": 0.1, "predictor_warmup_steps": 1,
        "predictor_every": 0,
    }
    records = {}
    source_commits = set()
    failures = []
    finite_values = []
    expected_names = {
        f"seed{seed}_{condition}.json"
        for seed in SEEDS for condition in CONDITIONS
    }
    observed_names = {
        os.path.basename(path)
        for path in glob.glob(os.path.join(args.input_dir, "*.json"))
    }
    if observed_names != expected_names:
        missing = sorted(expected_names - observed_names)
        extra = sorted(observed_names - expected_names)
        raise ValueError(
            f"D4 requires exactly the frozen ten records; "
            f"missing={missing}, extra={extra}")
    bp_macs_50k = int(round(
        float(full_gate["metrics"]["bp_total_macs"]) * 50_000 / 45_000))
    for seed in SEEDS:
        for condition in CONDITIONS:
            path = os.path.join(
                args.input_dir, f"seed{seed}_{condition}.json")
            with open(path) as handle:
                record = json.load(handle)
            records[(seed, condition)] = record
            expected = {
                **common, "seed": seed, "loader_seed": seed,
                "mode": "kp_traffic" if condition == "dynamic" else "kp",
            }
            if condition == "dynamic":
                expected.update(dynamic_expected)
                expected["traffic_seed"] = 5000 + seed
            for key, value in expected.items():
                if record["args"].get(key) != value:
                    raise ValueError(
                        f"D4 seed {seed} {condition} {key} drift")
            if record["provenance"]["git_tracked_dirty"]:
                raise ValueError(
                    f"tracked-dirty D4 seed {seed} {condition}")
            source_commits.add(record["provenance"]["git_commit"])
            split = record["split"]
            if not (split["train_examples"] == 50_000
                    and split["validation_examples"] == 0
                    and split["validation_index_sha256"] is None
                    and split["test_examples"] == 10_000):
                raise ValueError(f"D4 seed {seed} {condition} split drift")
            evaluation = record["evaluation_protocol"]
            if not (evaluation["validation_evaluations"] == 0
                    and evaluation["test_evaluations"] == 1
                    and evaluation["test_used_for_selection"] is False):
                raise ValueError(
                    f"D4 seed {seed} {condition} evaluation drift")
            hardware = record.get("hardware", {})
            if not (hardware.get("cuda_device_name") ==
                    "NVIDIA GeForce GTX 1080"
                    and hardware.get("cuda_visible_devices") in ("5", "7")):
                raise ValueError(
                    f"D4 seed {seed} {condition} unauthorized hardware")
            epochs = record["epochs"]
            if (len(epochs) != 200
                    or any(row["epoch"] != index + 1
                           for index, row in enumerate(epochs))):
                failures.append(f"seed{seed}_{condition}:trajectory")
                continue
            finite_values.extend(numeric_leaves({
                "final": record["final"], "epochs": epochs,
                "diagnostics": record["diagnostics"],
                "work": record["work"],
            }))
            if not record["final"]["finite"]:
                failures.append(f"seed{seed}_{condition}:nonfinite")
            if record["work"]["logical_batch_loss_queries"] != 0:
                failures.append(f"seed{seed}_{condition}:queries")
            if condition == "dynamic":
                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"seed{seed}:slow_fit")
                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]
                if any(value is None for value in projection + mixed + tracking):
                    failures.append(f"seed{seed}:mechanism_trajectory")
                    continue
                maximum_signal_ratio_error = max(abs(
                    float(values["teaching_rms"])
                    / max(float(values["instruction_rms"]), 1e-30) - 1.0)
                    for values in mixed)
                maximum_post_ratio = max(float(value[
                    "maximum_post_projection_traffic_rms_ratio"])
                    for value in projection)
                maximum_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:])
                finite_values.extend([
                    maximum_signal_ratio_error, maximum_post_ratio,
                    maximum_post_slope, early, final_feedback, late_feedback,
                ])
                counters = record["counters"]
                work = record["work"]
                if maximum_signal_ratio_error > 1e-4:
                    failures.append(f"seed{seed}:instruction_ratio")
                if maximum_post_ratio > 1e-5:
                    failures.append(f"seed{seed}:post_ratio")
                if maximum_post_slope > 1e-5:
                    failures.append(f"seed{seed}:post_slope")
                if instruction_observations != 0:
                    failures.append(f"seed{seed}: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"seed{seed}:observation_counts")
                if early < 0.85:
                    failures.append(f"seed{seed}:early_alignment")
                if final_feedback < 0.98 or late_feedback < 0.97:
                    failures.append(f"seed{seed}:kp_tracking")
                if work["total_macs_estimate"] > 1.34 * bp_macs_50k:
                    failures.append(f"seed{seed}:mac_cost")
                if not (work["elementwise_operations_estimate"] > 0
                        and work["neutral_projection_observations"]
                        == 10_000_000):
                    failures.append(f"seed{seed}:elementwise_cost")
                if hardware["peak_memory_allocated_bytes"] > int(
                        2.5 * 1024 ** 3):
                    failures.append(f"seed{seed}:peak_memory")
    if len(source_commits) != 1:
        raise ValueError("all D4 runs must share one source revision")

    accuracies = {
        condition: [float(records[(seed, condition)]["final"]["accuracy"])
                    for seed in SEEDS]
        for condition in CONDITIONS
    }
    deficits = [clean - dynamic for clean, dynamic in zip(
        accuracies["clean_kp"], accuracies["dynamic"])]
    early_alignments = [float(records[(seed, "dynamic")]["diagnostics"][
        "early_third_mean"]) for seed in SEEDS]
    all_finite = all(math.isfinite(value) for value in finite_values)
    checks = {
        "all_records_trajectories_and_metrics_finite": all_finite,
        "mean_dynamic_test_accuracy_at_least_0p89": (
            statistics.mean(accuracies["dynamic"]) >= 0.89),
        "every_dynamic_test_accuracy_at_least_0p87": (
            min(accuracies["dynamic"]) >= 0.87),
        "mean_paired_deficit_to_clean_kp_at_most_0p015": (
            statistics.mean(deficits) <= 0.015),
        "one_sided_95pct_upper_deficit_at_most_0p025": (
            upper_confidence_bound(deficits) <= 0.025),
        "at_least_four_seeds_within_2_points_of_clean_kp": (
            sum(value <= 0.02 for value in deficits) >= 4),
        "mean_early_alignment_at_least_0p90": (
            statistics.mean(early_alignments) >= 0.90),
        "all_mechanism_query_tracking_and_cost_invariants": not failures,
    }
    status = "passed" if all(checks.values()) else "failed"
    output = {
        "protocol": "kp_dynamic_neutral_projection_confirmation_v1",
        "status": status,
        "checks": checks,
        "metrics": {
            "accuracy_by_seed": accuracies,
            "mean_accuracy": {condition: statistics.mean(values)
                              for condition, values in accuracies.items()},
            "paired_clean_kp_deficit": deficits,
            "mean_paired_clean_kp_deficit": statistics.mean(deficits),
            "paired_deficit_one_sided_95pct_upper_bound": (
                upper_confidence_bound(deficits)),
            "dynamic_early_alignment_by_seed": early_alignments,
            "mean_dynamic_early_alignment": statistics.mean(early_alignments),
            "bp_50k_mac_reference": bp_macs_50k,
            "invariant_failures": failures,
            "source_commit": next(iter(source_commits)),
        },
        "test_evaluations": 10,
        "review_score_before": 6,
        "review_score_after": 7 if status == "passed" else 6,
        "score_change_rule": (
            "only a complete untouched paired five-seed test panel can "
            "establish the strict accept bar; depth scaling remains oral work"),
    }
    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)
        if existing != output:
            raise ValueError(
                "existing D4 gate differs from the 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()