"""Balanced two-chart mode transport for staged AP1-M1 background windows. G29 validated the coupled moving-split architecture on relative N=0..0.02, but its single v/u Riccati chart is not globally regular: a highly squeezed mode can make v/u arbitrarily large even though the underlying linear mode is finite. This module adds no physics. It advances the same canonical mode equation with a real, determinant-one 2x2 transfer and pivots between the dimensionless charts R = v / (Lambda u), Q = Lambda u / v = 1/R. The stored amplitude is log|u| in the R chart and log|v/Lambda| in the Q chart. The larger scaled component is always used as the chart denominator, so no Riccati pole or separate exp(2 log|Im R|) is formed. The Pauli-Villars basis, adiabatic order-0/2/4 handoff, moving q_cut=0.6 Lambda boundary and all background equations remain those of the hash-bound G29 implementation. """ from __future__ import annotations import argparse from dataclasses import dataclass 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_coupled_moving_split_background as g29 from ap1_m1_log_amplitude_transport_reference import seed_mode_state from chi_background_closure import ChiParameters, ChiTrajectory, PV_J, portal_terms EXPECTED_AUTHORITIES = { "AP1/APEIRON_AP1_M1_COUPLED_MOVING_SPLIT_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json": ( "c2f90da969a139fbb690b26fc12e821a6dcf82e126dffa0b1794a3ff6fd64cd9" ), "AP1/CODE/ap1_m1_coupled_moving_split_background.py": ( "3b6f2b38d047db3cb9510af9c2c92ef9c80c2c1ca16c40212f7e7c354a76f400" ), "AP1/APEIRON_AP1_M1_STAGED_LOG_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json": ( "6eeb0ef13ab072a1fb04bb47253d01498e2b7e28fd740c2de13fbdc8b96690df" ), "AP1/CODE/ap1_m1_staged_log_background_extension.py": ( "25ce7b0a19582b1e7e57be7339914ddebd41a902eb4503803bc69d18ea9ce5ae" ), } # These are structural numerical ceilings, fixed before the formal reference # evaluation below. The Wronskian ceiling is inherited unchanged from G27-G29. # None is a physical-background or response-kernel gate. REFERENCE_POLICY: dict[str, float | int] = { "overlap_span_N": 0.02, "overlap_nodes": 65, "pole_traversal_span_N": 0.13, "pole_traversal_nodes": 417, "adiabatic_momentum_support_nodes": 129, "entry_nodes_per_panel": 1, "coarse_mode_step_N": 2.5e-5, "fine_mode_step_N": 1.25e-5, "overlap_source_gross_relative_cap": 1.0e-5, "mode_step_source_gross_relative_cap": 5.0e-3, "wronskian_relative_cap": 1.0e-11, "active_ratio_cap": 1.0 + 2.0e-15, "transfer_determinant_defect_cap": 1.0e-4, } class BalancedChartError(g29.CoupledBackgroundError): """A balanced-chart numerical or invariant gate failed.""" @dataclass class BalancedState: """Per-sector, per-mode projective state in one of two regular charts.""" q_chart: np.ndarray log_denominator_amplitude: np.ndarray ratio: np.ndarray def validated(self) -> "BalancedState": chart = np.asarray(self.q_chart, dtype=bool) amplitude = np.asarray(self.log_denominator_amplitude, dtype=np.longdouble) ratio = np.asarray(self.ratio, dtype=np.clongdouble) if chart.shape != amplitude.shape or chart.shape != ratio.shape: raise BalancedChartError("balanced chart arrays must share one shape") if chart.ndim != 2 or chart.shape[0] != 4 or chart.shape[1] < 1: raise BalancedChartError("four PV sectors and at least one mode required") if not np.all(np.isfinite(amplitude)) or not np.all(np.isfinite(ratio)): raise BalancedChartError("balanced state is nonfinite") imag = np.imag(ratio) if np.any((~chart) & (imag >= 0.0)) or np.any(chart & (imag <= 0.0)): raise BalancedChartError("balanced chart lost canonical imaginary sign") if float(np.max(np.abs(ratio))) > 1.0 + 2.0e-15: raise BalancedChartError("balanced chart denominator is not dominant") return BalancedState(chart.copy(), amplitude.copy(), ratio.copy()) def _abs2(value: np.ndarray) -> np.ndarray: z = np.asarray(value, dtype=np.clongdouble) return np.asarray(z.real * z.real + z.imag * z.imag, dtype=np.longdouble) def from_u_riccati( log_u: np.ndarray, riccati_real: np.ndarray, log_abs_riccati_imag: np.ndarray, frequency_scale: float, ) -> BalancedState: """Convert the validated G27/G29 u-chart into the pivoted chart.""" nu = np.longdouble(frequency_scale) if not np.isfinite(nu) or nu <= 0.0: raise ValueError("positive finite chart frequency scale required") L = np.asarray(log_u, dtype=np.longdouble) x = np.asarray(riccati_real, dtype=np.longdouble) logb = np.asarray(log_abs_riccati_imag, dtype=np.longdouble) if L.shape != x.shape or L.shape != logb.shape: raise ValueError("Riccati arrays must share one shape") if float(np.max(logb)) > math.log(np.finfo(np.longdouble).max) - 4.0: raise BalancedChartError("input Riccati chart already exceeds long-double range") beta = np.exp(logb) R = np.asarray((x - np.clongdouble(1j) * beta) / nu, dtype=np.clongdouble) q_chart = np.asarray(np.abs(R) > 1.0, dtype=bool) amplitude = L.copy() ratio = R.copy() if np.any(q_chart): magnitude = np.asarray(np.abs(R[q_chart]), dtype=np.longdouble) amplitude[q_chart] = L[q_chart] + np.log(magnitude) ratio[q_chart] = np.clongdouble(1.0) / R[q_chart] return BalancedState(q_chart, amplitude, ratio).validated() def to_u_quadratics( state: BalancedState, frequency_scale: float ) -> tuple[np.ndarray, np.ndarray, np.ndarray]: """Return |u|^2, Re(conj(u)v), |v|^2 without a chart singularity.""" value = state.validated() nu = np.longdouble(frequency_scale) base = np.exp(np.longdouble(2.0) * value.log_denominator_amplitude) u2 = np.empty_like(base) cross = np.empty_like(base) v2 = np.empty_like(base) r_chart = ~value.q_chart if np.any(r_chart): R = value.ratio[r_chart] local = base[r_chart] u2[r_chart] = local cross[r_chart] = nu * np.asarray(R.real, dtype=np.longdouble) * local v2[r_chart] = nu * nu * _abs2(R) * local if np.any(value.q_chart): Q = value.ratio[value.q_chart] local = base[value.q_chart] u2[value.q_chart] = _abs2(Q) * local cross[value.q_chart] = nu * np.asarray(Q.real, dtype=np.longdouble) * local v2[value.q_chart] = nu * nu * local return u2, cross, v2 def log_wronskian(state: BalancedState, frequency_scale: float) -> np.ndarray: """Evaluate log[-2 Im(conj(u)v)] in the active regular chart.""" value = state.validated() nu = np.longdouble(frequency_scale) imag = np.asarray(np.imag(value.ratio), dtype=np.longdouble) signed = np.where(value.q_chart, imag, -imag) if np.any(signed <= 0.0): raise BalancedChartError("nonpositive chart Wronskian factor") return ( np.log(np.longdouble(2.0) * nu) + np.longdouble(2.0) * value.log_denominator_amplitude + np.log(signed) ) def _generator_coefficients( n_value: float, k: np.ndarray, trajectory: ChiTrajectory, p: ChiParameters, frequency_scale: np.longdouble, ) -> tuple[np.ndarray, np.ndarray]: H = np.longdouble(g29._trajectory_value(trajectory, "H", n_value)) Hdot = np.longdouble(g29._trajectory_value(trajectory, "Hdot", n_value)) sigma = g29._trajectory_value(trajectory, "sigma", n_value) theta = g29._trajectory_value(trajectory, "theta", n_value) mass2 = np.longdouble(float(portal_terms(sigma, theta, p)[0])) muR2 = np.longdouble((p.muR_over_Lambda * p.Lambda) ** 2) kval = np.asarray(k, dtype=np.longdouble) omega2 = ( kval[None, :] ** 2 * np.exp(np.longdouble(-2.0 * n_value)) + mass2 + PV_J.astype(np.longdouble)[:, None] * muR2 - np.longdouble(2.25) * H * H - np.longdouble(1.5) * Hdot ) if H <= 0.0 or not np.all(np.isfinite(omega2)): raise BalancedChartError("invalid balanced-chart generator") upper = np.full_like(omega2, frequency_scale / H) lower = -omega2 / (frequency_scale * H) return upper, lower def _left_multiply_generator( upper: np.ndarray, lower: np.ndarray, matrix: tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray], ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: m00, m01, m10, m11 = matrix return upper * m10, upper * m11, lower * m00, lower * m01 def _rk4_transfer( left: float, right: float, k: np.ndarray, trajectory: ChiTrajectory, p: ChiParameters, frequency_scale: np.longdouble, ) -> tuple[tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray], float]: """Fourth-order real transfer, projected onto determinant one.""" h = np.longdouble(right - left) if h <= 0.0: raise ValueError("positive transfer interval required") midpoint = 0.5 * (left + right) a1, b1 = _generator_coefficients(left, k, trajectory, p, frequency_scale) a2, b2 = _generator_coefficients(midpoint, k, trajectory, p, frequency_scale) a4, b4 = _generator_coefficients(right, k, trajectory, p, frequency_scale) zero = np.zeros_like(a1) one = np.ones_like(a1) k1 = (zero, a1, b1, zero) y2 = tuple(base + np.longdouble(0.5) * h * value for base, value in zip((one, zero, zero, one), k1)) k2 = _left_multiply_generator(a2, b2, y2) y3 = tuple(base + np.longdouble(0.5) * h * value for base, value in zip((one, zero, zero, one), k2)) k3 = _left_multiply_generator(a2, b2, y3) y4 = tuple(base + h * value for base, value in zip((one, zero, zero, one), k3)) k4 = _left_multiply_generator(a4, b4, y4) transfer = tuple( base + h * (v1 + np.longdouble(2.0) * v2 + np.longdouble(2.0) * v3 + v4) / np.longdouble(6.0) for base, v1, v2, v3, v4 in zip((one, zero, zero, one), k1, k2, k3, k4) ) t00, t01, t10, t11 = transfer determinant = t00 * t11 - t01 * t10 if np.any(determinant <= 0.0) or not np.all(np.isfinite(determinant)): raise BalancedChartError("RK4 transfer lost positive determinant") defect = float(np.max(np.abs(determinant - np.longdouble(1.0)))) normalization = np.sqrt(determinant) projected = tuple(value / normalization for value in transfer) return projected, defect def _apply_transfer( state: BalancedState, transfer: tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray], ) -> tuple[BalancedState, int]: value = state.validated() t00, t01, t10, t11 = transfer ratio = value.ratio u_factor = np.empty_like(ratio) w_factor = np.empty_like(ratio) r_chart = ~value.q_chart if np.any(r_chart): R = ratio[r_chart] u_factor[r_chart] = t00[r_chart] + t01[r_chart] * R w_factor[r_chart] = t10[r_chart] + t11[r_chart] * R if np.any(value.q_chart): Q = ratio[value.q_chart] u_factor[value.q_chart] = t00[value.q_chart] * Q + t01[value.q_chart] w_factor[value.q_chart] = t10[value.q_chart] * Q + t11[value.q_chart] abs_u = np.asarray(np.abs(u_factor), dtype=np.longdouble) abs_w = np.asarray(np.abs(w_factor), dtype=np.longdouble) q_chart = abs_w > abs_u selected = np.where(q_chart, abs_w, abs_u) if np.any(selected <= 0.0) or not np.all(np.isfinite(selected)): raise BalancedChartError("both balanced transfer components vanished") updated_ratio = np.empty_like(ratio) if np.any(~q_chart): updated_ratio[~q_chart] = w_factor[~q_chart] / u_factor[~q_chart] if np.any(q_chart): updated_ratio[q_chart] = u_factor[q_chart] / w_factor[q_chart] updated = BalancedState( q_chart, value.log_denominator_amplitude + np.log(selected), updated_ratio, ).validated() return updated, int(np.count_nonzero(q_chart != value.q_chart)) def advance_balanced_state( state: BalancedState, k: np.ndarray, left: float, right: float, trajectory: ChiTrajectory, max_step: float, p: ChiParameters | None = None, frequency_scale: float | None = None, ) -> tuple[BalancedState, dict[str, float | int]]: p = ChiParameters() if p is None else p nu = np.longdouble(p.Lambda if frequency_scale is None else frequency_scale) if right < left or max_step <= 0.0: raise ValueError("ordered interval and positive step required") current = state.validated() if right == left: return current, {"chart_switches": 0, "max_transfer_determinant_defect": 0.0} steps = max(1, int(math.ceil((right - left) / max_step))) h = (right - left) / steps switches = 0 defect = 0.0 cursor = float(left) for index in range(steps): endpoint = float(right) if index == steps - 1 else float(left + (index + 1) * h) transfer, local_defect = _rk4_transfer(cursor, endpoint, k, trajectory, p, nu) current, local_switches = _apply_transfer(current, transfer) switches += local_switches defect = max(defect, local_defect) cursor = endpoint return current, { "chart_switches": switches, "max_transfer_determinant_defect": defect, } def _balanced_mode_sector_sources( n_value: float, H: float, sigma: float, theta: float, k: np.ndarray, weights: np.ndarray, state: BalancedState, p: ChiParameters, frequency_scale: float, ) -> dict[str, np.ndarray]: value = state.validated() nu = np.longdouble(frequency_scale) n_ld = np.longdouble(n_value) H_ld = np.longdouble(H) dilution = np.exp(np.longdouble(-3.0) * n_ld) base = np.exp(np.longdouble(2.0) * value.log_denominator_amplitude) * dilution amplitude2 = np.empty_like(base) kinetic = np.empty_like(base) r_chart = ~value.q_chart if np.any(r_chart): R = value.ratio[r_chart] amplitude2[r_chart] = base[r_chart] kinetic[r_chart] = _abs2(nu * R - np.longdouble(1.5) * H_ld) * base[r_chart] if np.any(value.q_chart): Q = value.ratio[value.q_chart] amplitude2[value.q_chart] = _abs2(Q) * base[value.q_chart] kinetic[value.q_chart] = _abs2(nu - np.longdouble(1.5) * H_ld * Q) * base[value.q_chart] mass2 = np.longdouble(float(portal_terms(sigma, theta, p)[0])) muR2 = np.longdouble((p.muR_over_Lambda * p.Lambda) ** 2) kval = np.asarray(k, dtype=np.longdouble) q2 = kval[None, :] ** 2 * np.exp(np.longdouble(-2.0) * n_ld) masses2 = mass2 + PV_J.astype(np.longdouble)[:, None] * muR2 each = { "rho": np.longdouble(0.5) * (kinetic + (q2 + masses2) * amplitude2), "pressure": np.longdouble(0.5) * kinetic - ( q2 / np.longdouble(6.0) + np.longdouble(0.5) * masses2 ) * amplitude2, "chi2": amplitude2, } w = np.asarray(weights, dtype=np.longdouble) return { key: np.asarray(np.sum(local * w[None, :], axis=1, dtype=np.longdouble), dtype=np.longdouble) for key, local in each.items() } def _slice_state(state: BalancedState, stop: int) -> BalancedState: return BalancedState( state.q_chart[:, :stop], state.log_denominator_amplitude[:, :stop], state.ratio[:, :stop], ) def transport_resolved_modes_balanced( trajectory: ChiTrajectory, arrays: dict[str, np.ndarray], entry_nodes_per_panel: int, max_step: float, qcut_over_Lambda: float = 0.6, adiabatic_support_nodes: int | None = None, adiabatic_momentum_support_nodes: int = 257, ) -> dict[str, Any]: """G29-compatible resolved source transport with automatic chart pivots.""" tr = trajectory.validated() p = ChiParameters() if qcut_over_Lambda != 0.6: raise ValueError("the frozen moving split is q_cut/Lambda = 0.6") qcut = qcut_over_Lambda * p.Lambda nu = p.Lambda support_nodes = min(len(tr.N), 65) if adiabatic_support_nodes is None else int(adiabatic_support_nodes) handoff_table = g29.boundary_handoff_table( tr, support_nodes, qcut, p, adiabatic_momentum_support_nodes ) seed_indices = tuple(range(len(np.asarray(arrays["k_comoving_Mpl"])))) seed = seed_mode_state(arrays, seed_indices) state = from_u_riccati( seed["log_amplitude"], seed["riccati_real"], seed["log_abs_riccati_imag"], nu, ) log_wronskian_target = np.log(np.asarray(seed["wronskian"], dtype=np.longdouble)) k = np.asarray(arrays["k_comoving_Mpl"], dtype=np.longdouble).copy() weights = np.asarray(arrays["shell_weights_Mpl3"], dtype=np.longdouble).copy() seed_count = len(k) sector_total = {key: np.zeros((len(tr.N), 4), dtype=np.longdouble) for key in g29.SOURCE_KEYS} sector_seed = {key: np.zeros((len(tr.N), 4), dtype=np.longdouble) for key in g29.SOURCE_KEYS} sector_new = {key: np.zeros((len(tr.N), 4), dtype=np.longdouble) for key in g29.SOURCE_KEYS} entry_x, entry_w = np.polynomial.legendre.leggauss(int(entry_nodes_per_panel)) max_identity = 0.0 max_state_error = 0.0 max_initial_wronskian = 0.0 max_det_defect = 0.0 chart_switches = 0 min_entry_omega2 = math.inf min_entry_W = math.inf max_W2 = 0.0 max_W4 = 0.0 crossings = 0 def record(index: int) -> None: nonlocal max_identity local = _balanced_mode_sector_sources( float(tr.N[index]), float(tr.H[index]), float(tr.sigma[index]), float(tr.theta[index]), k, weights, state, p, nu, ) seed_local = _balanced_mode_sector_sources( float(tr.N[index]), float(tr.H[index]), float(tr.sigma[index]), float(tr.theta[index]), k[:seed_count], weights[:seed_count], _slice_state(state, seed_count), p, nu, ) for key in g29.SOURCE_KEYS: sector_total[key][index] = local[key] sector_seed[key][index] = seed_local[key] sector_new[key][index] = local[key] - seed_local[key] identity = log_wronskian(state, nu) - log_wronskian_target max_identity = max(max_identity, float(np.max(np.abs(identity)))) record(0) for panel in range(len(tr.N) - 1): left = float(tr.N[panel]) right = float(tr.N[panel + 1]) state, audit = advance_balanced_state(state, k, left, right, tr, max_step, p, nu) chart_switches += int(audit["chart_switches"]) max_det_defect = max(max_det_defect, float(audit["max_transfer_determinant_defect"])) midpoint = 0.5 * (left + right) halfwidth = 0.5 * (right - left) entries = midpoint + halfwidth * entry_x entry_weights = halfwidth * entry_w births: list[BalancedState] = [] birth_k: list[float] = [] birth_weights: list[float] = [] for entry, quadrature_weight in zip(entries, entry_weights): initial, handoff = g29._handoff_state_from_table(handoff_table, float(entry)) balanced = from_u_riccati( initial["log_amplitude"], initial["riccati_real"], initial["log_abs_riccati_imag"], nu, ) k_entry = qcut * math.exp(float(entry)) advanced, birth_audit = advance_balanced_state( balanced, np.array([k_entry], dtype=np.longdouble), float(entry), right, tr, max_step, p, nu, ) births.append(advanced) birth_k.append(k_entry) birth_weights.append(float(quadrature_weight) * k_entry**3 / (2.0 * math.pi**2)) chart_switches += int(birth_audit["chart_switches"]) max_det_defect = max(max_det_defect, float(birth_audit["max_transfer_determinant_defect"])) min_entry_omega2 = min(min_entry_omega2, handoff["min_omega2"]) min_entry_W = min(min_entry_W, handoff["min_W"]) max_W2 = max(max_W2, handoff["max_abs_W2_over_W0"]) max_W4 = max(max_W4, handoff["max_abs_W4_over_W0"]) max_state_error = max(max_state_error, handoff["state_transfer_relative_error"]) max_initial_wronskian = max(max_initial_wronskian, handoff["initial_wronskian_relative_error"]) crossings += 1 if births: state = BalancedState( np.concatenate([state.q_chart] + [birth.q_chart for birth in births], axis=1), np.concatenate([state.log_denominator_amplitude] + [birth.log_denominator_amplitude for birth in births], axis=1), np.concatenate([state.ratio] + [birth.ratio for birth in births], axis=1), ).validated() k = np.concatenate([k, np.asarray(birth_k, dtype=np.longdouble)]) weights = np.concatenate([weights, np.asarray(birth_weights, dtype=np.longdouble)]) log_wronskian_target = np.concatenate( [log_wronskian_target, np.zeros((4, len(birth_k)), dtype=np.longdouble)], axis=1 ) record(panel + 1) signed_imag = np.where(state.q_chart, np.imag(state.ratio), -np.imag(state.ratio)) return { "sector_total": sector_total, "sector_seed": sector_seed, "sector_new": sector_new, "diagnostics": { "seed_modes": seed_count, "new_entry_quadrature_modes": crossings, "final_resolved_modes": int(len(k)), "max_log_wronskian_identity_error": max_identity, "max_wronskian_relative_error": float(abs(math.expm1(min(max_identity, 1.0)))), "min_entry_omega2_Mpl2": min_entry_omega2, "min_entry_W_Mpl": min_entry_W, "max_abs_entry_W2_over_W0": max_W2, "max_abs_entry_W4_over_W0": max_W4, "max_handoff_state_relative_error": max_state_error, "max_initial_handoff_wronskian_relative_error": max_initial_wronskian, "vacuum_resets": 0, "entry_measure": "q_cut^3*exp(3*N_entry)/(2*pi^2) dN_entry", "adiabatic_support_nodes": support_nodes, "adiabatic_momentum_support_nodes": adiabatic_momentum_support_nodes, "max_local_to_independent_fixed_k_state_relative_delta": handoff_table[ "max_local_to_independent_fixed_k_state_relative_delta" ], "balanced_frequency_scale_Mpl": nu, "dynamic_chart_switches": chart_switches, "final_Q_chart_entries": int(np.count_nonzero(state.q_chart)), "max_abs_active_ratio": float(np.max(np.abs(state.ratio))), "min_abs_active_imaginary_ratio": float(np.min(signed_imag)), "max_transfer_determinant_defect_before_projection": max_det_defect, "transport_method": "dimensionless_pivoted_R_Q_charts_with_determinant_one_RK4_transfer", }, } 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(apeiron_root: Path) -> tuple[dict[str, np.ndarray], dict[str, str]]: observed = { name: file_sha256(apeiron_root / name) for name in EXPECTED_AUTHORITIES } if observed != EXPECTED_AUTHORITIES: mismatches = [ name for name, expected in EXPECTED_AUTHORITIES.items() if observed.get(name) != expected ] raise BalancedChartError( "balanced-chart authority mismatch: " + ", ".join(mismatches) ) checkpoint = json.loads( ( apeiron_root / "AP1/APEIRON_AP1_M1_COUPLED_MOVING_SPLIT_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json" ).read_text(encoding="utf-8") ) if not checkpoint.get("all_checkpoint_gates_pass"): raise BalancedChartError("G29 checkpoint is not PASS") staged_checkpoint = json.loads( ( apeiron_root / "AP1/APEIRON_AP1_M1_STAGED_LOG_BACKGROUND_REFERENCE_CHECKPOINT_LATEST.json" ).read_text(encoding="utf-8") ) if not staged_checkpoint.get("all_checkpoint_gates_pass"): raise BalancedChartError("G30 checkpoint is not PASS") _seed_report, arrays, _authorities = g29.load_inputs(apeiron_root) return arrays, observed def _fixed_stored_source_trajectory( arrays: dict[str, np.ndarray], span_N: float, nodes: int ) -> ChiTrajectory: grid = np.linspace(0.0, float(span_N), int(nodes)) stored = np.asarray(arrays["quantum_rho_pressure_chi2"], dtype=float) quantum = { key: np.full(len(grid), stored[index], dtype=float) for index, key in enumerate(g29.SOURCE_KEYS) } trajectory, _rows = g29.integrate_background( arrays, grid, quantum, rtol=2.0e-10 ) return trajectory def _resolved_signed_gross_relative_delta( left: dict[str, Any], right: dict[str, Any] ) -> tuple[float, dict[str, float]]: by_source: dict[str, float] = {} coefficients = g29.PV_C.astype(np.longdouble) for key in g29.SOURCE_KEYS: a = np.asarray(left["sector_total"][key], dtype=np.longdouble) b = np.asarray(right["sector_total"][key], dtype=np.longdouble) signed_a = np.sum(a * coefficients[None, :], axis=1, dtype=np.longdouble) signed_b = np.sum(b * coefficients[None, :], axis=1, dtype=np.longdouble) gross = np.maximum( np.sum(np.abs(a * coefficients[None, :]), axis=1, dtype=np.longdouble), np.sum(np.abs(b * coefficients[None, :]), axis=1, dtype=np.longdouble), ) by_source[key] = float( np.max(np.abs(signed_a - signed_b)) / max(float(np.max(gross)), 1.0e-300) ) return max(by_source.values()), by_source def build_reference(apeiron_root: Path, code_path: Path) -> dict[str, Any]: """Build a structural PASS reference without promoting a background row.""" arrays, authorities = verify_authorities(apeiron_root) policy = REFERENCE_POLICY overlap = _fixed_stored_source_trajectory( arrays, float(policy["overlap_span_N"]), int(policy["overlap_nodes"]) ) overlap_old = g29.transport_resolved_modes( overlap, arrays, int(policy["entry_nodes_per_panel"]), float(policy["fine_mode_step_N"]), adiabatic_support_nodes=int(policy["overlap_nodes"]), adiabatic_momentum_support_nodes=int(policy["adiabatic_momentum_support_nodes"]), ) overlap_balanced = transport_resolved_modes_balanced( overlap, arrays, int(policy["entry_nodes_per_panel"]), float(policy["fine_mode_step_N"]), adiabatic_support_nodes=int(policy["overlap_nodes"]), adiabatic_momentum_support_nodes=int(policy["adiabatic_momentum_support_nodes"]), ) overlap_delta, overlap_by_source = _resolved_signed_gross_relative_delta( overlap_old, overlap_balanced ) pole_trajectory = _fixed_stored_source_trajectory( arrays, float(policy["pole_traversal_span_N"]), int(policy["pole_traversal_nodes"]), ) old_chart_failed_closed = False old_chart_failure_class = None try: g29.transport_resolved_modes( pole_trajectory, arrays, int(policy["entry_nodes_per_panel"]), float(policy["fine_mode_step_N"]), adiabatic_support_nodes=int(policy["pole_traversal_nodes"]), adiabatic_momentum_support_nodes=int(policy["adiabatic_momentum_support_nodes"]), ) except g29.CoupledBackgroundError as exc: old_chart_failed_closed = True old_chart_failure_class = type(exc).__name__ pole_coarse = transport_resolved_modes_balanced( pole_trajectory, arrays, int(policy["entry_nodes_per_panel"]), float(policy["coarse_mode_step_N"]), adiabatic_support_nodes=int(policy["pole_traversal_nodes"]), adiabatic_momentum_support_nodes=int(policy["adiabatic_momentum_support_nodes"]), ) pole_fine = transport_resolved_modes_balanced( pole_trajectory, arrays, int(policy["entry_nodes_per_panel"]), float(policy["fine_mode_step_N"]), adiabatic_support_nodes=int(policy["pole_traversal_nodes"]), adiabatic_momentum_support_nodes=int(policy["adiabatic_momentum_support_nodes"]), ) step_delta, step_by_source = _resolved_signed_gross_relative_delta( pole_coarse, pole_fine ) coarse_diagnostics = pole_coarse["diagnostics"] fine_diagnostics = pole_fine["diagnostics"] gates = { "G29_and_G30_checkpoints_and_code_hash_bound_PASS": True, "frozen_qcut_PV_basis_and_background_equations_unchanged": bool( np.array_equal(g29.PV_C, np.array([1.0, -3.0, 3.0, -1.0])) and np.array_equal(PV_J, np.array([0.0, 1.0, 2.0, 3.0])) ), "short_overlap_with_single_chart_below_predeclared_cap": bool( overlap_delta <= float(policy["overlap_source_gross_relative_cap"]) ), "single_chart_pole_fails_closed": old_chart_failed_closed, "balanced_chart_crosses_same_pole_without_reset": bool( int(fine_diagnostics["dynamic_chart_switches"]) > 0 and int(fine_diagnostics["vacuum_resets"]) == 0 ), "nested_mode_step_delta_below_predeclared_cap": bool( step_delta <= float(policy["mode_step_source_gross_relative_cap"]) ), "coarse_and_fine_wronskians_below_existing_cap": bool( max( float(coarse_diagnostics["max_wronskian_relative_error"]), float(fine_diagnostics["max_wronskian_relative_error"]), ) <= float(policy["wronskian_relative_cap"]) ), "active_ratios_remain_in_dominant_denominator_chart": bool( max( float(coarse_diagnostics["max_abs_active_ratio"]), float(fine_diagnostics["max_abs_active_ratio"]), ) <= float(policy["active_ratio_cap"]) ), "preprojection_transfer_defect_below_predeclared_cap": bool( max( float(coarse_diagnostics["max_transfer_determinant_defect_before_projection"]), float(fine_diagnostics["max_transfer_determinant_defect_before_projection"]), ) <= float(policy["transfer_determinant_defect_cap"]) ), "no_background_seed_kernel_curve_fit_or_significance": True, } passed = bool(all(gates.values())) return { "schema": "apeiron-ap1-m1-balanced-chart-transport-reference-v1.0", "updated_utc": _utc_now(), "classification": ( "M1_BALANCED_CHART_TRANSPORT_REFERENCE_PASS_STAGED_BACKGROUND_CONTINUATION_OPEN" if passed else "M1_BALANCED_CHART_TRANSPORT_REFERENCE_NONPASS" ), "authority_sha256": { **authorities, "AP1/CODE/ap1_m1_balanced_chart_transport.py": file_sha256(code_path), }, "predeclared_reference_policy": policy, "metrics": { "short_overlap_source_gross_relative_delta": overlap_delta, "short_overlap_by_source": overlap_by_source, "pole_traversal_mode_step_source_gross_relative_delta": step_delta, "pole_traversal_mode_step_by_source": step_by_source, "old_single_chart_failed_closed": old_chart_failed_closed, "old_single_chart_failure_class": old_chart_failure_class, "coarse_balanced_diagnostics": coarse_diagnostics, "fine_balanced_diagnostics": fine_diagnostics, }, "gates": gates, "all_reference_gates_pass": passed, "checkpoint_eligible": passed, "fixed_stored_source_canary_only": True, "new_AP1_M1_background_candidates": 0, "trajectory_rows_persisted": 0, "physical_response_kernel_started": False, "seed_released": False, "nonpass_stored_or_used_as_seed": False, "equations_changed": False, "physics_changed": False, "parameters_changed": False, "existing_gate_thresholds_changed": False, "claim_boundary": ( "projective mode-chart and fixed-stored-source transport reference only; " "not a coupled long background, present-anchor candidate, response kernel, " "observable, curve, fit or significance" ), "next_required": ( "use the balanced chart in a staged coupled N-window pilot; retain the " "existing G29 physical and convergence gates" ), } def write_pass_reference(report: dict[str, Any], output: Path) -> None: if not report.get("all_reference_gates_pass"): raise BalancedChartError("refusing to persist a non-PASS balanced reference") output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("apeiron_root", type=Path) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() report = build_reference(args.apeiron_root, Path(__file__).resolve()) write_pass_reference(report, args.output) if __name__ == "__main__": main()