#!/usr/bin/env python3 """Audit and gate residual-response mirror causal capture.""" import argparse import glob import json import math import os SPLIT_HASH = "8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b" RATES = (0.03, 0.1, 0.3) def load(path): with open(path) as handle: record = json.load(handle) args = record["args"] mode = args.get("mode") if mode not in ("hfa", "rrm"): raise ValueError(f"{path}: unexpected method") 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, "alignment_probe": 64, "mirror_batch_size": 1, "mirror_noise_std": 1.0, "mirror_seed": 3000, } for key, value in expected.items(): if args.get(key) != value: raise ValueError(f"{path}: {key} drift") expected_steps = 0 if mode == "hfa" else 20 if args.get("mirror_warmup_steps") != expected_steps: raise ValueError(f"{path}: mirror warmup drift") if mode == "rrm" and float(args["mirror_eta"]) not in RATES: raise ValueError(f"{path}: unregistered mirror 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 "local_parent_child_response_residual") if record.get("calibration_metric_space") != expected_space: raise ValueError(f"metric-space drift: {path}") diagnostics = record["diagnostics"] 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), } if mode == "rrm": warmup = record.get("mirror_warmup", {}).get("mean") if warmup is None: raise ValueError(f"missing RRM aggregate: {path}") metrics.update({ "mean_mirror_update_rms": warmup["mirror_update_rms"], "mean_response_residual_rms": ( warmup["mirror_response_residual_rms"]), "mean_response_residual_fraction": ( warmup["mirror_response_residual_fraction"]), }) finite = (bool(record["final"]["finite"]) and all(math.isfinite(value) for value in metrics.values())) return { "path": path, "mode": mode, "mirror_eta": (None if mode == "hfa" else float(args["mirror_eta"])), "metrics": metrics, "finite": finite, "logical_batch_loss_queries": int( record["work"]["logical_batch_loss_queries"]), "mirror_events": int(record["counters"]["mirror_events"]), "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/residual_mirror_capture") parser.add_argument( "--out", default="results/residual_mirror_capture_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"] == "rrm"] if len(references) != 1 or len(candidates) != len(RATES): raise ValueError("incomplete RRM-1 method grid") if {row["mirror_eta"] for row in candidates} != set(RATES): raise ValueError("incomplete RRM-1 rate grid") if len({row["source_commit"] for row in rows}) != 1: raise ValueError("RRM-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["mirror_eta"])) selected = eligible[0] if eligible else None 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.65": False, "all_layer_at_least_0.70": False, "mean_feedback_forward_cosine_at_least_0.93": False, "feedback_norm_ratios_in_0.5_to_1.5": False, "zero_task_loss_queries": False, }) else: metrics = selected["metrics"] checks.update({ "early_third_at_least_0.65": ( metrics["early_third_alignment"] >= 0.65), "all_layer_at_least_0.70": ( metrics["all_layer_alignment"] >= 0.70), "mean_feedback_forward_cosine_at_least_0.93": ( metrics["mean_feedback_forward_cosine"] >= 0.93), "feedback_norm_ratios_in_0.5_to_1.5": ( metrics["min_feedback_forward_norm_ratio"] >= 0.5 and metrics["max_feedback_forward_norm_ratio"] <= 1.5), "zero_task_loss_queries": all( row["logical_batch_loss_queries"] == 0 for row in rows), }) output = { "protocol": "residual_response_mirror_capture_v1", "status": "passed" if all(checks.values()) else "failed", "checks": checks, "rows": rows, "matched_fixed_hfa": references[0], "selected_rrm": selected, "confirmation_test_seeds_touched": False, "review_score_before": 5, "review_score_after": 5, "score_change_rule": "inherited baseline capture 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": references[0], "selected_rrm": selected, }, indent=2)) if __name__ == "__main__": main()