from __future__ import annotations from contextlib import redirect_stdout from dataclasses import asdict import io import json import math import os from pathlib import Path import tempfile import unittest from unittest import mock import ap1_m1_logh_positive_branch_recovery_preflight as preflight import ap1_m1_logh_recovery_pilot as pilot ROOT = Path(os.environ.get("APEIRON_AP1_ROOT", Path(__file__).parents[2])).resolve() CODE = ROOT / "AP1/CODE/ap1_m1_logh_recovery_pilot.py" TEST = Path(__file__).resolve() def pair(background: float, source: float) -> dict[str, object]: return { "background_relative_delta_by_field": { "sigma": background, "sigmadot": background / 2.0, "theta": background / 3.0, "thetadot": background / 4.0, "H": background / 5.0, }, "source_gross_relative_delta_by_source": { "rho": source, "pressure": source / 2.0, "chi2": source / 3.0, }, "max_background_time_relative_delta": background, "max_source_time_gross_relative_delta": source, } def metrics_at_floor_fraction() -> dict[str, float]: policy = pilot.g32.PREDECLARED_POLICY["inherited_metric_freeze_policy"] return { name: float(specification["floor"]) / (2.0 * float(specification["multiplier"])) for name, specification in policy.items() } def synthetic_pilot() -> dict[str, object]: plan = preflight.LogHRecoveryPlan() return { "all_pilot_gates_pass": True, "prospective_heldout_evaluated": False, "authority_sha256": { "AP1/CODE/ap1_m1_logh_recovery_pilot.py": pilot.file_sha256(CODE), "AP1/CODE/test_ap1_m1_logh_recovery_pilot.py": pilot.file_sha256(TEST), }, "plan": asdict(plan), "recovery_pilot_ensemble": { "metrics_for_freeze_or_gate": metrics_at_floor_fraction(), "stage_endpoint_aggregate_not_a_seed": { "N_relative": plan.recovery_pilot_span_N, "H_Mpl": 1.25e-61, }, }, } def exact_chart_ensemble() -> dict[str, object]: diagnostic = { "transport_method": pilot.g32b.g32a.TRANSPORT_METHOD, "dynamic_chart_switches": 0, "vacuum_resets": 0, "max_abs_active_ratio": 0.0, "max_wronskian_relative_error": 0.0, "max_initial_handoff_wronskian_relative_error": 0.0, "max_transfer_determinant_defect_before_projection": 0.0, "final_resolved_modes": 10, } level = { "config": {"background_chart": preflight.LOGH_CHART}, "H_endpoint_Mpl": 1.0e-61, "resolved_diagnostics": diagnostic, } return { "levels": { label: { "config": dict(level["config"]), "H_endpoint_Mpl": level["H_endpoint_Mpl"], "resolved_diagnostics": dict(diagnostic), } for label in ("coarse", "middle", "fine") }, "operator_convergence": { "coarse_entry_diagnostics": dict(diagnostic), "coarse_mode_step_diagnostics": dict(diagnostic), "max_balanced_mode_step_gross_relative_delta": 0.0, }, "source_diagnostics": {"balanced_chart_used": True}, "time_convergence": { "three_level_audit": { "all_component_deltas_monotone": True, "all_component_orders_pass": True, } }, "same_operator_settings_on_all_levels": True, "trajectory_rows_persisted": 0, "physical_response_kernel_runs": 0, } class LogHRecoveryPilotTests(unittest.TestCase): def test_g32c_authorities_are_exact_and_recovery_only_is_open(self) -> None: report = pilot.verify_g32c_preflight(ROOT) self.assertTrue(report["all_preflight_gates_pass"]) self.assertTrue(report["recovery_pilot_eligible"]) self.assertFalse(report["prospective_heldout_eligible"]) self.assertEqual(report["background_runs"], 0) self.assertEqual(report["physical_response_kernel_runs"], 0) def test_plan_has_nested_equal_step_recovery_and_new_heldout_levels(self) -> None: plan = preflight.LogHRecoveryPlan() plan.validate() steps = [] for phase, span in ( ("recovery_pilot", plan.recovery_pilot_span_N), ("prospective_heldout", plan.prospective_heldout_span_N), ): coarse, middle, fine = plan._level_nodes(phase) self.assertEqual(fine - 1, 2 * (middle - 1)) self.assertEqual(middle - 1, 2 * (coarse - 1)) steps.append(span / (fine - 1)) self.assertAlmostEqual(steps[0], steps[1], places=18) self.assertGreater( plan.prospective_heldout_evaluation_start_N, plan.consumed_G32B_span_N, ) def test_phase_settings_are_exact_and_unknown_phase_fails(self) -> None: plan = preflight.LogHRecoveryPlan() self.assertEqual( pilot._phase_settings(plan, "recovery_pilot"), ((1281, 2561, 5121), 0.10, 0.085), ) self.assertEqual( pilot._phase_settings(plan, "prospective_heldout"), ((1537, 3073, 6145), 0.12, 0.105), ) with self.assertRaises(pilot.LogHRecoveryPilotError): pilot._phase_settings(plan, "invalid") def test_consumed_domain_is_nonblind_and_new_heldout_is_locked(self) -> None: report = json.loads( ( ROOT / "AP1/APEIRON_AP1_M1_LOGH_POSITIVE_BRANCH_RECOVERY_PREFLIGHT_LATEST.json" ).read_text() ) audit = report["failed_G32B_heldout_audit"] self.assertFalse(audit["consumed_N_le_0p10_domain_is_blind"]) self.assertFalse(report["prospective_heldout_eligible"]) self.assertEqual(audit["registered_G32B_heldout_attempts"], 1) self.assertFalse(audit["G32B_heldout_report_written"]) self.assertFalse(audit["G32B_heldout_checkpoint_or_seed_released"]) def test_floor_aware_quadratic_sequence_has_order_two(self) -> None: audit = pilot.g32b.three_level_convergence_audit( pair(4.0, 8.0), pair(1.0, 2.0), 0.5 ) self.assertTrue(audit["all_component_deltas_monotone"]) self.assertTrue(audit["all_component_orders_pass"]) self.assertAlmostEqual( audit["minimum_finite_observed_Richardson_order"], 2.0 ) def test_below_floor_monotone_is_unidentifiable_not_failed(self) -> None: audit = pilot.g32b.three_level_convergence_audit( pair(4.0e-7, 9.4e-17), pair(1.0e-7, 7.9e-17), 0.5 ) self.assertTrue(audit["all_component_deltas_monotone"]) self.assertTrue(audit["all_component_orders_pass"]) self.assertGreater( audit["components_below_inherited_resolution_floor"], 0 ) def test_below_floor_nonmonotone_still_fails(self) -> None: audit = pilot.g32b.three_level_convergence_audit( pair(4.0e-7, 7.9e-17), pair(1.0e-7, 9.4e-17), 0.5 ) self.assertFalse(audit["all_component_deltas_monotone"]) self.assertFalse(audit["all_component_orders_pass"]) def test_exact_zero_deltas_are_strict_json(self) -> None: audit = pilot.g32b.three_level_convergence_audit( pair(0.0, 0.0), pair(0.0, 0.0), 0.5 ) self.assertTrue(audit["all_component_deltas_monotone"]) self.assertTrue(audit["all_component_orders_pass"]) json.dumps(audit, allow_nan=False) def test_phase_gates_require_exact_logh_chart_and_positive_H(self) -> None: ensemble = exact_chart_ensemble() with mock.patch.object(pilot.g29, "_absolute_gates", return_value={}): gates = pilot._phase_gates( ensemble, preflight.LogHRecoveryPlan(), "recovery_pilot" ) self.assertTrue(gates["all_background_levels_use_exact_logH_chart"]) self.assertTrue(gates["all_level_endpoints_remain_finite_positive_H"]) ensemble["levels"]["middle"]["config"]["background_chart"] = "direct-H" ensemble["levels"]["fine"]["H_endpoint_Mpl"] = 0.0 with mock.patch.object(pilot.g29, "_absolute_gates", return_value={}): gates = pilot._phase_gates( ensemble, preflight.LogHRecoveryPlan(), "recovery_pilot" ) self.assertFalse(gates["all_background_levels_use_exact_logH_chart"]) self.assertFalse(gates["all_level_endpoints_remain_finite_positive_H"]) def test_phase_gates_reject_unknown_phase(self) -> None: with self.assertRaises(pilot.LogHRecoveryPilotError): pilot._phase_gates( exact_chart_ensemble(), preflight.LogHRecoveryPlan(), "invalid" ) def test_freeze_thresholds_obey_predeclared_floors(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "pilot.json" path.write_text(json.dumps(synthetic_pilot()) + "\n") report = pilot.build_freeze(path, CODE, TEST) policy = pilot.g32.PREDECLARED_POLICY["inherited_metric_freeze_policy"] for name, value in report["thresholds"].items(): self.assertEqual(value, float(policy[name]["floor"])) def test_freeze_is_hash_bound_disjoint_and_not_production(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "pilot.json" path.write_text(json.dumps(synthetic_pilot()) + "\n") report = pilot.build_freeze(path, CODE, TEST) self.assertTrue(report["all_tolerance_freeze_gates_pass"]) self.assertTrue(report["method_tolerances_frozen"]) self.assertFalse(report["production_background_tolerances_frozen"]) self.assertFalse(report["prospective_heldout_evaluated"]) self.assertEqual( report["prospective_heldout_plan"]["nodes"], [1537, 3073, 6145] ) self.assertGreater( report["prospective_heldout_plan"]["evaluation_start_N"], report["prospective_heldout_plan"]["overlap_reference"]["N_relative"], ) def test_freeze_rejects_incomplete_metric_set(self) -> None: report = synthetic_pilot() metrics = report["recovery_pilot_ensemble"]["metrics_for_freeze_or_gate"] metrics.pop(next(iter(metrics))) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "pilot.json" path.write_text(json.dumps(report) + "\n") with self.assertRaises(pilot.g32b.BalancedCoupledPilotError): pilot.build_freeze(path, CODE, TEST) def test_pass_writer_refuses_nonpass_without_creating_file(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "forbidden.json" with self.assertRaises(pilot.LogHRecoveryPilotError): pilot._write_pass_json( {"all_pilot_gates_pass": False}, output, "all_pilot_gates_pass", ) self.assertFalse(output.exists()) def test_pass_writer_serializes_strict_json_exclusively(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "pass.json" report = {"all_pilot_gates_pass": True, "value": 1.0} pilot._write_pass_json(report, output, "all_pilot_gates_pass") self.assertEqual(json.loads(output.read_text()), report) with self.assertRaises(FileExistsError): pilot._write_pass_json(report, output, "all_pilot_gates_pass") def test_nonfinite_report_is_not_materialized(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "nan.json" with self.assertRaises(ValueError): pilot._write_pass_json( {"all_pilot_gates_pass": True, "value": math.nan}, output, "all_pilot_gates_pass", ) self.assertFalse(output.exists()) def test_cli_nonpass_path_prints_diagnostic_but_writes_nothing(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "nonpass.json" stream = io.StringIO() with redirect_stdout(stream), self.assertRaises(SystemExit): pilot._write_or_fail( { "classification": "NONPASS", "all_pilot_gates_pass": False, "gates": {"example": False}, }, output, "all_pilot_gates_pass", ) self.assertFalse(output.exists()) self.assertIn('"output_written": false', stream.getvalue()) def test_build_pilot_keeps_kernel_seed_and_new_heldout_locked(self) -> None: plan = preflight.LogHRecoveryPlan() preflight_report = { "recovery_plan": asdict(plan), "all_preflight_gates_pass": True, } ensemble = { "stage_endpoint_aggregate_not_a_seed": { "N_relative": plan.recovery_pilot_span_N, "H_Mpl": 1.0, } } with ( mock.patch.object( pilot, "verify_g32c_preflight", return_value=preflight_report ), mock.patch.object(pilot, "_load_arrays", return_value={}), mock.patch.object( pilot, "evaluate_three_level_phase", return_value=ensemble ), mock.patch.object(pilot, "_phase_gates", return_value={"phase": True}), mock.patch.object(pilot, "_ceiling_gates", return_value={"ceiling": True}), ): report = pilot.build_pilot(ROOT, CODE, TEST) self.assertTrue(report["all_pilot_gates_pass"]) self.assertTrue(report["pilot_is_nonblind_development"]) self.assertFalse(report["prospective_heldout_evaluated"]) self.assertFalse(report["checkpoint_eligible"]) self.assertFalse(report["seed_released"]) self.assertEqual(report["physical_response_kernel_runs"], 0) self.assertFalse(report["production_background_tolerances_frozen"]) def test_heldout_rejects_metric_set_drift_after_freeze(self) -> None: with tempfile.TemporaryDirectory() as directory: base = Path(directory) pilot_path = base / "pilot.json" pilot_path.write_text(json.dumps(synthetic_pilot()) + "\n") freeze_report = pilot.build_freeze(pilot_path, CODE, TEST) freeze_path = base / "freeze.json" freeze_path.write_text(json.dumps(freeze_report) + "\n") ensemble = { "metrics_for_freeze_or_gate": { "unexpected_metric": 0.0, } } with ( mock.patch.object( pilot, "verify_g32c_preflight", return_value={"recovery_plan": asdict(preflight.LogHRecoveryPlan())}, ), mock.patch.object(pilot, "_load_arrays", return_value={}), mock.patch.object( pilot, "evaluate_three_level_phase", return_value=ensemble ), ): with self.assertRaises(pilot.LogHRecoveryPilotError): pilot.build_heldout( ROOT, pilot_path, freeze_path, CODE, TEST ) if __name__ == "__main__": unittest.main()