#!/usr/bin/env python3 """Audit and select the D4-gated temporal-difference oral-B R1 screen.""" import argparse import glob import hashlib import json import math import os ETAS = (0.03, 0.1) TASK_SEEDS = (0, 1, 2) ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) PROTOCOL_PATH = os.path.join(ROOT, "ORAL_B_RECOVERY.md") 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 finite_tree(value): if isinstance(value, dict): return all(finite_tree(item) for item in value.values()) if isinstance(value, (list, tuple)): return all(finite_tree(item) for item in value) if isinstance(value, (int, float)): return math.isfinite(value) return True def validate(row, path, d4_gate, d4_gate_sha256, protocol_sha256): require(row.get("schema_version") == 1, f"{path}: schema") protocol = row.get("protocol", {}) require(protocol.get("name") == "oral_b_td_development_v1", f"{path}: protocol") require(protocol.get("selection_split") == "development", f"{path}: split") require(protocol.get("confirmation_seeds_touched") is False, f"{path}: confirmation leakage") args = row.get("args", {}) require(args.get("task_seed") in TASK_SEEDS, f"{path}: task seed") require(args.get("model_seed") == 0, f"{path}: model seed") require(float(args.get("eta")) in ETAS, f"{path}: eta") require(protocol.get("evaluation_task_seed") == args["task_seed"] + 200_000, f"{path}: evaluation seed") require(protocol.get("d4_gate_source") == d4_gate, f"{path}: D4 gate source") require(protocol.get("d4_gate_sha256") == d4_gate_sha256, f"{path}: D4 gate digest") require(args.get("d4_gate") == d4_gate, f"{path}: D4 gate path") config = row.get("config", {}) expected = { "n_plus": 5, "n_minus": 5, "n_background": 30, "context_dim": 16, "steps_per_episode": 28, "episodes_per_day": 64, "days": 14, "target": 0.8, "inertia": 0.65, "process_noise": 0.12, "context_ar": 0.8, "coupling_scale": 1.0, "predictor_eta": 0.2, "vectorizer_eta": 0.03, "forward_eta": args["eta"], "perturb_sigma": 0.03, "perturb_every": 4, "kappa": 0.0, "feedback": "performance_velocity", } for key, value in expected.items(): require(config.get(key) == value, f"{path}: config {key}") source = row.get("provenance", {}) require(source.get("git_tracked_dirty") is False, f"{path}: dirty") require(source.get("runner_and_protocol_tracked") is True, f"{path}: untracked") require(source.get("d4_gate_tracked") is True, f"{path}: untracked D4 gate") require(source.get("d4_gate_sha256") == d4_gate_sha256, f"{path}: provenance D4 gate digest") require(source.get("protocol_sha256") == protocol_sha256, f"{path}: protocol digest") require(len(source.get("git_commit", "")) == 40, f"{path}: commit") require(row.get("finite") is True and finite_tree(row), f"{path}: finite") for name in ("intact", "fixed_vectorizer", "plasticity_lesion", "oracle_role"): warmup = row["warmup"][name] require(warmup["batches"] == 100 and warmup["examples"] == 6400, f"{path}: warmup count {name}") require(warmup["instruction_present"] is False, f"{path}: warmup instruction {name}") require(warmup["role_cursor_scalar_observations"] == 12800, f"{path}: role observations {name}") require(warmup["predictor_max_abs_error"] <= 1e-5, f"{path}: predictor warmup {name}") condition = row["conditions"][name] require(len(condition["daily_success"]) == 14, f"{path}: trajectory {name}") def seed_checks(row): intact = row["conditions"]["intact"] fixed = row["conditions"]["fixed_vectorizer"] lesion = row["conditions"]["plasticity_lesion"] oracle = row["conditions"]["oracle_role"] signatures = row["signatures"] return { "learning_gain_at_least_0p10": intact["learning_gain"] >= 0.10, "final_success_at_least_0p70": intact["final_success"] >= 0.70, "fixed_vectorizer_gap_at_least_0p20": ( intact["final_success"] - fixed["final_success"] >= 0.20), "within_0p10_of_oracle_role": ( oracle["final_success"] - intact["final_success"] <= 0.10), "sign_inversion_at_least_0p01": ( signatures["causal_role_sign_inversion_index"] >= 0.01), "velocity_advantage_at_least_0p05": ( signatures["velocity_minus_error_abs_cv_corr"] >= 0.05), "residual_soma_corr_at_most_0p10": ( signatures["mean_abs_residual_soma_corr"] <= 0.10), "raw_minus_residual_corr_at_least_0p20": ( signatures["raw_minus_residual_abs_soma_corr"] >= 0.20), "plasticity_lesion_removes_half_gain": ( lesion["learning_gain"] <= 0.5 * intact["learning_gain"]), "learned_role_cosine_at_least_0p80": ( intact["role_cosine_after_training"] >= 0.80), } def main(): parser = argparse.ArgumentParser() parser.add_argument("--results", default="results/bci_td_dev") parser.add_argument( "--d4_gate", default="results/kp_dynamic_projection_confirmation_gate.json") parser.add_argument("--out", default="results/bci_td_dev_gate.json") args = parser.parse_args() with open(args.d4_gate) as handle: d4 = json.load(handle) require(d4.get("protocol") == "kp_dynamic_neutral_projection_confirmation_v1", "D4 protocol") require(d4.get("status") == "passed", "D4 did not pass") require(d4.get("review_score_after") == 7, "D4 score rule") d4_gate_sha256 = sha256(args.d4_gate) protocol_sha256 = sha256(PROTOCOL_PATH) paths = sorted(glob.glob(os.path.join(args.results, "bci_td_v1_*.json"))) require(len(paths) == 6, f"expected six R1 records, found {len(paths)}") records = {} commits = set() for path in paths: with open(path) as handle: row = json.load(handle) validate(row, path, args.d4_gate, d4_gate_sha256, protocol_sha256) key = (float(row["args"]["eta"]), row["args"]["task_seed"]) require(key not in records, f"duplicate R1 cell {key}") records[key] = row commits.add(row["provenance"]["git_commit"]) require(set(records) == {(eta, seed) for eta in ETAS for seed in TASK_SEEDS}, "R1 grid drift") require(len(commits) == 1, "R1 source revision drift") candidates = [] for eta in ETAS: rows = [records[(eta, seed)] for seed in TASK_SEEDS] checks = [seed_checks(row) for row in rows] candidates.append({ "eta": eta, "eligible": all(all(values.values()) for values in checks), "seed_checks": checks, "task_seeds": list(TASK_SEEDS), "final_success": [row["conditions"]["intact"]["final_success"] for row in rows], "learning_gain": [row["conditions"]["intact"]["learning_gain"] for row in rows], "sign_inversion": [row["signatures"][ "causal_role_sign_inversion_index"] for row in rows], "velocity_advantage": [row["signatures"][ "velocity_minus_error_abs_cv_corr"] for row in rows], "role_cosine": [row["conditions"]["intact"][ "role_cosine_after_training"] for row in rows], "worst_final_success": min( row["conditions"]["intact"]["final_success"] for row in rows), }) eligible = [candidate for candidate in candidates if candidate["eligible"]] selected = None if eligible: eligible.sort(key=lambda value: (-value["worst_final_success"], value["eta"])) selected = { "eta": eligible[0]["eta"], "worst_final_success": eligible[0]["worst_final_success"], } output = { "protocol": "oral_b_td_development_v1", "status": "passed" if selected else "failed_no_eligible_candidate", "complete_grid": True, "confirmation_seeds_touched": False, "selected": selected, "candidates": candidates, "source_commit": next(iter(commits)), "protocol_sha256": protocol_sha256, "d4_gate_sha256": d4_gate_sha256, "source_sha256": {path: sha256(path) for path in paths}, "oral_b_confirmation_opened": selected is not None, "review_score_before": 7, "review_score_after": 7, "score_change_rule": ( "R1 is development evidence and cannot change the paper score"), } if os.path.exists(args.out): with open(args.out) as handle: existing = json.load(handle) require(existing == output, "existing R1 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()