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_balanced_coupled_stage_pilot as pilot import ap1_m1_balanced_coupled_precision_preflight as precision import ap1_m1_balanced_coupled_stage_preflight as preflight ROOT = Path(os.environ.get("APEIRON_AP1_ROOT", Path(__file__).parents[2])).resolve() CODE = ROOT / "AP1/CODE/ap1_m1_balanced_coupled_stage_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 = preflight.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.BalancedCoupledStagePlan() return { "all_pilot_gates_pass": True, "heldout_evaluated": False, "authority_sha256": { "AP1/CODE/ap1_m1_balanced_coupled_stage_pilot.py": pilot.file_sha256( CODE ), "AP1/CODE/test_ap1_m1_balanced_coupled_stage_pilot.py": pilot.file_sha256( TEST ), }, "plan": asdict(plan), "pilot_ensemble": { "metrics_for_freeze_or_gate": metrics_at_floor_fraction(), "stage_endpoint_aggregate_not_a_seed": { "N_relative": plan.pilot_span_N, "H_Mpl": 1.25e-61, }, }, } class BalancedCoupledPilotTests(unittest.TestCase): def test_g32_authorities_are_exact_and_preflight_is_pass(self) -> None: report = pilot.verify_g32_preflight(ROOT) self.assertTrue(report["all_preflight_gates_pass"]) self.assertEqual(report["background_runs"], 0) self.assertEqual(report["physical_response_kernel_runs"], 0) def test_g32a_authorities_are_exact_and_precision_preflight_is_pass(self) -> None: report = pilot.verify_g32a_preflight(ROOT) self.assertTrue(report["all_preflight_gates_pass"]) self.assertTrue(report["pilot_retry_eligible"]) self.assertEqual(report["background_runs"], 0) self.assertEqual(report["physical_response_kernel_runs"], 0) def test_plan_has_three_nested_pilot_and_heldout_levels(self) -> None: plan = preflight.BalancedCoupledStagePlan() plan.validate() for phase in ("pilot", "heldout"): coarse, middle, fine = plan._level_nodes(phase) self.assertEqual(fine - 1, 2 * (middle - 1)) self.assertEqual(middle - 1, 2 * (coarse - 1)) self.assertGreater( plan.heldout_evaluation_start_N, plan.prior_diagnostic_endpoint_N ) def test_pair_items_retains_every_field_source_and_maximum(self) -> None: items = pilot._pair_items(pair(4.0, 8.0)) self.assertEqual(len(items), 10) self.assertEqual(items["background.maximum"], 4.0) self.assertEqual(items["source.maximum"], 8.0) def test_three_level_quadratic_sequence_has_order_two(self) -> None: audit = pilot.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_three_level_nonmonotone_sequence_fails_closed(self) -> None: audit = pilot.three_level_convergence_audit( pair(1.0, 2.0), pair(1.1, 2.1), 0.5 ) self.assertFalse(audit["all_component_deltas_monotone"]) self.assertFalse(audit["all_component_orders_pass"]) def test_three_level_order_below_half_fails(self) -> None: ratio = 2.0 ** -0.49 audit = pilot.three_level_convergence_audit( pair(1.0, 1.0), pair(ratio, ratio), 0.5 ) self.assertTrue(audit["all_component_deltas_monotone"]) self.assertFalse(audit["all_component_orders_pass"]) def test_below_floor_monotone_source_order_is_unidentifiable_not_failed(self) -> None: audit = pilot.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 ) self.assertAlmostEqual( audit["minimum_finite_observed_Richardson_order_above_floor"], 2.0 ) def test_below_floor_nonmonotone_source_still_fails(self) -> None: audit = pilot.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_accepted_without_infinite_json(self) -> None: audit = pilot.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"]) self.assertIsNone(audit["minimum_finite_observed_Richardson_order"]) json.dumps(audit, allow_nan=False) def test_phase_gate_requires_exact_chart_transport_not_switch_occurrence(self) -> None: diagnostic = { "transport_method": precision.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, } ensemble = { "levels": { label: {"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, } with mock.patch.object(pilot.g29, "_absolute_gates", return_value={}): gates = pilot._phase_gates( ensemble, preflight.BalancedCoupledStagePlan(), "pilot" ) self.assertTrue( gates[ "all_balanced_transports_use_exact_G31_method_without_vacuum_reset" ] ) def test_freeze_thresholds_obey_predeclared_floors(self) -> None: thresholds = pilot._thresholds_from_metrics(metrics_at_floor_fraction()) policy = preflight.PREDECLARED_POLICY["inherited_metric_freeze_policy"] for name, value in thresholds.items(): self.assertEqual(value, float(policy[name]["floor"])) def test_freeze_rejects_an_incomplete_metric_set(self) -> None: incomplete = metrics_at_floor_fraction() incomplete.pop(next(iter(incomplete))) with self.assertRaises(pilot.BalancedCoupledPilotError): pilot._thresholds_from_metrics(incomplete) 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.BalancedCoupledPilotError): 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_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["heldout_evaluated"]) self.assertEqual(report["heldout_plan"]["nodes"], [1281, 2561, 5121]) self.assertGreater( report["heldout_plan"]["evaluation_start_N"], report["heldout_plan"]["overlap_reference"]["N_relative"], ) def test_build_pilot_keeps_kernel_seed_and_checkpoint_locked(self) -> None: plan = preflight.BalancedCoupledStagePlan() preflight_report = { "plan": asdict(plan), "all_preflight_gates_pass": True, } ensemble = { "stage_endpoint_aggregate_not_a_seed": { "N_relative": plan.pilot_span_N, "H_Mpl": 1.0, } } checkpoint = { "heldout_physical_diagnostics": {"H_endpoint_Mpl": 1.0} } with ( mock.patch.object(pilot, "verify_g32_preflight", return_value=preflight_report), mock.patch.object( pilot, "verify_g32a_preflight", return_value={ "all_preflight_gates_pass": True, "precision_policy": {}, }, ), mock.patch.object(preflight, "verify_authorities", return_value=(checkpoint, {})), 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.assertFalse(report["checkpoint_eligible"]) self.assertFalse(report["seed_released"]) self.assertEqual(report["physical_response_kernel_runs"], 0) self.assertFalse(report["production_background_tolerances_frozen"]) if __name__ == "__main__": unittest.main()