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