#!/usr/bin/env python3 """Audit the frozen D1 dynamic neutral-projection training prefix.""" import argparse import json import math import os import statistics SPLIT_HASH = "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b" 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/dynamic.json") parser.add_argument( "--out", default="results/kp_dynamic_projection_gate.json") args = parser.parse_args() with open(args.input) as handle: record = json.load(handle) expected = { "protocol": "kp_dynamic_neutral_projection_diagnosis_v1", "scope": "training_only_no_validation_or_test_evaluation", "rule": "innovation", "predictor_mode": "closed_form", "predictor_every": 0, "stability_margin": 0.0, "neutral_projection": True, "max_steps": 352, "validation_evaluations": 0, "test_evaluations": 0, } for key, value in expected.items(): if record.get(key) != value: raise ValueError(f"D1 {key} drift") if record["provenance"]["git_tracked_dirty"]: raise ValueError("tracked-dirty D1 record") if record["split"]["validation_index_sha256"] != SPLIT_HASH: raise ValueError("D1 split drift") trajectory = record["trajectory"] if len(trajectory) != 352: raise ValueError("D1 trajectory is incomplete") losses = [float(row["batch_loss"]) for row in trajectory] signal_ratios = [ float(row["teaching_rms"]) / max(float(row["instruction_rms"]), 1e-30) for row in trajectory] reports = [row["neutral_projection"] for row in trajectory] if any(report is None for report in reports): raise ValueError("D1 projection report is missing") state_finite = all( value["all_finite"] for row in trajectory for value in row["parameter_state"].values()) all_finite = all(math.isfinite(value) for value in numeric_leaves(record)) def state_max(group): return max(float(row["parameter_state"][group][ "max_abs_over_finite_tensors"]) for row in trajectory) forward_weight_max = state_max("forward_weight") feedback_weight_max = state_max("feedback_weight") forward_momentum_max = state_max("forward_momentum") feedback_momentum_max = state_max("feedback_momentum") fit = record["predictor_warmup"]["closed_form_fit"] ratio_errors = [abs(float(value) - 4.0) for value in record["traffic_calibration"][ "realized_traffic_instruction_rms_ratio"]] maximum_post_ratio = max(float(report[ "post_projection_traffic_rms_ratio"]) for report in reports) maximum_post_slope = max(float(report[ "max_absolute_post_projection_soma_slope"]) for report in reports) maximum_pre_ratio = max(float(report[ "pre_projection_traffic_rms_ratio"]) for report in reports) maximum_correction_slope = max(float(report[ "max_absolute_correction_slope"]) for report in reports) observation_counts = [int(report["observations"]) for report in reports] instruction_observations = [ int(report["instruction_observations"]) for report in reports] checks = { "record_trajectory_and_state_finite": ( all_finite and state_finite and record["first_any_nonfinite_step"] is None and record["first_training_failure_step"] is None), "maximum_batch_loss_at_most_10": max(losses) <= 10.0, "final_32_mean_loss_at_most_2p5": ( statistics.mean(losses[-32:]) <= 2.5), "used_instruction_rms_ratio_within_1e_minus_4": ( max(abs(value - 1.0) for value in signal_ratios) <= 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), "paired_local_neutral_observations_only": ( min(observation_counts) >= 72 and max(observation_counts) <= 128 and max(instruction_observations) == 0), "forward_and_feedback_weight_max_at_most_10": ( forward_weight_max <= 10.0 and feedback_weight_max <= 10.0), "forward_and_feedback_momentum_max_at_most_50": ( forward_momentum_max <= 50.0 and feedback_momentum_max <= 50.0), "zero_margin_64_observation_slow_fit": ( int(fit["observations"]) == 64 and float(fit["stability_margin"]) == 0.0), "slow_predictor_frozen_during_task": all( row["predictor_updated"] is False for row in trajectory), "traffic_ratio_calibrated": max(ratio_errors) <= 1e-5, "task_loader_state_restored": ( record["predictor_warmup"][ "task_loader_state_restored"] is True), "no_held_out_evaluations": ( record["validation_evaluations"] == 0 and record["test_evaluations"] == 0), } passed = all(checks.values()) output = { "protocol": "kp_dynamic_neutral_projection_training_prefix_v1", "status": "passed" if passed else "failed", "checks": checks, "metrics": { "maximum_batch_loss": max(losses), "final_32_mean_loss": statistics.mean(losses[-32:]), "maximum_used_instruction_rms_ratio_error": max( abs(value - 1.0) for value in signal_ratios), "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, "maximum_correction_slope": maximum_correction_slope, "minimum_projection_observations": min(observation_counts), "maximum_projection_observations": max(observation_counts), "maximum_forward_weight": forward_weight_max, "maximum_feedback_weight": feedback_weight_max, "maximum_forward_momentum": forward_momentum_max, "maximum_feedback_momentum": feedback_momentum_max, "maximum_initial_traffic_ratio_error": max(ratio_errors), }, "source_commit": record["provenance"]["git_commit"], "validation_evaluations": 0, "test_evaluations": 0, "review_score_before": 5, "review_score_after": 5, "score_change_rule": ( "training-only stability evidence cannot change the paper 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(output, indent=2)) if __name__ == "__main__": main()