#!/usr/bin/env python3 """Make the registered R0 go/no-go decision from development artifacts.""" from __future__ import annotations import argparse import json import math from pathlib import Path from typing import Any def atomic_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") temporary.replace(path) def load_pass(path: Path, name: str) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if payload.get("status") != "PASS": raise RuntimeError(f"{name} is not a valid PASS artifact: {path}") return payload def reduction(reference: float, candidate: float) -> float: if reference <= 0.0 or candidate < 0.0 or not math.isfinite(candidate): raise ValueError("costs must be finite and non-negative with positive reference") return 1.0 - candidate / reference def execute(args: argparse.Namespace) -> dict[str, Any]: paired = load_pass(args.paired_state, "paired state") transfer = load_pass(args.transfer, "transfer diagnostic") e2e = load_pass(args.pilot_e2e, "P1 E2E replay") red_flags = [] if len(paired.get("examples", [])) != 12: red_flags.append("paired_state_not_12_anchors") if transfer.get("sanity", {}).get("transitions") != 120: red_flags.append("transfer_not_120_cross_config_transitions") if transfer.get("red_flags"): red_flags.append("transfer_has_red_flags") examples = paired["examples"] state_available = all( row["state_residual"]["coverage"]["missing"] == 0 and row["state_residual"]["coverage"]["available"] > 0 for row in examples ) state_vectors = { tuple(sorted(row["state_residual"]["values"].items())) for row in examples } state_varies = len(state_vectors) > 1 error_examples = [row for row in examples if row["simulator_error"]] simulator_errors = len(error_examples) error_state_discrepancy = any( any( abs(float(value)) > 1e-12 for value in row["state_residual"]["values"].values() ) for row in error_examples ) simulator = transfer["simulator"] prior_safe = [] direct_sensitivity = [] hybrid_incremental = [] for regularization, detail in transfer["regularization_sensitivity"].items(): for weight, models in detail["hybrid"]["prior_shrinkage"].items(): if float(weight) == 0.0: continue telemetry = models["raw_simulator_prior"][ "sim_plus_outcome_plus_telemetry" ] decision_safe = ( telemetry["simulator_errors_corrected"] >= 1 and telemetry["simulator_errors_corrected"] >= telemetry["simulator_correct_corrupted"] ) continuous_safe = ( telemetry["rmse"] <= simulator["rmse"] + 1e-12 and telemetry["mae"] <= simulator["mae"] + 1e-12 ) if decision_safe and continuous_safe: prior_safe.append( { "regularization": float(regularization), "prior_weight": float(weight), "simulator_errors_corrected": telemetry[ "simulator_errors_corrected" ], "simulator_correct_corrupted": telemetry[ "simulator_correct_corrupted" ], "rmse": telemetry["rmse"], "mae": telemetry["mae"], } ) direct = detail["direct"] direct_cmp = direct["comparison"] direct_sensitivity.append( { "regularization": float(regularization), "accuracy_delta": direct_cmp["delta_telemetry_minus_baseline"][ "feasibility_accuracy" ], "rmse_delta": direct_cmp["delta_telemetry_minus_baseline"]["rmse"], "mae_delta": direct_cmp["delta_telemetry_minus_baseline"]["mae"], "errors_corrected": direct_cmp["baseline_errors_corrected"], "correct_corrupted": direct_cmp["baseline_correct_corrupted"], "telemetry_accuracy": direct["telemetry_only"][ "feasibility_accuracy" ], } ) hybrid_cmp = detail["hybrid"]["comparison"] hybrid_incremental.append( { "regularization": float(regularization), "accuracy_delta": hybrid_cmp["delta_telemetry_minus_baseline"][ "feasibility_accuracy" ], "rmse_delta": hybrid_cmp["delta_telemetry_minus_baseline"]["rmse"], "mae_delta": hybrid_cmp["delta_telemetry_minus_baseline"]["mae"], "errors_corrected": hybrid_cmp["baseline_errors_corrected"], "correct_corrupted": hybrid_cmp["baseline_correct_corrupted"], } ) k1 = e2e["by_k"]["1"]["sim_top_k_plus_real_final"] k2 = e2e["by_k"]["2"]["sim_top_k_plus_real_final"] headroom = { "interpretation": ( "oracle correction stops after the simulator top-1 real final instead " "of evaluating the frozen safety top-2" ), "online": { "reference_k2_h20_hours": k2["online_h20_hours"], "oracle_k1_h20_hours": k1["online_h20_hours"], "absolute_h20_hours": k2["online_h20_hours"] - k1["online_h20_hours"], "fraction": reduction(k2["online_h20_hours"], k1["online_h20_hours"]), }, "with_prior_failure": { "reference_k2_h20_hours": k2["conservative_h20_hours_with_prior_failure"], "oracle_k1_h20_hours": k1["conservative_h20_hours_with_prior_failure"], "absolute_h20_hours": k2["conservative_h20_hours_with_prior_failure"] - k1["conservative_h20_hours_with_prior_failure"], "fraction": reduction( k2["conservative_h20_hours_with_prior_failure"], k1["conservative_h20_hours_with_prior_failure"], ), }, "versus_observed_safe_k1_fraction": 0.0, "k1_zero_regret": k1["real_regret"] == 0.0, "k2_zero_regret": k2["real_regret"] == 0.0, } condition_1 = state_available and state_varies condition_2 = simulator_errors >= 1 and error_state_discrepancy condition_3 = bool(prior_safe) condition_4 = headroom["online"]["fraction"] >= 0.15 conditions = { "state_available_and_varies": condition_1, "known_simulator_error_has_state_discrepancy": condition_2, "prior_preserving_safe_correction_exists": condition_3, "oracle_online_headroom_at_least_15pct": condition_4, } gate_pass = not red_flags and all(conditions.values()) direct_incremental = all( row["accuracy_delta"] >= -1e-12 and row["errors_corrected"] >= row["correct_corrupted"] for row in direct_sensitivity ) result = { "schema": "telemetry-residual-r0-gate-v1", "status": "STOP" if red_flags else "PASS", "scope": "P1 development premise/headroom audit; not headline evidence", "decision": "PROCEED_TO_R1" if gate_pass else "STOP_BEFORE_R1", "r0_gate_pass": gate_pass, "conditions": conditions, "route_findings": { "hybrid_prior_safe_candidates": prior_safe, "hybrid_incremental_regularization": hybrid_incremental, "direct_incremental_regularization": direct_sensitivity, "direct_incremental_decision_signal_all_lambdas": direct_incremental, "direct_best_absolute_accuracy": max( row["telemetry_accuracy"] for row in direct_sensitivity ), "raw_simulator_accuracy": simulator["feasibility_accuracy"], }, "headroom": headroom, "red_flags": red_flags, "sanity": { "anchors": { "n": len(examples), "min": min(row["real_pass_rate_rep1"] for row in examples), "max": max(row["real_pass_rate_rep1"] for row in examples), "distinct_n": len( {row["real_pass_rate_rep1"] for row in examples} ), }, "state_vectors": { "n": len(examples), "min": min(len(row["state_residual"]["values"]) for row in examples), "max": max(len(row["state_residual"]["values"]) for row in examples), "distinct_n": len(state_vectors), }, "costs_h20_hours": { "n": 4, "min": min( k1["online_h20_hours"], k2["online_h20_hours"], k1["conservative_h20_hours_with_prior_failure"], k2["conservative_h20_hours_with_prior_failure"], ), "max": max( k1["online_h20_hours"], k2["online_h20_hours"], k1["conservative_h20_hours_with_prior_failure"], k2["conservative_h20_hours_with_prior_failure"], ), "distinct_n": len( { k1["online_h20_hours"], k2["online_h20_hours"], k1["conservative_h20_hours_with_prior_failure"], k2["conservative_h20_hours_with_prior_failure"], } ), }, "invariants": { "no_data_red_flags": not red_flags, "state_nonempty_and_varied": condition_1, "pass_rates_bounded": all( 0.0 <= row["real_pass_rate_rep1"] <= 1.0 and 0.0 <= row["sim_pass_rate"] <= 1.0 for row in examples ), "costs_nonnegative": all( value >= 0.0 for value in ( k1["online_h20_hours"], k2["online_h20_hours"], k1["conservative_h20_hours_with_prior_failure"], k2["conservative_h20_hours_with_prior_failure"], ) ), "per_config_not_identical": len( {row["real_pass_rate_rep1"] for row in examples} ) > 1, }, }, } atomic_json(args.output, result) if result["status"] != "PASS": raise RuntimeError(red_flags) return result def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser() result.add_argument("--paired-state", type=Path, required=True) result.add_argument("--transfer", type=Path, required=True) result.add_argument("--pilot-e2e", type=Path, required=True) result.add_argument("--output", type=Path, required=True) return result def main() -> None: result = execute(parser().parse_args()) print( json.dumps( { "status": result["status"], "decision": result["decision"], "r0_gate_pass": result["r0_gate_pass"], "conditions": result["conditions"], "headroom": result["headroom"], "sanity": result["sanity"], "red_flags": result["red_flags"], }, sort_keys=True, ) ) if __name__ == "__main__": main()