#!/usr/bin/env python3 """Audit and gate hierarchical feedback-parameter causal capture.""" import argparse import glob import json import math import os SPLIT_HASH = "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b" RATES = (0.1, 1.0, 10.0) def load(path): with open(path) as handle: record = json.load(handle) args = record["args"] mode = args.get("mode") if mode not in ("hfa", "lhfa"): raise ValueError(f"{path}: unexpected mode") expected = { "depth": 20, "width": 16, "seed": 0, "loader_seed": 0, "epochs": 0, "train_limit": 10000, "val_examples": 5000, "split_seed": 2027, "eval_split": "validation", "eval_every": 0, "normalization": "batchnorm", "a_scale": 1.0, "pert_directions": 1, "pert_every": 4, "pert_sigma": 0.01, "perturb_seed": 1000, "alignment_probe": 64, } for key, value in expected.items(): if args.get(key) != value: raise ValueError(f"{path}: {key} drift") expected_warmup = 0 if mode == "hfa" else 400 if args.get("a_warmup_steps") != expected_warmup: raise ValueError(f"{path}: feedback warmup drift") if mode == "lhfa" and float(args["eta_A"]) not in RATES: raise ValueError(f"{path}: unregistered feedback rate") if record["provenance"]["git_tracked_dirty"]: raise ValueError(f"tracked-dirty result: {path}") if record["split"]["validation_index_sha256"] != SPLIT_HASH: raise ValueError(f"split drift: {path}") if record["evaluation_protocol"]["test_evaluations"]: raise ValueError(f"test touched: {path}") expected_space = (None if mode == "hfa" else "hierarchical_feedback_parameters") if record.get("calibration_metric_space") != expected_space: raise ValueError(f"metric-space drift: {path}") diagnostics = record.get("diagnostics") if diagnostics is None: raise ValueError(f"missing diagnostics: {path}") values = diagnostics["teaching_negative_gradient_cosine"] early_count = max(1, len(values) // 3) norm_ratios = diagnostics["feedback_forward_norm_ratio"] feedback_cosines = diagnostics["feedback_forward_cosine"] metrics = { "early_third_alignment": sum(values[:early_count]) / early_count, "all_layer_alignment": sum(values) / len(values), "mean_feedback_forward_cosine": ( sum(feedback_cosines) / len(feedback_cosines)), "min_feedback_forward_norm_ratio": min(norm_ratios), "max_feedback_forward_norm_ratio": max(norm_ratios), } warmup = record.get("apical_warmup", {}).get("mean") if mode == "lhfa": if warmup is None: raise ValueError(f"missing calibration aggregate: {path}") metrics.update({ "mean_calibration_mse": warmup["calibration_mse"], "mean_target_power": warmup["target_power"], "mean_prediction_target_cosine": ( warmup["prediction_target_cosine"]), "mean_parameter_update_rms": warmup["parameter_update_rms"], }) finite = (bool(record["final"]["finite"]) and all(math.isfinite(value) for value in metrics.values())) return { "path": path, "mode": mode, "eta_A": (None if mode == "hfa" else float(args["eta_A"])), "metrics": metrics, "finite": finite, "logical_batch_loss_queries": int( record["work"]["logical_batch_loss_queries"]), "total_macs": int(record["work"]["total_macs_estimate"]), "source_commit": record["provenance"]["git_commit"], } def main(): parser = argparse.ArgumentParser() parser.add_argument("--input", default="results/oral_a_v4_calibration") parser.add_argument( "--out", default="results/oral_a_v4_calibration_gate.json") args = parser.parse_args() rows = [load(path) for path in sorted( glob.glob(os.path.join(args.input, "*.json")))] references = [row for row in rows if row["mode"] == "hfa"] candidates = [row for row in rows if row["mode"] == "lhfa"] if len(references) != 1 or len(candidates) != len(RATES): raise ValueError("incomplete V4-1 method grid") if {row["eta_A"] for row in candidates} != set(RATES): raise ValueError("incomplete V4-1 rate grid") if len({row["source_commit"] for row in rows}) != 1: raise ValueError("V4-1 source commits differ") eligible = [row for row in candidates if row["finite"]] eligible.sort(key=lambda row: ( -row["metrics"]["early_third_alignment"], -row["metrics"]["all_layer_alignment"], row["eta_A"])) selected = eligible[0] if eligible else None reference = references[0] checks = { "all_four_records_finite": all(row["finite"] for row in rows), "candidate_selected": selected is not None, } if selected is None: checks.update({ "early_third_at_least_0.05": False, "all_layer_at_least_0.10": False, "early_gain_over_fixed_hfa_at_least_0.04": False, "feedback_norm_ratios_in_0.1_to_3": False, }) else: metrics = selected["metrics"] checks.update({ "early_third_at_least_0.05": ( metrics["early_third_alignment"] >= 0.05), "all_layer_at_least_0.10": ( metrics["all_layer_alignment"] >= 0.10), "early_gain_over_fixed_hfa_at_least_0.04": ( metrics["early_third_alignment"] - reference["metrics"]["early_third_alignment"] >= 0.04), "feedback_norm_ratios_in_0.1_to_3": ( metrics["min_feedback_forward_norm_ratio"] >= 0.1 and metrics["max_feedback_forward_norm_ratio"] <= 3.0), }) output = { "protocol": "oral_a_v4_hierarchical_causal_capture_v1", "status": "passed" if all(checks.values()) else "failed", "checks": checks, "rows": rows, "matched_fixed_hfa": reference, "selected_v4": selected, "confirmation_test_seeds_touched": False, "review_score_before": 5, "review_score_after": 5, "score_change_rule": "causal capture alone cannot raise 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({ "status": output["status"], "checks": checks, "reference": reference, "selected_v4": selected, }, indent=2)) if __name__ == "__main__": main()