from __future__ import annotations import json from pathlib import Path import tempfile import unittest import numpy as np from ap1_m1_uv_tail_handoff_reference import ( FREEZE_POLICY, HandoffConfig, UVHandoffError, _pv_moments_exact, adiabatic_boundary_primitives, build_freeze, build_pilot, decimal_direct_flux, decimal_expanded_tail_flux, direct_resolved_flux, expanded_tail_flux, handoff_state_metrics, ) from chi_background_closure import ChiParameters, ChiTrajectory class TestUVTailHandoffReference(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.root = Path(__file__).resolve().parents[2] cls.code = Path(__file__).resolve().with_name( "ap1_m1_uv_tail_handoff_reference.py" ) cls.pilot = build_pilot(cls.root, cls.code) @staticmethod def analytic_trajectory(nodes: int = 401) -> ChiTrajectory: N = np.linspace(-1.0, -0.95, nodes) H = 1.0e-4 * np.exp(-0.08 * (N - N[0])) return ChiTrajectory( N=N, H=H, Hdot=-0.08 * H**2, sigma=0.02 + 5.0e-4 * (N - N[0]), theta=-0.69 + 8.0e-4 * (N - N[0]), ).validated() def test_config_is_chronology_disjoint_and_frozen_qcut(self) -> None: HandoffConfig().validate() with self.assertRaises(ValueError): HandoffConfig(qcut_over_Lambda=0.59).validate() with self.assertRaises(ValueError): HandoffConfig( heldout_sample_relative_N=(0.0125, 0.08, 0.10) ).validate() def test_decimal_pv_moments_zero_through_order_two(self) -> None: self.assertEqual(_pv_moments_exact(), {"0": "0", "1": "0", "2": "0"}) def test_boundary_primitives_are_positive_and_hierarchical(self) -> None: tr = self.analytic_trajectory() p = ChiParameters() q = p.Lambda * np.array([0.595, 0.6, 0.605]) primitive = adiabatic_boundary_primitives(tr, 0.025, q, p) omega = np.asarray(primitive["omega"]) W2 = np.asarray(primitive["W2"]) W4 = np.asarray(primitive["W4"]) W = np.asarray(primitive["W"]) self.assertTrue(np.all(omega > 0.0)) self.assertTrue(np.all(W > 0.0)) self.assertLess(np.max(np.abs(W4 / omega)), np.max(np.abs(W2 / omega))) def test_direct_and_expanded_fluxes_are_independent_and_finite(self) -> None: tr = self.analytic_trajectory() p = ChiParameters() primitive = adiabatic_boundary_primitives( tr, 0.025, np.array([0.6 * p.Lambda]), p ) direct = direct_resolved_flux(primitive) expanded = expanded_tail_flux(primitive) for key in ("rho", "pressure", "chi2"): self.assertTrue(np.all(np.isfinite(direct[key]))) self.assertTrue(np.all(np.isfinite(expanded[key]))) scale = max(float(np.max(np.abs(direct[key]))), 1.0e-300) self.assertLess(float(np.max(np.abs(direct[key] - expanded[key]))) / scale, 1e-3) def test_decimal_signed_sum_is_precision_stable(self) -> None: tr = self.analytic_trajectory() p = ChiParameters() primitive = adiabatic_boundary_primitives( tr, 0.025, np.array([0.6 * p.Lambda]), p ) low = decimal_direct_flux(primitive, 0, 60) high = decimal_direct_flux(primitive, 0, 80) tail_low = decimal_expanded_tail_flux(primitive, 0, 60) tail_high = decimal_expanded_tail_flux(primitive, 0, 80) for key in ("rho", "pressure", "chi2"): low_value = float(low[key]["signed_PV_flux"]) high_value = float(high[key]["signed_PV_flux"]) gross = float(high[key]["gross_abs_PV_flux"]) self.assertLessEqual(abs(low_value - high_value) / gross, 1e-15) tail_low_value = float(tail_low[key]["signed_PV_flux"]) tail_high_value = float(tail_high[key]["signed_PV_flux"]) tail_gross = float(tail_high[key]["gross_abs_PV_flux"]) self.assertLessEqual( abs(tail_low_value - tail_high_value) / tail_gross, 1e-15 ) def test_handoff_state_preserves_wronskian_without_vacuum_reset(self) -> None: tr = self.analytic_trajectory() p = ChiParameters() primitive = adiabatic_boundary_primitives( tr, 0.025, np.array([0.6 * p.Lambda]), p ) metrics = handoff_state_metrics(primitive, 0) self.assertFalse(metrics["state_reinitialized_to_instantaneous_order0_vacuum"]) self.assertLess( metrics["max_tail_to_log_chart_state_relative_error"], FREEZE_POLICY["state_transfer_relative_cap"], ) self.assertLess( metrics["max_wronskian_relative_error"], FREEZE_POLICY["wronskian_relative_cap"], ) def test_actual_pilot_is_complete_but_not_checkpoint_eligible(self) -> None: self.assertTrue(self.pilot["all_pilot_gates_pass"]) self.assertFalse(self.pilot["heldout_evaluated"]) self.assertFalse(self.pilot["checkpoint_eligible"]) self.assertEqual(self.pilot["new_AP1_M1_background_runs"], 0) self.assertEqual(self.pilot["trajectory_rows_persisted"], 0) def test_freeze_is_hash_bound_and_does_not_evaluate_heldout(self) -> None: with tempfile.TemporaryDirectory() as directory: pilot_path = Path(directory) / "pilot.json" pilot_path.write_text(json.dumps(self.pilot), encoding="utf-8") freeze = build_freeze(pilot_path, self.code) self.assertTrue(freeze["all_tolerance_freeze_gates_pass"]) self.assertTrue(freeze["tolerances_frozen_before_heldout"]) self.assertFalse(freeze["heldout_evaluated"]) def test_freeze_rejects_incomplete_or_seen_pilot(self) -> None: changed = json.loads(json.dumps(self.pilot)) changed["heldout_evaluated"] = True with tempfile.TemporaryDirectory() as directory: pilot_path = Path(directory) / "pilot.json" pilot_path.write_text(json.dumps(changed), encoding="utf-8") with self.assertRaises(UVHandoffError): build_freeze(pilot_path, self.code) def test_pilot_records_local_memory_scope_and_no_kernel(self) -> None: self.assertFalse(self.pilot["physical_response_kernel_started"]) self.assertFalse(self.pilot["background_tolerances_frozen"]) self.assertFalse(self.pilot["seed_released"]) self.assertIn("no physical background", self.pilot["claim_boundary"]) if __name__ == "__main__": unittest.main()