from __future__ import annotations import json import math from pathlib import Path import tempfile import unittest import numpy as np import ap1_m1_logh_positive_branch_recovery_preflight as recovery ROOT = Path(__file__).resolve().parents[2] CODE = ROOT / "AP1/CODE/ap1_m1_logh_positive_branch_recovery_preflight.py" TEST = Path(__file__).resolve() class LogHPositiveBranchRecoveryPreflightTests(unittest.TestCase): def test_G32B_authority_chain_is_exact_and_preheldout_pass(self) -> None: pilot, freeze = recovery.verify_authorities(ROOT) self.assertTrue(pilot["all_pilot_gates_pass"]) self.assertTrue(freeze["all_tolerance_freeze_gates_pass"]) self.assertFalse(pilot["heldout_evaluated"]) def test_recovery_plan_is_nested_and_chronology_safe(self) -> None: plan = recovery.LogHRecoveryPlan() plan.validate() self.assertEqual(plan.recovery_pilot_span_N, 0.10) self.assertGreater( plan.prospective_heldout_evaluation_start_N, plan.consumed_G32B_span_N, ) for phase in ("recovery_pilot", "prospective_heldout"): coarse, middle, fine = plan._level_nodes(phase) self.assertEqual(fine - 1, 2 * (middle - 1)) self.assertEqual(middle - 1, 2 * (coarse - 1)) def test_unknown_recovery_phase_fails_closed(self) -> None: with self.assertRaises(ValueError): recovery.LogHRecoveryPlan()._level_nodes("unknown") def test_new_heldout_must_be_beyond_consumed_domain(self) -> None: plan = recovery.LogHRecoveryPlan( prospective_heldout_evaluation_start_N=0.10 ) with self.assertRaises(ValueError): plan.validate() def test_G32B_numerical_settings_are_retained(self) -> None: plan = recovery.LogHRecoveryPlan() self.assertEqual(plan.fixed_point_steps, 10) self.assertEqual(plan.relaxation, 0.9) self.assertEqual(plan.qcut_over_Lambda, 0.6) self.assertEqual(plan.max_mode_step_N, 1.25e-5) self.assertEqual(plan.background_rtol, 2.0e-10) def test_H_roundtrip_is_machine_accurate(self) -> None: reference = 4.3e-7 for ratio in (2.0**-20, 0.2, 1.0, 3.0, 2.0**20): state = np.array([0.1, -2.0e-8, 0.2, 3.0e-8, reference * ratio]) transformed = recovery.physical_to_logh_state(state, reference) restored = recovery.logh_to_physical_state(transformed, reference) np.testing.assert_allclose(restored[:4], state[:4], rtol=0.0, atol=0.0) self.assertLessEqual( abs(restored[4] - state[4]) / state[4], 64.0 * np.finfo(float).eps, ) self.assertGreater(restored[4], 0.0) def test_direct_and_logH_rhs_are_algebraically_equivalent(self) -> None: state = np.array([0.2, -3.0e-8, -0.4, 2.0e-8, 4.1e-7]) direct = np.array([1.0e-3, -2.0e-4, 3.0e-3, -4.0e-4, -5.0e-9]) logh = recovery.direct_rhs_to_logh(state, direct) restored = recovery.logh_rhs_to_direct(state, logh) np.testing.assert_allclose(restored, direct, rtol=2.0e-16, atol=0.0) def test_nonpositive_physical_H_fails_closed(self) -> None: for H in (0.0, -1.0): with self.assertRaises(recovery.LogHRecoveryPreflightError): recovery.physical_to_logh_state( np.array([0.0, 0.0, 0.0, 0.0, H]), 1.0 ) def test_nonpositive_reference_fails_closed(self) -> None: with self.assertRaises(recovery.LogHRecoveryPreflightError): recovery.logh_to_physical_state(np.zeros(5), 0.0) def test_nonfinite_state_fails_closed(self) -> None: state = np.array([0.0, 0.0, math.nan, 0.0, 1.0]) with self.assertRaises(recovery.LogHRecoveryPreflightError): recovery.physical_to_logh_state(state, 1.0) def test_chart_canary_passes_strict_json(self) -> None: canary = recovery.algebraic_equivalence_canary() self.assertLessEqual( canary["maximum_H_roundtrip_relative_defect"], canary["relative_defect_cap"], ) self.assertLessEqual( canary["maximum_rhs_reconstruction_relative_defect"], canary["relative_defect_cap"], ) self.assertTrue(canary["all_finite_logH_states_reconstruct_positive_H"]) json.dumps(canary, allow_nan=False) def test_preflight_records_consumed_failed_heldout_without_NONPASS_artifact(self) -> None: report = recovery.build_preflight(ROOT, CODE, TEST) audit = report["failed_G32B_heldout_audit"] self.assertEqual(audit["registered_G32B_heldout_attempts"], 1) self.assertEqual( audit["G32B_heldout_result"], "NONPASS_before_first_level_completed" ) self.assertFalse(audit["G32B_heldout_report_written"]) self.assertFalse(audit["G32B_heldout_checkpoint_or_seed_released"]) self.assertFalse(report["G33_released"]) def test_preflight_passes_with_zero_background_and_kernel_runs(self) -> None: report = recovery.build_preflight(ROOT, CODE, TEST) self.assertTrue(report["all_preflight_gates_pass"]) self.assertTrue(all(report["gates"].values())) self.assertEqual(report["background_runs"], 0) self.assertEqual(report["physical_response_kernel_runs"], 0) self.assertEqual(report["trajectory_rows_persisted"], 0) def test_existing_equations_physics_parameters_and_thresholds_stay_frozen(self) -> None: report = recovery.build_preflight(ROOT, CODE, TEST) self.assertFalse(report["equations_changed"]) self.assertFalse(report["physics_changed"]) self.assertFalse(report["parameters_changed"]) self.assertFalse(report["existing_gate_thresholds_changed"]) self.assertEqual(report["AP1_status"], "ORANGE") def test_prospective_heldout_remains_locked(self) -> None: report = recovery.build_preflight(ROOT, CODE, TEST) self.assertTrue(report["recovery_pilot_eligible"]) self.assertFalse(report["prospective_heldout_eligible"]) self.assertFalse(report["production_background_tolerances_frozen"]) self.assertFalse(report["seed_released"]) def test_NONPASS_preflight_is_never_written(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "forbidden.json" checkpoint = Path(directory) / "forbidden.md" with self.assertRaises(recovery.LogHRecoveryPreflightError): recovery.write_pass_preflight( {"all_preflight_gates_pass": False}, output, checkpoint ) self.assertFalse(output.exists()) self.assertFalse(checkpoint.exists()) if __name__ == "__main__": unittest.main()