"""Construct the exact hash-bound AP1-M1 junction seed after held-out PASS.""" from __future__ import annotations from dataclasses import asdict from datetime import datetime, timezone from hashlib import sha256 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 chi_background_closure import ChiParameters, PV_J, physical_shells, portal_terms from planck2018_neutrino_closure import ( Planck2018Pilot, derived_reference, shared_standard_background, ) EXPECTED_HELDOUT_CLASSIFICATION = ( "M1_EXACT_INITIAL_JUNCTION_96_SHELL_PASS_SEED_RELEASED_BACKGROUND_SOLVE_OPEN" ) EXISTING_WRONSKIAN_LIMIT = 1.0e-11 class JunctionSeedError(RuntimeError): """The held-out result cannot authorize a physical junction seed.""" def file_sha256(path: Path) -> str: digest = sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def array_sha256(value: np.ndarray) -> str: array = np.ascontiguousarray(value) digest = sha256() digest.update(array.dtype.str.encode("ascii")) digest.update(str(array.shape).encode("ascii")) digest.update(array.tobytes()) return digest.hexdigest() def canonical_mode_state( background_state_local: np.ndarray, mode_nodes: int, ) -> dict[str, np.ndarray]: state = np.asarray(background_state_local, dtype=float) if state.shape != (6,) or not np.all(np.isfinite(state)): raise ValueError("one finite six-component background state required") if mode_nodes < 4: raise ValueError("at least four momentum shells required") parameters = ChiParameters() k, weights = physical_shells(parameters.Lambda, nodes=mode_nodes) mass2 = float(portal_terms(state[0], state[2], parameters)[0]) muR2 = (parameters.muR_over_Lambda * parameters.Lambda) ** 2 omega2 = ( k[None, :] ** 2 * np.exp(-2.0 * state[5]) + mass2 + PV_J[:, None] * muR2 ) if np.any(omega2 <= 0.0) or not np.all(np.isfinite(omega2)): raise JunctionSeedError("canonical junction frequencies are not positive") omega = np.sqrt(omega2) u_real = 1.0 / np.sqrt(2.0 * omega) u_imag = np.zeros_like(u_real) v_real = np.zeros_like(u_real) v_imag = -omega * u_real target = np.ones_like(u_real) observed = -2.0 * (u_real * v_imag - u_imag * v_real) return { "k_comoving_Mpl": k, "shell_weights_Mpl3": weights, "PV_J": np.asarray(PV_J, dtype=np.int64), "omega2_Mpl2": omega2, "u_real": u_real, "u_imag": u_imag, "v_real": v_real, "v_imag": v_imag, "wronskian_target": target, "wronskian_observed": observed, } def validate_seed_authorities( apeiron_root: Path, heldout: dict, heldout_path: Path, manifest_path: Path, pilot_path: Path, convergence_code_path: Path, boundary_code_path: Path, heldout_code_path: Path, ) -> dict[str, str]: if heldout.get("classification") != EXPECTED_HELDOUT_CLASSIFICATION: raise JunctionSeedError("held-out junction validation is not PASS") if not heldout.get("all_exact_initial_junction_gates_pass"): raise JunctionSeedError("not every held-out gate passed") if not heldout.get("checkpoint_released") or not heldout.get("seed_released"): raise JunctionSeedError("held-out PASS did not release the junction seed") manifest = json.loads(manifest_path.read_text(encoding="utf-8")) pilot = json.loads(pilot_path.read_text(encoding="utf-8")) if not manifest.get("all_tolerance_freeze_gates_pass"): raise JunctionSeedError("strict tolerance manifest is not PASS") if not manifest.get("tolerances_frozen_before_96_shell_heldout"): raise JunctionSeedError("tolerances were not frozen before held-out use") observed = { "AP1/APEIRON_AP1_M1_EXACT_INITIAL_JUNCTION_HELDOUT_LATEST.json": file_sha256( heldout_path ), "AP1/APEIRON_AP1_M1_EXACT_INITIAL_JUNCTION_TOLERANCES_LATEST.json": file_sha256( manifest_path ), "AP1/APEIRON_AP1_M1_JUNCTION_CONVERGENCE_PILOT_LATEST.json": file_sha256( pilot_path ), "AP1/CODE/ap1_m1_junction_convergence.py": file_sha256( convergence_code_path ), "AP1/CODE/ap1_m1_boundary_match.py": file_sha256(boundary_code_path), "AP1/CODE/ap1_m1_junction_heldout.py": file_sha256(heldout_code_path), } heldout_authority = heldout["input_sha256"] heldout_checks = { pilot_path.name: observed[ "AP1/APEIRON_AP1_M1_JUNCTION_CONVERGENCE_PILOT_LATEST.json" ], manifest_path.name: observed[ "AP1/APEIRON_AP1_M1_EXACT_INITIAL_JUNCTION_TOLERANCES_LATEST.json" ], "AP1/CODE/ap1_m1_junction_heldout.py": observed[ "AP1/CODE/ap1_m1_junction_heldout.py" ], } for name, digest in heldout_checks.items(): if heldout_authority.get(name) != digest: raise JunctionSeedError(f"held-out authority mismatch: {name}") if manifest["input_sha256"].get(pilot_path.name) != observed[ "AP1/APEIRON_AP1_M1_JUNCTION_CONVERGENCE_PILOT_LATEST.json" ]: raise JunctionSeedError("pilot changed after the strict tolerance freeze") base = audit_authorities(apeiron_root) for name, digest in base.items(): if pilot.get("authority_sha256", {}).get(name) != digest: raise JunctionSeedError(f"frozen authority drift: {name}") if heldout_authority.get("HANDOFF/CURRENT_STATE_LATEST.npz") != base[ "HANDOFF/CURRENT_STATE_LATEST.npz" ]: raise JunctionSeedError("held-out frozen-state hash drift") return {**base, **observed} def build_seed_payload( 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") state_local = np.asarray(frozen[0], dtype=float) probe = heldout["heldout_level"]["exact_initial_HARD_PASS_junction_probe"] mode_nodes = int(heldout["heldout_level"]["mode_nodes"]) mode_state = canonical_mode_state(state_local, mode_nodes) N_physical = float(probe["N_physical"]) state_physical = state_local.copy() state_physical[5] = N_physical trajectory = complete_frozen_trajectory(frozen) Hdot = float(trajectory.Hdot[0]) standard_parameters = Planck2018Pilot() standard = shared_standard_background( np.array([N_physical]), 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, ) quantum_vector = np.array( [ probe["quantum_rho_Mpl4"], probe["quantum_pressure_Mpl4"], probe["quantum_chi2_Mpl2"], ], dtype=float, ) arrays = { "frozen_state_local": state_local, "m1_state_physical_N": state_physical, "N_offset_physical_minus_local": np.array( [probe["N_offset_physical_minus_local"]], dtype=float ), "Hdot_Mpl2": np.array([Hdot], dtype=float), "standard_rho_b_rho_c_rho_gamma_rho_nu_p_nu_Mpl4": standard_vector, "quantum_rho_pressure_chi2": quantum_vector, **mode_state, } wronskian_error = float( np.max( np.abs( mode_state["wronskian_observed"] - mode_state["wronskian_target"] ) ) ) gates = { "heldout_96_all_gates_pass": bool( heldout["all_exact_initial_junction_gates_pass"] ), "exact_frozen_stored_index_zero": bool( np.array_equal(state_local, frozen[0]) ), "physical_N_offset_matches_heldout": bool( state_physical[5] == N_physical ), "all_seed_arrays_finite": all(np.all(np.isfinite(value)) for value in arrays.values()), "all_96x4_frequencies_positive": bool(np.all(mode_state["omega2_Mpl2"] > 0.0)), "canonical_initial_Wronskian_below_existing_limit": wronskian_error < EXISTING_WRONSKIAN_LIMIT, "required_standard_density_positive": float( np.sum(standard_vector[:4]) ) > 0.0, "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, ) ), "frozen_authorities_unchanged": True, "old_solver_or_physical_map_not_called": True, "mode_state_is_original_junction_initial_condition_not_endpoint_reinitialization": True, } passed = bool(all(gates.values())) report = { "schema": "apeiron-ap1-m1-junction-seed-v1.0", "updated_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "classification": ( "M1_JUNCTION_SEED_PASS_BACKGROUND_PILOT_OPEN" if passed else "M1_JUNCTION_SEED_NONPASS_NOT_RELEASED" ), "authority_sha256": authority_sha256, "junction": { "frozen_stored_index": 0, "frozen_state_local": state_local.tolist(), "m1_state_physical_N": state_physical.tolist(), "N_offset_physical_minus_local": float( probe["N_offset_physical_minus_local"] ), "Hdot_Mpl2": Hdot, "heldout_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": int(len(PV_J)), "min_omega2_Mpl2": float(np.min(mode_state["omega2_Mpl2"])), "max_omega2_Mpl2": float(np.max(mode_state["omega2_Mpl2"])), "canonical_initial_Wronskian_max_abs_error": wronskian_error, "origin": "canonical initial state at the exact first node of the frozen inherited trajectory; no endpoint 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 the new N-parametrized AP1-M1 background pilot; derive and freeze its convergence tolerances before candidate assessment" if passed else "stop fail-closed; do not write or use this seed" ), "claim_boundary": "accepted junction seed only; no late-time M1 background, response kernel, observable, fit or significance", } return arrays, report def write_seed_checkpoint( arrays: dict[str, np.ndarray], report: dict, seed_output: Path, report_output: Path, ) -> dict: if not report.get("all_seed_gates_pass"): raise JunctionSeedError("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 JunctionSeedError("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["classification"] = "M1_JUNCTION_SEED_PASS_BACKGROUND_PILOT_OPEN" 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("manifest", type=Path) parser.add_argument("pilot", type=Path) parser.add_argument("convergence_code", type=Path) parser.add_argument("boundary_code", type=Path) parser.add_argument("heldout_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() heldout = json.loads(args.heldout.read_text(encoding="utf-8")) authorities = validate_seed_authorities( args.apeiron_root, heldout, args.heldout, args.manifest, args.pilot, args.convergence_code, args.boundary_code, args.heldout_code, ) arrays, report = build_seed_payload(args.apeiron_root, heldout, authorities) write_seed_checkpoint(arrays, report, args.seed_output, args.report_output) if __name__ == "__main__": main()