#!/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") 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 record["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()