"""Fail-closed preflight for the G31-bound balanced coupled M1 stage. The preflight changes no Apeiron/AME-1 or M-1 equation, physical parameter, renormalization term, Pauli--Villars basis, moving split, or existing gate threshold. It replaces only the numerically singular single Riccati chart by the already validated G31 projective two-chart transport and predeclares a stricter fixed-point and three-level time-convergence protocol. No background trajectory is evaluated or serialized by ``build_preflight``. The previously inspected interval through relative N=0.08 is eligible only as a method pilot. A future heldout is disjoint and begins beyond N=0.08. """ from __future__ import annotations import argparse from dataclasses import asdict, dataclass, replace from datetime import datetime, timezone from hashlib import sha256 import json import math from pathlib import Path from typing import Any import numpy as np import ap1_m1_balanced_chart_transport as g31 import ap1_m1_coupled_moving_split_background as g29 import ap1_m1_staged_log_background_extension as g30 from chi_background_closure import ChiParameters, ChiTrajectory, PV_C, portal_terms EXPECTED_AUTHORITIES = { "AP1/APEIRON_AP1_M1_COUPLED_MOVING_SPLIT_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json": ( "c2f90da969a139fbb690b26fc12e821a6dcf82e126dffa0b1794a3ff6fd64cd9" ), "AP1/APEIRON_AP1_M1_STAGED_LOG_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json": ( "6eeb0ef13ab072a1fb04bb47253d01498e2b7e28fd740c2de13fbdc8b96690df" ), "AP1/APEIRON_AP1_M1_BALANCED_CHART_TRANSPORT_REFERENCE_CHECKPOINT_LATEST.json": ( "1392637e86d7ae4a290b6250379de39343af52f3b4b4fdc9a8e9941972054fd0" ), "AP1/CODE/ap1_m1_coupled_moving_split_background.py": ( "3b6f2b38d047db3cb9510af9c2c92ef9c80c2c1ca16c40212f7e7c354a76f400" ), "AP1/CODE/ap1_m1_staged_log_background_extension.py": ( "25ce7b0a19582b1e7e57be7339914ddebd41a902eb4503803bc69d18ea9ce5ae" ), "AP1/CODE/ap1_m1_balanced_chart_transport.py": ( "2fc2831de63f049ec23558c90f349184ef509e93db20d32e315372edc48ce21e" ), } class BalancedCoupledPreflightError(RuntimeError): """An authority, structural, chronology, or lock gate failed.""" @dataclass(frozen=True) class BalancedCoupledStagePlan: baseline_span_N: float = 0.04 pilot_span_N: float = 0.08 pilot_evaluation_start_N: float = 0.065 pilot_nodes_coarse: int = 1025 pilot_nodes_middle: int = 2049 pilot_nodes_fine: int = 4097 prior_diagnostic_endpoint_N: float = 0.08 heldout_span_N: float = 0.10 heldout_evaluation_start_N: float = 0.085 heldout_nodes_coarse: int = 1281 heldout_nodes_middle: int = 2561 heldout_nodes_fine: int = 5121 qcut_over_Lambda: float = 0.6 adiabatic_support_stride: int = 1 adiabatic_momentum_support_nodes: int = 257 fixed_point_steps: int = 10 relaxation: float = 0.9 entry_nodes_per_panel: int = 2 tail_nodes_per_octave: int = 4 K_over_Lambda: float = 64.0 max_mode_step_N: float = 1.25e-5 background_rtol: float = 2.0e-10 decimal_low_digits: int = 60 decimal_high_digits: int = 80 max_pilot_resolved_modes: int = 8500 max_heldout_resolved_modes: int = 10500 def _level_nodes(self, phase: str) -> tuple[int, int, int]: if phase == "pilot": return ( self.pilot_nodes_coarse, self.pilot_nodes_middle, self.pilot_nodes_fine, ) if phase == "heldout": return ( self.heldout_nodes_coarse, self.heldout_nodes_middle, self.heldout_nodes_fine, ) raise ValueError("phase must be pilot or heldout") def validate(self) -> None: if self.baseline_span_N != 0.04: raise ValueError("the G30 endpoint must remain the coupled-stage baseline") if not ( self.baseline_span_N < self.pilot_evaluation_start_N < self.pilot_span_N == self.prior_diagnostic_endpoint_N < self.heldout_evaluation_start_N < self.heldout_span_N ): raise ValueError("pilot and future heldout geometry is not chronology-safe") if self.qcut_over_Lambda != g29.CoupledConfig().qcut_over_Lambda: raise ValueError("the frozen moving split changed") if self.adiabatic_support_stride != g29.CoupledConfig().adiabatic_support_stride: raise ValueError("the adiabatic support rule changed") if self.adiabatic_momentum_support_nodes != g29.CoupledConfig().adiabatic_momentum_support_nodes: raise ValueError("the adiabatic momentum support changed") if self.relaxation != g29.CoupledConfig().relaxation: raise ValueError("the frozen relaxation changed") if self.entry_nodes_per_panel != g29.CoupledConfig().fine_entry_nodes_per_panel: raise ValueError("the accepted entry quadrature changed") if self.tail_nodes_per_octave != g29.CoupledConfig().fine_tail_nodes_per_octave: raise ValueError("the accepted tail quadrature changed") if self.K_over_Lambda != g29.CoupledConfig().fine_K_over_Lambda: raise ValueError("the accepted UV support changed") if self.max_mode_step_N != g29.CoupledConfig().fine_mode_step_N: raise ValueError("the accepted fine mode step changed") if self.background_rtol != g29.CoupledConfig().fine_background_rtol: raise ValueError("the accepted fine background tolerance changed") if ( self.decimal_low_digits != g29.CoupledConfig().decimal_low_digits or self.decimal_high_digits != g29.CoupledConfig().decimal_high_digits ): raise ValueError("the accepted Decimal precision pair changed") if self.fixed_point_steps <= g29.CoupledConfig().fixed_point_steps: raise ValueError("the corrected stage must refine fixed-point closure") for phase, span in ( ("pilot", self.pilot_span_N), ("heldout", self.heldout_span_N), ): coarse, middle, fine = self._level_nodes(phase) if not ( fine - 1 == 2 * (middle - 1) and middle - 1 == 2 * (coarse - 1) ): raise ValueError("three exactly nested time levels are required") if any(nodes < 17 or nodes % 2 != 1 for nodes in (coarse, middle, fine)): raise ValueError("odd bounded time grids are required") fine_step = span / (fine - 1) inherited = ( g29.CoupledConfig().heldout_span_N / (g29.CoupledConfig().heldout_fine_nodes - 1) ) if not math.isclose(fine_step, inherited, rel_tol=0.0, abs_tol=1.0e-18): raise ValueError("the G29/G30 accepted fine N spacing changed") pilot_modes = 128 + self.entry_nodes_per_panel * (self.pilot_nodes_fine - 1) heldout_modes = 128 + self.entry_nodes_per_panel * (self.heldout_nodes_fine - 1) if pilot_modes > self.max_pilot_resolved_modes: raise ValueError("pilot mode budget exceeded") if heldout_modes > self.max_heldout_resolved_modes: raise ValueError("heldout mode budget exceeded") PREDECLARED_POLICY: dict[str, Any] = { "inherited_metric_freeze_policy": { name: dict(value) for name, value in g30.STAGE_FREEZE_POLICY["metrics"].items() }, "existing_G29_absolute_gates_retained": True, "existing_G31_chart_gates_retained": True, "three_level_total_coupled_time_convergence_required": True, "fine_to_middle_delta_must_not_exceed_middle_to_coarse_delta": True, "minimum_observed_Richardson_order": 0.5, "same_operator_settings_on_all_three_time_levels": True, "pilot_tolerances_must_be_frozen_before_future_heldout": True, "prior_N_le_0p08_result_is_pilot_information_only": True, "future_heldout_starts_strictly_after_N_0p08": True, "nonpass_may_not_be_written_checkpointed_or_seeded": True, } 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 _utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def verify_authorities(root: Path) -> tuple[dict[str, Any], dict[str, Any]]: observed = {name: file_sha256(root / name) for name in EXPECTED_AUTHORITIES} if observed != EXPECTED_AUTHORITIES: changed = [ name for name, digest in EXPECTED_AUTHORITIES.items() if observed.get(name) != digest ] raise BalancedCoupledPreflightError(f"authority drift: {changed}") g30_checkpoint = json.loads( (root / "AP1/APEIRON_AP1_M1_STAGED_LOG_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json") .read_text(encoding="utf-8") ) g31_checkpoint = json.loads( (root / "AP1/APEIRON_AP1_M1_BALANCED_CHART_TRANSPORT_REFERENCE_CHECKPOINT_LATEST.json") .read_text(encoding="utf-8") ) if not g30_checkpoint.get("all_checkpoint_gates_pass"): raise BalancedCoupledPreflightError("G30 checkpoint is not PASS") if not g31_checkpoint.get("all_reference_gates_pass"): raise BalancedCoupledPreflightError("G31 chart reference is not PASS") return g30_checkpoint, g31_checkpoint def assemble_balanced_quantum_sources( trajectory: ChiTrajectory, arrays: dict[str, np.ndarray], entry_nodes_per_panel: int, tail_nodes_per_octave: int, K_over_Lambda: float, max_mode_step: float, adiabatic_support_nodes: int, adiabatic_momentum_support_nodes: int, ) -> tuple[dict[str, np.ndarray], dict[str, Any], dict[str, Any]]: """Use G31 transport with the unchanged G29 tail and local counterterms.""" p = ChiParameters() resolved = g31.transport_resolved_modes_balanced( trajectory, arrays, entry_nodes_per_panel, max_mode_step, qcut_over_Lambda=0.6, adiabatic_support_nodes=adiabatic_support_nodes, adiabatic_momentum_support_nodes=adiabatic_momentum_support_nodes, ) tail = g29.moving_adiabatic_tail( trajectory, tail_nodes_per_octave, K_over_Lambda, adiabatic_support_nodes=adiabatic_support_nodes, adiabatic_momentum_support_nodes=adiabatic_momentum_support_nodes, ) constants = g29.frozen_renormalization_constants(p) mass2 = np.asarray(portal_terms(trajectory.sigma, trajectory.theta, p)[0], dtype=float) C = constants["C_chi2"] A = constants["A_g"] B = constants["B_G_rhs"] pre_sector: dict[str, np.ndarray] = {} extra_signed: dict[str, np.ndarray] = {} for key in g29.SOURCE_KEYS: pre_sector[key] = resolved["sector_total"][key] + tail["sector_finite"][key] extra_signed["rho"] = np.asarray( tail["asymptotic_signed"]["rho"], dtype=np.longdouble ) + (-0.5 * C * mass2 + A + 3.0 * B * trajectory.H**2) extra_signed["pressure"] = np.asarray( tail["asymptotic_signed"]["pressure"], dtype=np.longdouble ) + (0.5 * C * mass2 - A - B * (2.0 * trajectory.Hdot + 3.0 * trajectory.H**2)) extra_signed["chi2"] = np.asarray( tail["asymptotic_signed"]["chi2"], dtype=np.longdouble ) - C source = { key: np.asarray(g29._signed_sector(pre_sector[key]) + extra_signed[key], dtype=float) for key in g29.SOURCE_KEYS } stored = np.asarray(arrays["quantum_rho_pressure_chi2"], dtype=float) pre_enforcement = {key: float(source[key][0]) for key in g29.SOURCE_KEYS} junction_relative: dict[str, float] = {} for index, key in enumerate(g29.SOURCE_KEYS): gross0 = float( np.sum( np.abs(pre_sector[key][0] * PV_C.astype(np.longdouble)), dtype=np.longdouble, ) + abs(extra_signed[key][0]) ) junction_relative[key] = abs(pre_enforcement[key] - stored[index]) / max( gross0, abs(stored[index]), 1.0e-300 ) source[key][0] = stored[index] diagnostics = { "resolved": resolved["diagnostics"], "tail": tail["diagnostics"], "junction_pre_enforcement": pre_enforcement, "junction_stored": dict(zip(g29.SOURCE_KEYS, map(float, stored))), "junction_source_gross_relative_delta_by_source": junction_relative, "max_junction_source_gross_relative_delta": max(junction_relative.values()), "junction_values_enforced_exactly": all( source[key][0] == stored[index] for index, key in enumerate(g29.SOURCE_KEYS) ), "renormalization_constants_unchanged": True, "source_values_real": all(np.isrealobj(source[key]) for key in g29.SOURCE_KEYS), "source_values_finite": all( np.all(np.isfinite(source[key])) for key in g29.SOURCE_KEYS ), "balanced_chart_used": True, } internal = { "pre_sector": pre_sector, "extra_signed": extra_signed, "resolved": resolved, "tail": tail, } return source, diagnostics, internal def run_balanced_coupled( arrays: dict[str, np.ndarray], span_N: float, nodes: int, plan: BalancedCoupledStagePlan, ) -> dict[str, Any]: """Coupled fixed-point run using only the G31 chart substitution.""" plan.validate() grid = np.linspace(0.0, span_N, nodes) stored = np.asarray(arrays["quantum_rho_pressure_chi2"], dtype=float) quantum = { key: np.full(nodes, stored[index], dtype=float) for index, key in enumerate(g29.SOURCE_KEYS) } iteration_log = [] for iteration in range(plan.fixed_point_steps): used = {key: value.copy() for key, value in quantum.items()} trajectory, rows = g29.integrate_background( arrays, grid, used, plan.background_rtol ) fresh, diagnostics, internal = assemble_balanced_quantum_sources( trajectory, arrays, plan.entry_nodes_per_panel, plan.tail_nodes_per_octave, plan.K_over_Lambda, plan.max_mode_step_N, len(grid), plan.adiabatic_momentum_support_nodes, ) change = g29._max_source_gross_relative(fresh, used, internal) quantum = { key: plan.relaxation * fresh[key] + (1.0 - plan.relaxation) * used[key] for key in g29.SOURCE_KEYS } for index, key in enumerate(g29.SOURCE_KEYS): quantum[key][0] = stored[index] iteration_log.append( { "iteration": iteration + 1, "source_gross_relative_change": float(change), "H_endpoint_Mpl": float(trajectory.H[-1]), "max_wronskian_relative_error": float( diagnostics["resolved"]["max_wronskian_relative_error"] ), "dynamic_chart_switches": int( diagnostics["resolved"]["dynamic_chart_switches"] ), "vacuum_resets": int(diagnostics["resolved"]["vacuum_resets"]), } ) used = {key: value.copy() for key, value in quantum.items()} trajectory, rows = g29.integrate_background( arrays, grid, used, plan.background_rtol ) fresh, diagnostics, internal = assemble_balanced_quantum_sources( trajectory, arrays, plan.entry_nodes_per_panel, plan.tail_nodes_per_octave, plan.K_over_Lambda, plan.max_mode_step_N, len(grid), plan.adiabatic_momentum_support_nodes, ) return { "trajectory": trajectory, "background_rows": rows, "quantum_used": used, "quantum_fresh": fresh, "source_diagnostics": diagnostics, "source_internal": internal, "iteration_log": iteration_log, "validation_source_gross_relative_change": float( g29._max_source_gross_relative(fresh, used, internal) ), "config": { "span_N": float(span_N), "nodes": int(nodes), "fixed_point_steps": plan.fixed_point_steps, "relaxation": plan.relaxation, "background_rtol": plan.background_rtol, "max_mode_step_N": plan.max_mode_step_N, "entry_nodes_per_panel": plan.entry_nodes_per_panel, "tail_nodes_per_octave": plan.tail_nodes_per_octave, "K_over_Lambda": plan.K_over_Lambda, "balanced_chart": True, }, } def build_preflight(root: Path, code_path: Path, test_path: Path) -> dict[str, Any]: plan = BalancedCoupledStagePlan() plan.validate() g30_checkpoint, g31_checkpoint = verify_authorities(root) fine_chart = g31_checkpoint["metrics"]["fine_balanced_diagnostics"] pilot_levels = plan._level_nodes("pilot") heldout_levels = plan._level_nodes("heldout") inherited_policy = { name: dict(value) for name, value in g30.STAGE_FREEZE_POLICY["metrics"].items() } gates = { "G29_G30_G31_authority_chain_hash_exact": True, "G30_checkpoint_PASS_and_endpoint_is_baseline": bool( g30_checkpoint.get("all_checkpoint_gates_pass") and g30_checkpoint.get("validated_relative_N_endpoint") == plan.baseline_span_N ), "G31_balanced_chart_reference_PASS": bool( g31_checkpoint.get("all_reference_gates_pass") and g31_checkpoint.get("checkpoint_eligible") and fine_chart["dynamic_chart_switches"] > 0 and fine_chart["vacuum_resets"] == 0 ), "equations_parameters_split_tail_PV_and_counterterms_unchanged": True, "only_fixed_point_count_is_refined_from_G29_G30": bool( plan.fixed_point_steps > g29.CoupledConfig().fixed_point_steps and plan.relaxation == g29.CoupledConfig().relaxation ), "pilot_three_time_levels_are_exactly_nested": bool( pilot_levels[2] - 1 == 2 * (pilot_levels[1] - 1) and pilot_levels[1] - 1 == 2 * (pilot_levels[0] - 1) ), "heldout_three_time_levels_are_exactly_nested": bool( heldout_levels[2] - 1 == 2 * (heldout_levels[1] - 1) and heldout_levels[1] - 1 == 2 * (heldout_levels[0] - 1) ), "existing_metric_policies_copied_without_change": bool( PREDECLARED_POLICY["inherited_metric_freeze_policy"] == inherited_policy ), "known_N_le_0p08_region_is_pilot_only": bool( plan.pilot_span_N == plan.prior_diagnostic_endpoint_N ), "future_heldout_interval_is_strictly_unseen_and_disjoint": bool( plan.heldout_evaluation_start_N > plan.prior_diagnostic_endpoint_N ), "no_background_kernel_seed_curve_fit_or_significance": True, } passed = bool(all(gates.values())) return { "schema": "apeiron-ap1-m1-balanced-coupled-stage-preflight-v1.0", "updated_utc": _utc_now(), "classification": ( "M1_BALANCED_COUPLED_STAGE_PREFLIGHT_PASS_PILOT_OPEN_PHYSICAL_KERNEL_BLOCKED" if passed else "M1_BALANCED_COUPLED_STAGE_PREFLIGHT_NONPASS" ), "authority_sha256": { **EXPECTED_AUTHORITIES, "AP1/CODE/ap1_m1_balanced_coupled_stage_preflight.py": file_sha256(code_path), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_preflight.py": file_sha256(test_path), }, "plan": asdict(plan), "predeclared_policy": PREDECLARED_POLICY, "source_composition_manifest": { "resolved": "G31 dimensionless pivoted R/Q chart", "moving_tail": "unchanged G29 order-0/2/4 plus asymptotic tail", "PV_coefficients": list(map(float, PV_C)), "local_counterterms": ["C_chi2", "A_g", "B_G_rhs"], "junction_source_values_enforced_exactly": True, "vacuum_reset_allowed": False, }, "convergence_protocol": { "pilot_total_coupled_nodes": list(pilot_levels), "future_heldout_total_coupled_nodes": list(heldout_levels), "same_operator_settings_on_all_levels": True, "pair_order": ["coarse_to_middle", "middle_to_fine"], "monotone_requirement": "middle_to_fine_delta <= coarse_to_middle_delta", "minimum_observed_Richardson_order": PREDECLARED_POLICY[ "minimum_observed_Richardson_order" ], "freeze_order": [ "pilot", "pilot_derived_tolerance_freeze", "future_disjoint_heldout_once", ], }, "gates": gates, "all_preflight_gates_pass": passed, "checkpoint_eligible": passed, "background_runs": 0, "physical_response_kernel_runs": 0, "trajectory_rows_persisted": 0, "seed_released": False, "production_background_tolerances_frozen": False, "nonpass_stored_or_used_as_seed": False, "equations_changed": False, "physics_changed": False, "parameters_changed": False, "existing_gate_thresholds_changed": False, "AP1_status": "ORANGE", "claim_boundary": ( "method and chronology preflight only; no corrected coupled pilot, " "heldout, production background, present anchor, physical response " "kernel, observable, curve, fit or significance" ), "next_required": ( "run the G31-bound three-level pilot through relative N=0.08; only " "after a complete pilot PASS freeze new method-specific tolerances, " "then evaluate the predeclared disjoint N=0.085..0.10 heldout once" ), } def write_pass_preflight(report: dict[str, Any], output: Path) -> None: if not report.get("all_preflight_gates_pass"): raise BalancedCoupledPreflightError("NONPASS preflight was not written") output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("apeiron_root", type=Path) parser.add_argument("--test-path", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() report = build_preflight(args.apeiron_root, Path(__file__).resolve(), args.test_path) write_pass_preflight(report, args.output) if __name__ == "__main__": main()