"""Materialize the chronology-safe 128-shell AP1-M1 junction seed.""" from __future__ import annotations from dataclasses import asdict from datetime import datetime, timezone from pathlib import Path import json import numpy as np from ap1_m1_background_preflight import decode_hard_pass_state from ap1_m1_boundary_match import audit_authorities from ap1_m1_junction_convergence import complete_frozen_trajectory from ap1_m1_junction_seed import array_sha256, canonical_mode_state, file_sha256 from planck2018_neutrino_closure import ( Planck2018Pilot, derived_reference, shared_standard_background, ) EXPECTED_CLASSIFICATION = ( "M1_EXACT_INITIAL_JUNCTION_128_SHELL_PASS_SEED_CONSTRUCTION_OPEN" ) EXISTING_WRONSKIAN_LIMIT = 1.0e-11 class Junction128SeedError(RuntimeError): """The chronology-safe 128-shell seed chain is incomplete.""" def validate_authorities( apeiron_root: Path, heldout: dict, heldout_path: Path, freeze_path: Path, revalidation_code_path: Path, seed_code_path: Path, ) -> dict[str, str]: if heldout.get("classification") != EXPECTED_CLASSIFICATION: raise Junction128SeedError("128-shell held-out classification is not PASS") if not heldout.get("all_128_heldout_gates_pass"): raise Junction128SeedError("not every 128-shell held-out gate passed") if not heldout.get("checkpoint_eligible") or not heldout.get( "seed_construction_open" ): raise Junction128SeedError("128-shell seed construction is not open") if heldout.get("seed_released"): raise Junction128SeedError("128-shell report already claims a seed") freeze = json.loads(freeze_path.read_text(encoding="utf-8")) if freeze.get("classification") != ( "M1_EXACT_INITIAL_JUNCTION_128_TOLERANCES_FROZEN_HELDOUT_UNSEEN" ): raise Junction128SeedError("128-shell tolerance freeze is not canonical") if not freeze.get("all_128_freeze_gates_pass") or not freeze.get( "tolerances_frozen_before_128_shell_heldout" ): raise Junction128SeedError("128-shell tolerance freeze gates are incomplete") pilot_path = ( apeiron_root / "AP1/APEIRON_AP1_M1_JUNCTION_CONVERGENCE_PILOT_LATEST.json" ) convergence_code_path = apeiron_root / "AP1/CODE/ap1_m1_junction_convergence.py" heldout_code_path = apeiron_root / "AP1/CODE/ap1_m1_junction_heldout.py" helper_seed_code_path = apeiron_root / "AP1/CODE/ap1_m1_junction_seed.py" strict_96_tolerance_path = ( apeiron_root / "AP1/APEIRON_AP1_M1_EXACT_INITIAL_JUNCTION_TOLERANCES_LATEST.json" ) strict_96_result_path = ( apeiron_root / "AP1/APEIRON_AP1_M1_EXACT_INITIAL_JUNCTION_HELDOUT_LATEST.json" ) frozen_state_path = apeiron_root / "HANDOFF/CURRENT_STATE_LATEST.npz" actual_inputs = { freeze_path.name: freeze_path, "AP1/CODE/ap1_m1_junction_128_revalidation.py": revalidation_code_path, "AP1/CODE/ap1_m1_junction_convergence.py": convergence_code_path, "AP1/CODE/ap1_m1_junction_heldout.py": heldout_code_path, strict_96_tolerance_path.name: strict_96_tolerance_path, strict_96_result_path.name: strict_96_result_path, pilot_path.name: pilot_path, "HANDOFF/CURRENT_STATE_LATEST.npz": frozen_state_path, } for chain_name, expected_inputs in ( ("128 tolerance freeze", freeze["input_sha256"]), ("128 held-out", heldout["input_sha256"]), ): for name, expected_digest in expected_inputs.items(): path = actual_inputs.get(name) if path is None: raise Junction128SeedError( f"unresolved {chain_name} authority: {name}" ) if file_sha256(path) != expected_digest: raise Junction128SeedError( f"{chain_name} authority mismatch: {name}" ) pilot = json.loads(pilot_path.read_text(encoding="utf-8")) base_authorities = audit_authorities(apeiron_root) for name, digest in base_authorities.items(): if pilot.get("authority_sha256", {}).get(name) != digest: raise Junction128SeedError(f"frozen base authority mismatch: {name}") heldout_key = heldout_path.relative_to(apeiron_root).as_posix() observed = { **base_authorities, heldout_key: file_sha256(heldout_path), freeze_path.name: file_sha256(freeze_path), pilot_path.name: file_sha256(pilot_path), "AP1/CODE/ap1_m1_junction_convergence.py": file_sha256( convergence_code_path ), "AP1/CODE/ap1_m1_junction_heldout.py": file_sha256(heldout_code_path), "AP1/CODE/ap1_m1_junction_seed.py": file_sha256(helper_seed_code_path), "AP1/CODE/ap1_m1_junction_128_revalidation.py": file_sha256( revalidation_code_path ), "AP1/CODE/ap1_m1_junction_128_seed.py": file_sha256(seed_code_path), strict_96_tolerance_path.name: file_sha256(strict_96_tolerance_path), strict_96_result_path.name: file_sha256(strict_96_result_path), } return observed def build_seed( apeiron_root: Path, heldout: dict, authority_sha256: dict[str, str], ) -> tuple[dict[str, np.ndarray], dict]: frozen = decode_hard_pass_state( apeiron_root / "HANDOFF/CURRENT_STATE_LATEST.npz" ) local_state = np.asarray(frozen[0], dtype=float) level = heldout["heldout_128_level"] probe = level["exact_initial_HARD_PASS_junction_probe"] mode_nodes = int(level["mode_nodes"]) if mode_nodes != 128: raise Junction128SeedError("exactly 128 shells are required") mode = canonical_mode_state(local_state, mode_nodes) physical_state = local_state.copy() physical_state[5] = float(probe["N_physical"]) trajectory = complete_frozen_trajectory(frozen) standard_parameters = Planck2018Pilot() standard = shared_standard_background( np.array([physical_state[5]]), standard_parameters ) standard_vector = np.array( [ standard["rho_b"][0], standard["rho_c"][0], standard["rho_gamma"][0], standard["rho_nu"][0], standard["p_nu"][0], ], dtype=float, ) arrays = { "frozen_state_local": local_state, "m1_state_physical_N": physical_state, "N_offset_physical_minus_local": np.array( [probe["N_offset_physical_minus_local"]], dtype=float ), "Hdot_Mpl2": np.array([trajectory.Hdot[0]], dtype=float), "standard_rho_b_rho_c_rho_gamma_rho_nu_p_nu_Mpl4": standard_vector, "quantum_rho_pressure_chi2": np.array( [ probe["quantum_rho_Mpl4"], probe["quantum_pressure_Mpl4"], probe["quantum_chi2_Mpl2"], ], dtype=float, ), **mode, } wronskian_error = float( np.max(np.abs(mode["wronskian_observed"] - mode["wronskian_target"])) ) gates = { "all_128_heldout_gates_pass": bool(heldout["all_128_heldout_gates_pass"]), "chronology_safe_128_heldout": bool( heldout["heldout_was_unseen_when_128_tolerances_frozen"] ), "prior_96_not_used_as_seed": heldout["prior_96_role"] == "pilot_only_seed_eligibility_revoked_by_chronology_audit", "exact_frozen_index_zero": bool(np.array_equal(local_state, frozen[0])), "all_seed_arrays_finite": all( bool(np.all(np.isfinite(value))) for value in arrays.values() ), "all_128x4_frequencies_positive": bool( np.all(mode["omega2_Mpl2"] > 0.0) ), "canonical_initial_Wronskian_below_existing_limit": wronskian_error < EXISTING_WRONSKIAN_LIMIT, "standard_density_sum_matches_heldout": bool( np.isclose( np.sum(standard_vector[:4]), float(probe["required_standard_density_Mpl4"]), rtol=2.0e-12, atol=0.0, ) ), "old_solver_or_physical_map_not_called": True, "mode_state_is_original_index_zero_initial_condition": True, } passed = bool(all(gates.values())) report = { "schema": "apeiron-ap1-m1-junction-128-seed-v1.0", "updated_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "classification": ( "M1_JUNCTION_128_SEED_PASS_BACKGROUND_PILOT_OPEN" if passed else "M1_JUNCTION_128_SEED_NOT_PASS_NOT_RELEASED" ), "authority_sha256": authority_sha256, "chronology": { "96_shell_role": "pilot_only_seed_eligibility_revoked", "128_tolerances_frozen_before_heldout": True, "128_heldout_pass": bool(heldout["all_128_heldout_gates_pass"]), }, "junction": { "frozen_stored_index": 0, "frozen_state_local": local_state.tolist(), "m1_state_physical_N": physical_state.tolist(), "N_offset_physical_minus_local": float( probe["N_offset_physical_minus_local"] ), "Hdot_Mpl2": float(trajectory.Hdot[0]), "heldout_128_probe": probe, }, "shared_standard_parameters": asdict(standard_parameters), "shared_standard_derived": derived_reference(standard_parameters), "standard_components_at_junction": { "rho_b_Mpl4": float(standard_vector[0]), "rho_c_Mpl4": float(standard_vector[1]), "rho_gamma_Mpl4": float(standard_vector[2]), "rho_nu_Mpl4": float(standard_vector[3]), "p_nu_Mpl4": float(standard_vector[4]), "rho_sum_Mpl4": float(np.sum(standard_vector[:4])), }, "mode_seed": { "momentum_nodes": mode_nodes, "PV_sectors": 4, "min_omega2_Mpl2": float(np.min(mode["omega2_Mpl2"])), "max_omega2_Mpl2": float(np.max(mode["omega2_Mpl2"])), "canonical_initial_Wronskian_max_abs_error": wronskian_error, "origin": "canonical initial state at exact frozen index zero; no later-time vacuum reinitialization", }, "array_sha256": { name: array_sha256(value) for name, value in arrays.items() }, "gates": gates, "all_seed_gates_pass": passed, "seed_binary_roundtrip_exact": False, "seed_released": False, "new_AP1_M1_background_started": False, "physical_response_kernel_started": False, "nonpass_stored_or_used_as_seed": False, "equations_changed": False, "physics_changed": False, "parameters_changed": False, "existing_gate_thresholds_changed": False, "next_required": ( "implement and run a bounded new N-parametrized AP1-M1 background pilot; freeze its convergence gates before candidate assessment" if passed else "stop fail-closed; do not write or use a seed" ), "claim_boundary": "chronology-safe exact M1 junction seed only; no late-time background, kernel, observable, fit or significance", } return arrays, report def write_seed( arrays: dict[str, np.ndarray], report: dict, seed_output: Path, report_output: Path, ) -> dict: if not report.get("all_seed_gates_pass"): raise Junction128SeedError("NONPASS seed must not be written") np.savez_compressed(seed_output, **arrays) with np.load(seed_output, allow_pickle=False) as stored: roundtrip = set(stored.files) == set(arrays) and all( np.array_equal(stored[name], value) for name, value in arrays.items() ) if not roundtrip: raise Junction128SeedError("128-shell seed binary round-trip failed") report = dict(report) report["seed_npz"] = { "filename": seed_output.name, "sha256": file_sha256(seed_output), "arrays": sorted(arrays), } report["seed_binary_roundtrip_exact"] = True report["seed_released"] = True report_output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") return report def main() -> None: import argparse parser = argparse.ArgumentParser() parser.add_argument("apeiron_root", type=Path) parser.add_argument("heldout", type=Path) parser.add_argument("freeze", type=Path) parser.add_argument("revalidation_code", type=Path) parser.add_argument("seed_code", type=Path) parser.add_argument("--seed-output", type=Path, required=True) parser.add_argument("--report-output", type=Path, required=True) args = parser.parse_args() def rooted(path: Path) -> Path: return path if path.is_absolute() else args.apeiron_root / path heldout_path = rooted(args.heldout) freeze_path = rooted(args.freeze) revalidation_code_path = rooted(args.revalidation_code) seed_code_path = rooted(args.seed_code) seed_output = rooted(args.seed_output) report_output = rooted(args.report_output) heldout = json.loads(heldout_path.read_text(encoding="utf-8")) authorities = validate_authorities( args.apeiron_root, heldout, heldout_path, freeze_path, revalidation_code_path, seed_code_path, ) arrays, report = build_seed(args.apeiron_root, heldout, authorities) write_seed(arrays, report, seed_output, report_output) if __name__ == "__main__": main()