#!/usr/bin/env python3 """Audit the conditionally frozen D3 full dynamic-projection endpoint.""" import argparse import json import math import os SPLIT_HASH = "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b" KP_FULL_ACCURACY = 0.9126 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 main(): parser = argparse.ArgumentParser() parser.add_argument( "--input", default="results/kp_dynamic_projection_full/dynamic.json") parser.add_argument( "--d2_gate", default="results/kp_dynamic_projection_short_gate.json") parser.add_argument( "--bp_selection", default="results/oral_a_bp_selection.json") parser.add_argument( "--out", default="results/kp_dynamic_projection_full_gate.json") args = parser.parse_args() with open(args.d2_gate) as handle: d2 = json.load(handle) if (d2.get("protocol") != "kp_dynamic_neutral_projection_short_v1" or d2.get("status") != "passed" or d2.get("full_validation_opened") is not True): raise ValueError("D3 requires the audited D2 pass") with open(args.bp_selection) as handle: bp_selection = json.load(handle) if bp_selection.get("status") != "passed_primary": raise ValueError("D3 requires the frozen full BP reference") with open(bp_selection["selected"]["path"]) as handle: bp = json.load(handle) bp_accuracy = float(bp["final"]["accuracy"]) bp_macs = int(bp["work"]["total_macs_estimate"]) if bp_accuracy != 0.9162: raise ValueError("D3 BP accuracy reference drift") with open(args.input) as handle: record = json.load(handle) expected = { "mode": "kp_traffic", "traffic_rule": "innovation", "predictor_mode": "closed_form", "neutral_projection": 1, "depth": 20, "width": 16, "seed": 0, "loader_seed": 0, "batch_size": 128, "epochs": 200, "train_limit": 0, "val_examples": 5000, "split_seed": 2027, "eval_split": "validation", "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, "traffic_seed": 4000, "traffic_ratio": 4.0, "traffic_calibration_examples": 64, "learn_P": 1, "eta_P": 0.1, "predictor_warmup_steps": 1, "predictor_every": 0, "alignment_probe": 32, } for key, value in expected.items(): if record["args"].get(key) != value: raise ValueError(f"D3 {key} drift") if record["provenance"]["git_tracked_dirty"]: raise ValueError("tracked-dirty D3 record") if record["split"]["validation_index_sha256"] != SPLIT_HASH: raise ValueError("D3 split drift") evaluation = record["evaluation_protocol"] if (evaluation["validation_evaluations"] != 1 or evaluation["test_evaluations"] != 0 or evaluation["test_used_for_selection"] is not False): raise ValueError("D3 evaluation boundary drift") if record.get("calibration_metric_space") != ( "reciprocal_local_activity_products_with_mixed_apical_traffic"): raise ValueError("D3 metric-space drift") warmup = record.get("predictor_warmup", {}) if (warmup.get("mode") != "closed_form" or warmup.get("steps") != 1 or warmup.get("examples") != 64 or warmup.get("instruction_present") is not False or warmup.get("task_loader_state_restored") is not True or warmup.get("reuses_traffic_calibration_forward") is not True): raise ValueError("D3 neutral slow-fit invariant failed") epochs = record["epochs"] if len(epochs) != 200 or any(row["epoch"] != index + 1 for index, row in enumerate(epochs)): raise ValueError("D3 epoch trajectory is incomplete") 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): raise ValueError("D3 audited trajectory is incomplete") final = record["final"] diagnostics = record["diagnostics"] accuracy = float(final["accuracy"]) early = float(diagnostics["early_third_mean"]) final_feedback = float(diagnostics["mean_feedback_forward_cosine"]) late_feedback = sum(float(value["mean_feedback_forward_cosine"]) for value in tracking[150:]) / 50 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) maximum_pre_ratio = max(float(value[ "maximum_pre_projection_traffic_rms_ratio"]) for value in projection) instruction_observations = sum(int(value["instruction_observations"]) for value in projection) initial_ratio_error = max(abs(float(value) - 4.0) for value in record["traffic_calibration"][ "realized_traffic_instruction_rms_ratio"]) work = record["work"] counters = record["counters"] mac_ratio = float(work["total_macs_estimate"]) / bp_macs all_finite = bool(final["finite"]) and all( math.isfinite(value) for value in numeric_leaves({ "final": final, "epochs": epochs, "diagnostics": diagnostics, "warmup": warmup, "traffic": record["traffic_calibration"], "work": work, })) checks = { "record_trajectory_and_diagnostics_finite": all_finite, "accuracy_at_least_0p89": accuracy >= 0.89, "within_1p5_points_of_bp": accuracy >= bp_accuracy - 0.015, "within_1p5_points_of_clean_kp": ( accuracy >= KP_FULL_ACCURACY - 0.015), "early_alignment_at_least_0p90": early >= 0.90, "final_feedback_cosine_at_least_0p98": final_feedback >= 0.98, "epoch151_to200_feedback_cosine_at_least_0p97": late_feedback >= 0.97, "used_instruction_rms_ratio_within_1e_minus_4": ( maximum_signal_ratio_error <= 1e-4), "post_projection_traffic_ratio_at_most_1e_minus_5": ( maximum_post_ratio <= 1e-5), "post_projection_soma_slope_at_most_1e_minus_5": ( maximum_post_slope <= 1e-5), "zero_instruction_observations_in_fast_fit": ( instruction_observations == 0), "one_frozen_64_example_slow_fit": ( counters["predictor_update_examples"] == 64 and counters["predictor_warmup_examples"] == 0 and record["args"]["predictor_every"] == 0), "projection_observes_each_ordinary_example": ( counters["neutral_projection_examples"] == counters["ordinary_examples"] == 9_000_000), "initial_traffic_ratio_error_at_most_1e_minus_5": ( initial_ratio_error <= 1e-5), "zero_task_loss_queries": work["logical_batch_loss_queries"] == 0, "macs_at_most_1p34x_bp": mac_ratio <= 1.34, "elementwise_and_neutral_cost_reported": ( work["elementwise_operations_estimate"] > 0 and work["neutral_projection_observations"] == 9_000_000), "peak_allocated_memory_at_most_2p5_gib": ( record["hardware"]["peak_memory_allocated_bytes"] <= int(2.5 * 1024 ** 3)), "one_validation_and_zero_test_evaluations": ( evaluation["validation_evaluations"] == 1 and evaluation["test_evaluations"] == 0), } passed = all(checks.values()) output = { "protocol": "kp_dynamic_neutral_projection_full_v1", "status": "passed" if passed else "failed", "checks": checks, "metrics": { "accuracy": accuracy, "loss": float(final["loss"]), "bp_accuracy": bp_accuracy, "clean_kp_accuracy": KP_FULL_ACCURACY, "early_third_alignment": early, "final_feedback_forward_cosine": final_feedback, "epoch151_to200_feedback_forward_cosine": late_feedback, "maximum_used_instruction_rms_ratio_error": ( maximum_signal_ratio_error), "maximum_pre_projection_traffic_rms_ratio": maximum_pre_ratio, "maximum_post_projection_traffic_rms_ratio": maximum_post_ratio, "maximum_post_projection_soma_slope": maximum_post_slope, "total_macs": int(work["total_macs_estimate"]), "bp_total_macs": bp_macs, "mac_ratio_to_bp": mac_ratio, "elementwise_operations_estimate": int( work["elementwise_operations_estimate"]), "peak_memory_allocated_bytes": int(record["hardware"][ "peak_memory_allocated_bytes"]), "wall_s": float(record["timing"]["total_timed_wall_s"]), "source_commit": record["provenance"]["git_commit"], }, "independent_confirmation_opened": passed, "confirmation_test_seeds_touched": False, "review_score_before": 5, "review_score_after": 6 if passed else 5, "score_change_rule": ( "a fully passed frozen near-BP standard-ResNet innovation endpoint " "resolves the primary accept objection; confirmation remains open"), } 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()