"""Three-level fail-closed pilot and heldout for the G32 AP1-M1 method. This module is numerical infrastructure only. It changes no Apeiron/AME-1 v7.13 or M-1 equation, physical parameter, Pauli--Villars sector, moving split, counterterm, or existing gate threshold. The G31 projective R/Q transport is used inside the G29 coupled fixed-point background solver exactly as declared by the hash-bound G32 preflight. Only aggregate diagnostics may be serialized. A pilot NONPASS, tolerance freeze NONPASS, or heldout NONPASS is returned to the caller but is never written by the command-line interface. No trajectory row or seed is emitted. """ from __future__ import annotations import argparse from dataclasses import asdict from datetime import datetime, timezone from hashlib import sha256 import json import math from pathlib import Path from typing import Any, Callable import numpy as np import ap1_m1_balanced_chart_transport as g31 import ap1_m1_balanced_coupled_precision_preflight as g32a import ap1_m1_balanced_coupled_stage_preflight as g32 import ap1_m1_coupled_moving_split_background as g29 EXPECTED_G32_AUTHORITIES = { "AP1/APEIRON_AP1_M1_BALANCED_COUPLED_STAGE_PREFLIGHT_LATEST.json": ( "5e544bfbe4803fe8668285a80267549ee5d0f3516d6db2ce03f548ed2d0f3333" ), "AP1/APEIRON_AP1_M1_BALANCED_COUPLED_STAGE_PREFLIGHT_CHECKPOINT_LATEST.md": ( "d009955b1a6c40dd07c413d468925f05967141a3f59a89f95141f855ea9e9ef0" ), "AP1/CODE/ap1_m1_balanced_coupled_stage_preflight.py": ( "a2e1c19db0d18c05c38e62254bbe4dc1ef4d9b6488f11e3bcc15640bd3a71da2" ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_preflight.py": ( "475f3f89043de4aba8f0c2dde041c7425cbfb9c917579f8f546c321016cca833" ), } EXPECTED_G32A_AUTHORITIES = { "AP1/APEIRON_AP1_M1_BALANCED_COUPLED_PRECISION_PREFLIGHT_LATEST.json": ( "13ef13ad1d30d7640c492630a763ae6e7ddbbe4a0359cd9cee77eb70e50894ea" ), "AP1/APEIRON_AP1_M1_BALANCED_COUPLED_PRECISION_PREFLIGHT_CHECKPOINT_LATEST.md": ( "f1c66c68b785501ce70daed8d9fdff3a9e080e4b85ecb28981a8319585e0c2b1" ), "AP1/CODE/ap1_m1_balanced_coupled_precision_preflight.py": ( "aa9ce3b7aea90ed5101dadc40a23d3cc354d38c329c8f5a73135ba0d2384fa0b" ), "AP1/CODE/test_ap1_m1_balanced_coupled_precision_preflight.py": ( "d0412ca930f02506c2eed99a1a4e5bdaaa71939f79950b8a8601494076d50a44" ), } class BalancedCoupledPilotError(g29.CoupledBackgroundError): """An authority, chronology, numerical, or physical gate failed.""" 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_g32_preflight(root: Path) -> dict[str, Any]: observed = { name: file_sha256(root / name) for name in EXPECTED_G32_AUTHORITIES } if observed != EXPECTED_G32_AUTHORITIES: changed = [ name for name, expected in EXPECTED_G32_AUTHORITIES.items() if observed.get(name) != expected ] raise BalancedCoupledPilotError(f"G32 authority drift: {changed}") report = json.loads( ( root / "AP1/APEIRON_AP1_M1_BALANCED_COUPLED_STAGE_PREFLIGHT_LATEST.json" ).read_text(encoding="utf-8") ) if not report.get("all_preflight_gates_pass") or not report.get( "checkpoint_eligible" ): raise BalancedCoupledPilotError("G32 preflight is not PASS") if any( int(report.get(name, -1)) != 0 for name in ("background_runs", "physical_response_kernel_runs") ): raise BalancedCoupledPilotError("G32 preflight contains an unexpected run") plan = g32.BalancedCoupledStagePlan(**report["plan"]) plan.validate() g32.verify_authorities(root) return report def verify_g32a_preflight(root: Path) -> dict[str, Any]: observed = { name: file_sha256(root / name) for name in EXPECTED_G32A_AUTHORITIES } if observed != EXPECTED_G32A_AUTHORITIES: changed = [ name for name, expected in EXPECTED_G32A_AUTHORITIES.items() if observed.get(name) != expected ] raise BalancedCoupledPilotError(f"G32A authority drift: {changed}") report = json.loads( ( root / "AP1/APEIRON_AP1_M1_BALANCED_COUPLED_PRECISION_PREFLIGHT_LATEST.json" ).read_text(encoding="utf-8") ) if not report.get("all_preflight_gates_pass") or not report.get( "pilot_retry_eligible" ): raise BalancedCoupledPilotError("G32A precision preflight is not PASS") if any( int(report.get(name, -1)) != 0 for name in ("background_runs", "physical_response_kernel_runs") ): raise BalancedCoupledPilotError( "G32A precision preflight contains an unexpected run" ) g32a.verify_g32_authorities(root) return report def _replacement_source( run: dict[str, Any], resolved: dict[str, Any] | None = None, tail: dict[str, Any] | None = None, ) -> dict[str, np.ndarray]: reference = run["quantum_fresh"] internal = run["source_internal"] use_resolved = internal["resolved"] if resolved is None else resolved use_tail = internal["tail"] if tail is None else tail answer: dict[str, np.ndarray] = {} for key in g29.SOURCE_KEYS: sector = ( use_resolved["sector_total"][key] + use_tail["sector_finite"][key] ) counterterm = ( internal["extra_signed"][key] - internal["tail"]["asymptotic_signed"][key] ) answer[key] = np.asarray( g29._signed_sector(sector) + use_tail["asymptotic_signed"][key] + counterterm, dtype=float, ) answer[key][0] = reference[key][0] return answer def balanced_operator_convergence_metrics( run: dict[str, Any], arrays: dict[str, np.ndarray], plan: g32.BalancedCoupledStagePlan, evaluation_start: float, ) -> dict[str, Any]: """Retain G29 operator variations while replacing only its Riccati chart.""" trajectory = run["trajectory"] support_nodes = int(run["config"]["nodes"]) mask = g29._evaluation_mask(trajectory.N, evaluation_start) frozen = g29.CoupledConfig() reference = run["quantum_fresh"] internal = run["source_internal"] entry_coarse = g31.transport_resolved_modes_balanced( trajectory, arrays, frozen.coarse_entry_nodes_per_panel, frozen.coarse_mode_step_N, qcut_over_Lambda=plan.qcut_over_Lambda, adiabatic_support_nodes=support_nodes, adiabatic_momentum_support_nodes=plan.adiabatic_momentum_support_nodes, ) mode_step_coarse = g31.transport_resolved_modes_balanced( trajectory, arrays, plan.entry_nodes_per_panel, frozen.coarse_mode_step_N, qcut_over_Lambda=plan.qcut_over_Lambda, adiabatic_support_nodes=support_nodes, adiabatic_momentum_support_nodes=plan.adiabatic_momentum_support_nodes, ) tail_coarse = g29.moving_adiabatic_tail( trajectory, frozen.coarse_tail_nodes_per_octave, plan.K_over_Lambda, adiabatic_support_nodes=support_nodes, adiabatic_momentum_support_nodes=plan.adiabatic_momentum_support_nodes, ) uv_coarse = g29.moving_adiabatic_tail( trajectory, plan.tail_nodes_per_octave, frozen.coarse_K_over_Lambda, adiabatic_support_nodes=support_nodes, adiabatic_momentum_support_nodes=plan.adiabatic_momentum_support_nodes, ) gross = { key: g29.source_gross_scale(internal, key) for key in g29.SOURCE_KEYS } entry_source = _replacement_source(run, resolved=entry_coarse) mode_step_source = _replacement_source(run, resolved=mode_step_coarse) tail_source = _replacement_source(run, tail=tail_coarse) uv_source = _replacement_source(run, tail=uv_coarse) entry_by, entry_max = g29._replacement_metric( reference, entry_source, gross, mask ) mode_by, mode_max = g29._replacement_metric( reference, mode_step_source, gross, mask ) tail_by, tail_max = g29._replacement_metric( reference, tail_source, gross, mask ) uv_by, uv_max = g29._replacement_metric( reference, uv_source, gross, mask ) return { "entry_quadrature_gross_relative_delta_by_source": entry_by, "balanced_mode_step_gross_relative_delta_by_source": mode_by, "tail_quadrature_gross_relative_delta_by_source": tail_by, "uv_cutoff_gross_relative_delta_by_source": uv_by, "max_entry_quadrature_gross_relative_delta": entry_max, "max_balanced_mode_step_gross_relative_delta": mode_max, "max_tail_quadrature_gross_relative_delta": tail_max, "max_uv_cutoff_gross_relative_delta": uv_max, "coarse_entry_diagnostics": entry_coarse["diagnostics"], "coarse_mode_step_diagnostics": mode_step_coarse["diagnostics"], "coarse_tail_diagnostics": tail_coarse["diagnostics"], "coarse_uv_diagnostics": uv_coarse["diagnostics"], } def _pair_items(metrics: dict[str, Any]) -> dict[str, float]: values = { f"background.{name}": float(value) for name, value in metrics["background_relative_delta_by_field"].items() } values.update( { f"source.{name}": float(value) for name, value in metrics[ "source_gross_relative_delta_by_source" ].items() } ) values["background.maximum"] = float( metrics["max_background_time_relative_delta"] ) values["source.maximum"] = float( metrics["max_source_time_gross_relative_delta"] ) return values def three_level_convergence_audit( coarse_to_middle: dict[str, Any], middle_to_fine: dict[str, Any], minimum_order: float, ) -> dict[str, Any]: first = _pair_items(coarse_to_middle) second = _pair_items(middle_to_fine) if set(first) != set(second): raise BalancedCoupledPilotError("three-level metric keys differ") by_metric: dict[str, Any] = {} finite_orders: list[float] = [] finite_orders_above_floor: list[float] = [] for name in sorted(first): item = g32a.assess_richardson_pair( name, first[name], second[name], minimum_order ) order = item["observed_Richardson_order"] if order is not None: finite_orders.append(order) if not item["both_deltas_below_resolution_floor"]: finite_orders_above_floor.append(order) by_metric[name] = item return { "minimum_required_Richardson_order": float(minimum_order), "precision_contract": "G32A inherited-floor-aware-v1.0", "by_metric": by_metric, "all_component_deltas_monotone": bool( all(item["monotone"] for item in by_metric.values()) ), "all_component_orders_pass": bool( all(item["minimum_order_pass"] for item in by_metric.values()) ), "minimum_finite_observed_Richardson_order": ( min(finite_orders) if finite_orders else None ), "minimum_finite_observed_Richardson_order_above_floor": ( min(finite_orders_above_floor) if finite_orders_above_floor else None ), "components_below_inherited_resolution_floor": sum( bool(item["both_deltas_below_resolution_floor"]) for item in by_metric.values() ), } def _grid_value(run: dict[str, Any], n_value: float, field: str) -> float: trajectory = run["trajectory"] grid = np.asarray(trajectory.N, dtype=float) index = int(np.argmin(np.abs(grid - n_value))) if abs(float(grid[index]) - n_value) > 1.0e-14: raise BalancedCoupledPilotError("overlap point is not on every nested grid") return float(np.asarray(getattr(trajectory, field))[index]) def _level_summary(run: dict[str, Any]) -> dict[str, Any]: diagnostics = run["source_diagnostics"]["resolved"] return { "config": dict(run["config"]), "iteration_log": list(run["iteration_log"]), "validation_source_gross_relative_change": float( run["validation_source_gross_relative_change"] ), "H_endpoint_Mpl": float(run["trajectory"].H[-1]), "resolved_diagnostics": dict(diagnostics), } def evaluate_three_level_phase( arrays: dict[str, np.ndarray], plan: g32.BalancedCoupledStagePlan, phase: str, overlap_reference: dict[str, float], progress: Callable[[str], None] | None = None, ) -> dict[str, Any]: plan.validate() nodes = plan._level_nodes(phase) span = plan.pilot_span_N if phase == "pilot" else plan.heldout_span_N evaluation_start = ( plan.pilot_evaluation_start_N if phase == "pilot" else plan.heldout_evaluation_start_N ) runs: dict[str, dict[str, Any]] = {} for label, count in zip(("coarse", "middle", "fine"), nodes): if progress is not None: progress(f"{phase}:{label}:start nodes={count}") runs[label] = g32.run_balanced_coupled(arrays, span, count, plan) if progress is not None: progress( f"{phase}:{label}:done closure=" f"{runs[label]['validation_source_gross_relative_change']:.17g}" ) coarse_to_middle = g29.time_convergence_metrics( runs["middle"], runs["coarse"], evaluation_start ) middle_to_fine = g29.time_convergence_metrics( runs["fine"], runs["middle"], evaluation_start ) convergence = three_level_convergence_audit( coarse_to_middle, middle_to_fine, float(g32.PREDECLARED_POLICY["minimum_observed_Richardson_order"]), ) if progress is not None: progress(f"{phase}:operator_checks:start") physical = g29.physical_diagnostics(runs["fine"], arrays, evaluation_start) operator = balanced_operator_convergence_metrics( runs["fine"], arrays, plan, evaluation_start ) precision = g29.decimal_precision_metrics( runs["fine"], evaluation_start, plan.decimal_low_digits, plan.decimal_high_digits, ) if progress is not None: progress(f"{phase}:operator_checks:done") overlap_N = float(overlap_reference["N_relative"]) overlap_expected = float(overlap_reference["H_Mpl"]) overlap_values = { label: _grid_value(run, overlap_N, "H") for label, run in runs.items() } overlap_by_level = { label: abs(value - overlap_expected) / max(abs(value), abs(overlap_expected), 1.0e-300) for label, value in overlap_values.items() } fine = runs["fine"] metrics = { "max_fixed_point_source_gross_relative_change": float( fine["validation_source_gross_relative_change"] ), "max_background_time_relative_delta": float( middle_to_fine["max_background_time_relative_delta"] ), "max_source_time_gross_relative_delta": float( middle_to_fine["max_source_time_gross_relative_delta"] ), "max_entry_quadrature_gross_relative_delta": float( operator["max_entry_quadrature_gross_relative_delta"] ), "max_tail_quadrature_gross_relative_delta": float( operator["max_tail_quadrature_gross_relative_delta"] ), "max_uv_cutoff_gross_relative_delta": float( operator["max_uv_cutoff_gross_relative_delta"] ), "max_friedmann_normalized": float( physical["max_friedmann_normalized"] ), "max_raychaudhuri_normalized": float( physical["max_raychaudhuri_normalized"] ), "ward_normalized": float(physical["ward_normalized"]), "standard_continuity_normalized": float( physical["standard_continuity_normalized"] ), "max_stage_overlap_H_relative_delta": float(overlap_by_level["fine"]), } return { "phase": phase, "gate_interval_N": [float(evaluation_start), float(span)], "level_nodes": list(nodes), "same_operator_settings_on_all_levels": True, "levels": {name: _level_summary(run) for name, run in runs.items()}, "physical_diagnostics": physical, "time_convergence": { "coarse_to_middle": coarse_to_middle, "middle_to_fine": middle_to_fine, "three_level_audit": convergence, }, "operator_convergence": operator, "precision": precision, "source_diagnostics": fine["source_diagnostics"], "overlap_audit": { "reference_N_relative": overlap_N, "reference_H_Mpl": overlap_expected, "observed_H_Mpl_by_level": overlap_values, "relative_delta_by_level": overlap_by_level, }, "stage_endpoint_aggregate_not_a_seed": { "N_relative": float(span), "physical_N": float(physical["physical_N_endpoint"]), "H_Mpl": float(physical["H_endpoint_Mpl"]), "delta_H_over_H": float(physical["delta_H_over_H"]), }, "metrics_for_freeze_or_gate": metrics, "trajectory_rows_persisted": 0, "new_AP1_M1_background_resolution_runs": 3, "physical_response_kernel_runs": 0, } def _phase_gates( ensemble: dict[str, Any], plan: g32.BalancedCoupledStagePlan, phase: str, ) -> dict[str, bool]: absolute = g29._absolute_gates(ensemble) level_diagnostics = [ ensemble["levels"][label]["resolved_diagnostics"] for label in ("coarse", "middle", "fine") ] operator = ensemble["operator_convergence"] chart_diagnostics = level_diagnostics + [ operator["coarse_entry_diagnostics"], operator["coarse_mode_step_diagnostics"], ] budget = ( plan.max_pilot_resolved_modes if phase == "pilot" else plan.max_heldout_resolved_modes ) audit = ensemble["time_convergence"]["three_level_audit"] return { **absolute, "all_balanced_transports_use_exact_G31_method_without_vacuum_reset": bool( g32a.exact_transport_used(chart_diagnostics) and all(item["vacuum_resets"] == 0 for item in chart_diagnostics) and ensemble["source_diagnostics"].get("balanced_chart_used", False) ), "all_active_ratios_in_dominant_chart": bool( max(float(item["max_abs_active_ratio"]) for item in chart_diagnostics) <= float(g31.REFERENCE_POLICY["active_ratio_cap"]) ), "all_wronskians_below_existing_limit": bool( max( max( float(item["max_wronskian_relative_error"]), float(item["max_initial_handoff_wronskian_relative_error"]), ) for item in chart_diagnostics ) <= float(g31.REFERENCE_POLICY["wronskian_relative_cap"]) ), "all_preprojection_transfer_defects_below_G31_cap": bool( max( float(item["max_transfer_determinant_defect_before_projection"]) for item in chart_diagnostics ) <= float(g31.REFERENCE_POLICY["transfer_determinant_defect_cap"]) ), "balanced_mode_step_delta_below_G31_cap": bool( float(operator["max_balanced_mode_step_gross_relative_delta"]) <= float(g31.REFERENCE_POLICY["mode_step_source_gross_relative_cap"]) ), "three_level_component_deltas_monotone": bool( audit["all_component_deltas_monotone"] ), "three_level_component_Richardson_orders_pass": bool( audit["all_component_orders_pass"] ), "resolved_mode_budget_bounded": bool( max(int(item["final_resolved_modes"]) for item in level_diagnostics) <= int(budget) ), "same_operator_settings_on_all_three_levels": bool( ensemble["same_operator_settings_on_all_levels"] ), "no_trajectory_seed_kernel_curve_fit_or_significance": bool( ensemble["trajectory_rows_persisted"] == 0 and ensemble["physical_response_kernel_runs"] == 0 ), } def _ceiling_gates(ensemble: dict[str, Any]) -> dict[str, bool]: policy = g32.PREDECLARED_POLICY["inherited_metric_freeze_policy"] metrics = ensemble["metrics_for_freeze_or_gate"] if set(metrics) != set(policy): raise BalancedCoupledPilotError("pilot metric set differs from G32 policy") return { f"{name}_below_predeclared_ceiling": bool( math.isfinite(float(value)) and float(value) <= float(policy[name]["ceiling"]) ) for name, value in metrics.items() } def _load_arrays(root: Path) -> dict[str, np.ndarray]: _, arrays, _ = g29.load_inputs(root) return arrays def build_pilot( root: Path, code_path: Path, test_path: Path, progress: Callable[[str], None] | None = None, ) -> dict[str, Any]: preflight = verify_g32_preflight(root) precision_preflight = verify_g32a_preflight(root) plan = g32.BalancedCoupledStagePlan(**preflight["plan"]) plan.validate() g30_checkpoint, _ = g32.verify_authorities(root) arrays = _load_arrays(root) overlap = { "N_relative": plan.baseline_span_N, "H_Mpl": float( g30_checkpoint["heldout_physical_diagnostics"]["H_endpoint_Mpl"] ), } ensemble = evaluate_three_level_phase( arrays, plan, "pilot", overlap, progress=progress ) gates = { "G32_preflight_hash_bound_and_PASS": True, "G32A_precision_preflight_hash_bound_and_PASS": bool( precision_preflight["all_preflight_gates_pass"] ), "known_N_le_0p08_region_used_as_pilot_only": bool( plan.pilot_span_N == plan.prior_diagnostic_endpoint_N ), "future_heldout_strictly_disjoint_and_unseen": bool( plan.heldout_evaluation_start_N > plan.pilot_span_N ), **_phase_gates(ensemble, plan, "pilot"), **_ceiling_gates(ensemble), "production_tolerances_present_anchor_and_response_kernel_locked": True, } passed = bool(all(gates.values())) return { "schema": "apeiron-ap1-m1-balanced-coupled-three-level-pilot-v1.1", "updated_utc": _utc_now(), "classification": ( "M1_BALANCED_COUPLED_THREE_LEVEL_PILOT_PASS_TOLERANCE_FREEZE_OPEN_PHYSICAL_KERNEL_BLOCKED" if passed else "M1_BALANCED_COUPLED_THREE_LEVEL_PILOT_NONPASS" ), "authority_sha256": { **EXPECTED_G32_AUTHORITIES, **EXPECTED_G32A_AUTHORITIES, "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( code_path ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( test_path ), }, "plan": asdict(plan), "predeclared_policy": g32.PREDECLARED_POLICY, "precision_predeclared_policy": precision_preflight["precision_policy"], "pilot_ensemble": ensemble, "gates": gates, "all_pilot_gates_pass": passed, "heldout_evaluated": False, "checkpoint_eligible": False, "method_tolerances_frozen": False, "production_background_tolerances_frozen": False, "trajectory_rows_persisted": 0, "new_AP1_M1_background_resolution_runs": 3, "physical_response_kernel_runs": 0, "retarded_response_applicable": False, "memory_convergence_applicable": 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, "AP1_status": "ORANGE", "claim_boundary": ( "known-domain three-level numerical pilot only; no heldout, " "production background, present anchor, response kernel, memory " "kernel, observable, curve, fit or significance" ), "next_required": ( "freeze the pilot-derived method tolerances before evaluating the " "predeclared disjoint N=0.085..0.10 heldout once" if passed else "stop fail-closed; do not persist, checkpoint, freeze or seed" ), } def _thresholds_from_metrics(metrics: dict[str, Any]) -> dict[str, float]: policy = g32.PREDECLARED_POLICY["inherited_metric_freeze_policy"] if set(metrics) != set(policy): raise BalancedCoupledPilotError("freeze metric set differs from G32 policy") return { name: max( float(specification["floor"]), float(specification["multiplier"]) * float(metrics[name]), ) for name, specification in policy.items() } def _verify_pilot_code_hashes( pilot: dict[str, Any], code_path: Path, test_path: Path, ) -> None: expected = { "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( code_path ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( test_path ), } for name, digest in expected.items(): if pilot["authority_sha256"].get(name) != digest: raise BalancedCoupledPilotError(f"pilot authority changed: {name}") def build_freeze( pilot_path: Path, code_path: Path, test_path: Path, ) -> dict[str, Any]: pilot = json.loads(pilot_path.read_text(encoding="utf-8")) if not pilot.get("all_pilot_gates_pass") or pilot.get("heldout_evaluated"): raise BalancedCoupledPilotError("complete pre-heldout pilot PASS required") _verify_pilot_code_hashes(pilot, code_path, test_path) plan = g32.BalancedCoupledStagePlan(**pilot["plan"]) plan.validate() metrics = pilot["pilot_ensemble"]["metrics_for_freeze_or_gate"] thresholds = _thresholds_from_metrics(metrics) policy = g32.PREDECLARED_POLICY["inherited_metric_freeze_policy"] gates = { "pilot_hash_bound_before_heldout": True, **{ f"{name}_threshold_below_predeclared_ceiling": bool( math.isfinite(thresholds[name]) and thresholds[name] <= float(specification["ceiling"]) ) for name, specification in policy.items() }, "heldout_interval_registered_disjoint_and_unseen": bool( plan.heldout_evaluation_start_N > plan.prior_diagnostic_endpoint_N ), "production_tolerances_present_anchor_and_response_kernel_locked": True, } passed = bool(all(gates.values())) endpoint = pilot["pilot_ensemble"]["stage_endpoint_aggregate_not_a_seed"] return { "schema": "apeiron-ap1-m1-balanced-coupled-three-level-freeze-v1.0", "updated_utc": _utc_now(), "classification": ( "M1_BALANCED_COUPLED_METHOD_TOLERANCES_FROZEN_HELDOUT_UNSEEN" if passed else "M1_BALANCED_COUPLED_METHOD_TOLERANCE_FREEZE_NONPASS" ), "input_sha256": { pilot_path.name: file_sha256(pilot_path), "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( code_path ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( test_path ), }, "predeclared_policy": g32.PREDECLARED_POLICY, "thresholds": thresholds, "heldout_plan": { "span_N": plan.heldout_span_N, "evaluation_start_N": plan.heldout_evaluation_start_N, "nodes": list(plan._level_nodes("heldout")), "overlap_reference": { "N_relative": float(endpoint["N_relative"]), "H_Mpl": float(endpoint["H_Mpl"]), }, }, "gates": gates, "all_tolerance_freeze_gates_pass": passed, "tolerances_frozen_before_heldout": passed, "heldout_evaluated": False, "method_tolerances_frozen": passed, "production_background_tolerances_frozen": False, "trajectory_rows_persisted": 0, "physical_response_kernel_runs": 0, "seed_released": False, "nonpass_stored_or_used_as_seed": False, "AP1_status": "ORANGE", "claim_boundary": ( "pilot-derived three-level method tolerances only; heldout, " "production background, present anchor and response kernel unseen" ), "next_required": ( "evaluate the registered disjoint N=0.085..0.10 heldout once and " "persist only a full PASS" if passed else "stop fail-closed; do not evaluate the heldout" ), } def build_heldout( root: Path, pilot_path: Path, freeze_path: Path, code_path: Path, test_path: Path, progress: Callable[[str], None] | None = None, ) -> dict[str, Any]: pilot = json.loads(pilot_path.read_text(encoding="utf-8")) freeze = json.loads(freeze_path.read_text(encoding="utf-8")) if not pilot.get("all_pilot_gates_pass"): raise BalancedCoupledPilotError("pilot is not PASS") if not freeze.get("all_tolerance_freeze_gates_pass"): raise BalancedCoupledPilotError("tolerance freeze is not PASS") if freeze["input_sha256"].get(pilot_path.name) != file_sha256(pilot_path): raise BalancedCoupledPilotError("pilot changed after tolerance freeze") _verify_pilot_code_hashes(pilot, code_path, test_path) for name, path in ( ("AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py", code_path), ("AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py", test_path), ): if freeze["input_sha256"].get(name) != file_sha256(path): raise BalancedCoupledPilotError(f"code changed after freeze: {name}") preflight = verify_g32_preflight(root) verify_g32a_preflight(root) plan = g32.BalancedCoupledStagePlan(**pilot["plan"]) if asdict(plan) != preflight["plan"]: raise BalancedCoupledPilotError("pilot plan differs from G32 preflight") plan.validate() heldout_plan = freeze["heldout_plan"] if not ( heldout_plan["span_N"] == plan.heldout_span_N and heldout_plan["evaluation_start_N"] == plan.heldout_evaluation_start_N and heldout_plan["nodes"] == list(plan._level_nodes("heldout")) ): raise BalancedCoupledPilotError("heldout plan changed after freeze") arrays = _load_arrays(root) ensemble = evaluate_three_level_phase( arrays, plan, "heldout", heldout_plan["overlap_reference"], progress=progress, ) metrics = ensemble["metrics_for_freeze_or_gate"] numerical = { f"{name}_below_frozen_threshold": bool( math.isfinite(float(value)) and float(value) <= float(freeze["thresholds"][name]) ) for name, value in metrics.items() } gates = { "pilot_and_freeze_hash_bound_before_heldout": True, "registered_disjoint_heldout_used_once": True, **_phase_gates(ensemble, plan, "heldout"), **numerical, "production_tolerances_present_anchor_and_response_kernel_locked": True, } passed = bool(all(gates.values())) return { "schema": "apeiron-ap1-m1-balanced-coupled-three-level-heldout-v1.0", "updated_utc": _utc_now(), "classification": ( "M1_BALANCED_COUPLED_THREE_LEVEL_HELDOUT_PASS_CHECKPOINT_OPEN_PHYSICAL_KERNEL_BLOCKED" if passed else "M1_BALANCED_COUPLED_THREE_LEVEL_HELDOUT_NONPASS" ), "authority_sha256": { **EXPECTED_G32_AUTHORITIES, **EXPECTED_G32A_AUTHORITIES, pilot_path.name: file_sha256(pilot_path), freeze_path.name: file_sha256(freeze_path), "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( code_path ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( test_path ), }, "frozen_thresholds": freeze["thresholds"], "heldout_was_unseen_when_tolerances_frozen": True, "heldout_ensemble": ensemble, "gates": gates, "all_heldout_gates_pass": passed, "checkpoint_eligible": passed, "new_AP1_M1_background_resolution_runs": 6, "method_tolerances_frozen": True, "production_background_tolerances_frozen": False, "trajectory_rows_persisted": 0, "physical_response_kernel_runs": 0, "retarded_response_applicable": False, "memory_convergence_applicable": 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, "AP1_status": "ORANGE", "claim_boundary": ( "three-level coupled prefix through relative N=0.10 only; no " "production background, present anchor, response kernel, memory " "kernel, observable, curve, fit or significance" ), "next_required": ( "promote the bounded G33 checkpoint and predeclare the next " "logarithmic continuation" if passed else "stop fail-closed; do not persist, checkpoint or seed" ), } def build_checkpoint( pilot_path: Path, freeze_path: Path, heldout_path: Path, code_path: Path, test_path: Path, ) -> dict[str, Any]: pilot = json.loads(pilot_path.read_text(encoding="utf-8")) freeze = json.loads(freeze_path.read_text(encoding="utf-8")) heldout = json.loads(heldout_path.read_text(encoding="utf-8")) if not ( pilot.get("all_pilot_gates_pass") and freeze.get("all_tolerance_freeze_gates_pass") and heldout.get("all_heldout_gates_pass") ): raise BalancedCoupledPilotError("full pilot/freeze/heldout PASS required") expected = { pilot_path.name: file_sha256(pilot_path), freeze_path.name: file_sha256(freeze_path), "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( code_path ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( test_path ), } for name, digest in expected.items(): if heldout["authority_sha256"].get(name) != digest: raise BalancedCoupledPilotError(f"heldout authority mismatch: {name}") ensemble = heldout["heldout_ensemble"] return { "schema": "apeiron-ap1-m1-balanced-coupled-three-level-checkpoint-v1.0", "updated_utc": _utc_now(), "classification": "M1_BALANCED_COUPLED_THREE_LEVEL_HELDOUT_PASS_NEXT_LOG_STAGE_OPEN_PHYSICAL_KERNEL_BLOCKED", "authority_sha256": { pilot_path.name: file_sha256(pilot_path), freeze_path.name: file_sha256(freeze_path), heldout_path.name: file_sha256(heldout_path), "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( code_path ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": file_sha256( test_path ), }, "frozen_method_thresholds": freeze["thresholds"], "heldout_metrics": ensemble["metrics_for_freeze_or_gate"], "heldout_three_level_convergence": ensemble["time_convergence"], "heldout_physical_diagnostics": ensemble["physical_diagnostics"], "heldout_source_diagnostics": ensemble["source_diagnostics"], "heldout_operator_convergence": ensemble["operator_convergence"], "heldout_precision": ensemble["precision"], "heldout_overlap_audit": ensemble["overlap_audit"], "gates": { "pilot_complete": True, "method_tolerances_frozen_before_heldout": True, "heldout_full_PASS": True, "only_PASS_promoted": True, "no_production_background_kernel_seed_or_observable": True, }, "all_checkpoint_gates_pass": True, "checkpoint_eligible": True, "new_AP1_M1_background_resolution_runs": 6, "validated_relative_N_endpoint": 0.10, "extension_factor_over_G30": 2.5, "method_tolerances_frozen": True, "production_background_tolerances_frozen": False, "trajectory_rows_persisted": 0, "physical_response_kernel_runs": 0, "seed_released": 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": ( "bounded three-level AP1-M1 background prefix through relative " "N=0.10 only; no production trajectory, present anchor, response " "kernel, observable, curve, fit or significance" ), "next_required": ( "predeclare the next logarithmic continuation; production " "tolerances, present anchor and physical response kernel remain locked" ), } def checkpoint_markdown(report: dict[str, Any]) -> str: metrics = report["heldout_metrics"] physical = report["heldout_physical_diagnostics"] source = report["heldout_source_diagnostics"] audit = report["heldout_three_level_convergence"]["three_level_audit"] operator = report["heldout_operator_convergence"] return f"""# Apeiron AP1 – G33 balanced coupled three-level checkpoint **Updated UTC:** {report['updated_utc']} **Classification:** `{report['classification']}` **AP1 status:** ORANGE The G32-bound pilot used three exactly nested total-coupled grids through relative N=0.08. Only after its complete PASS were method-specific tolerances frozen. The disjoint N=0.085..0.10 heldout was then evaluated once on three nested grids. Equations, physics, parameters, PV sectors, moving split, counterterms, and all pre-existing gate thresholds remained unchanged. ## Heldout aggregate - validated relative endpoint: `{report['validated_relative_N_endpoint']}` - fixed-point source gross-relative closure: `{metrics['max_fixed_point_source_gross_relative_change']:.17g}` - middle-to-fine background delta: `{metrics['max_background_time_relative_delta']:.17g}` - middle-to-fine source gross-relative delta: `{metrics['max_source_time_gross_relative_delta']:.17g}` - minimum finite Richardson order above inherited floors: `{audit['minimum_finite_observed_Richardson_order_above_floor']}` - components classified below inherited resolution floors: `{audit['components_below_inherited_resolution_floor']}` - entry quadrature gross-relative delta: `{metrics['max_entry_quadrature_gross_relative_delta']:.17g}` - UV-tail quadrature gross-relative delta: `{metrics['max_tail_quadrature_gross_relative_delta']:.17g}` - UV-cutoff gross-relative delta: `{metrics['max_uv_cutoff_gross_relative_delta']:.17g}` - balanced mode-step gross-relative delta: `{operator['max_balanced_mode_step_gross_relative_delta']:.17g}` - Friedmann residual: `{physical['max_friedmann_normalized']:.17g}` - Raychaudhuri residual: `{physical['max_raychaudhuri_normalized']:.17g}` - flux-balanced Ward residual: `{physical['ward_normalized']:.17g}` - standard continuity residual: `{physical['standard_continuity_normalized']:.17g}` - Wronskian error: `{source['resolved']['max_wronskian_relative_error']:.17g}` All frozen heldout gates pass. Six background-resolution runs were performed across pilot and heldout. Zero trajectory rows and zero seeds were released. ## Boundary This checkpoint validates only the bounded background prefix through relative N=0.10. It is not a production background and does not reach the present anchor. Production tolerances, AP2/AP3, production curves, observables, and the physical two-time response kernel remain locked. No fit or significance is claimed. """ def _write_pass_json(report: dict[str, Any], output: Path, pass_key: str) -> None: if not report.get(pass_key): raise BalancedCoupledPilotError("NONPASS report was not written") serialized = json.dumps(report, indent=2, allow_nan=False) + "\n" output.parent.mkdir(parents=True, exist_ok=True) with output.open("x", encoding="utf-8") as stream: stream.write(serialized) def _nonpass_summary(report: dict[str, Any]) -> dict[str, Any]: gates = report.get("gates", {}) ensemble = report.get("pilot_ensemble") or report.get("heldout_ensemble") or {} return { "classification": report.get("classification"), "failed_gates": [name for name, passed in gates.items() if not passed], "metrics": ensemble.get("metrics_for_freeze_or_gate"), "three_level_audit": ensemble.get("time_convergence", {}).get( "three_level_audit" ), "output_written": False, } def _write_or_fail( report: dict[str, Any], output: Path, pass_key: str, ) -> None: if not report.get(pass_key): print(json.dumps(_nonpass_summary(report), indent=2, allow_nan=False)) raise SystemExit(2) _write_pass_json(report, output, pass_key) def main() -> None: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) pilot_parser = sub.add_parser("pilot") pilot_parser.add_argument("apeiron_root", type=Path) pilot_parser.add_argument("--test-path", type=Path, required=True) pilot_parser.add_argument("--output", type=Path, required=True) freeze_parser = sub.add_parser("freeze") freeze_parser.add_argument("pilot", type=Path) freeze_parser.add_argument("--test-path", type=Path, required=True) freeze_parser.add_argument("--output", type=Path, required=True) heldout_parser = sub.add_parser("heldout") heldout_parser.add_argument("apeiron_root", type=Path) heldout_parser.add_argument("pilot", type=Path) heldout_parser.add_argument("freeze", type=Path) heldout_parser.add_argument("--test-path", type=Path, required=True) heldout_parser.add_argument("--output", type=Path, required=True) checkpoint_parser = sub.add_parser("checkpoint") checkpoint_parser.add_argument("pilot", type=Path) checkpoint_parser.add_argument("freeze", type=Path) checkpoint_parser.add_argument("heldout", type=Path) checkpoint_parser.add_argument("--test-path", type=Path, required=True) checkpoint_parser.add_argument("--json-output", type=Path, required=True) checkpoint_parser.add_argument("--md-output", type=Path, required=True) args = parser.parse_args() code_path = Path(__file__).resolve() progress = lambda message: print(message, flush=True) if args.command == "pilot": report = build_pilot( args.apeiron_root, code_path, args.test_path, progress=progress ) _write_or_fail(report, args.output, "all_pilot_gates_pass") elif args.command == "freeze": report = build_freeze(args.pilot, code_path, args.test_path) _write_or_fail( report, args.output, "all_tolerance_freeze_gates_pass" ) elif args.command == "heldout": report = build_heldout( args.apeiron_root, args.pilot, args.freeze, code_path, args.test_path, progress=progress, ) _write_or_fail(report, args.output, "all_heldout_gates_pass") else: report = build_checkpoint( args.pilot, args.freeze, args.heldout, code_path, args.test_path ) _write_pass_json(report, args.json_output, "all_checkpoint_gates_pass") with args.md_output.open("x", encoding="utf-8") as stream: stream.write(checkpoint_markdown(report)) if __name__ == "__main__": main()