from __future__ import annotations import hashlib import json import os from pathlib import Path import unittest from app.main import compile_system_xml_network from app.simulation.benchmark_regression import ( DEFAULT_MANIFEST_PATH, evaluate_regression_golden, load_regression_golden, load_regression_manifest, run_regression_suite, source_simulation_config, ) from app.simulation.systems.generic import GenericFluidSystem from app.system_xml import validate_system_xml_document LONG_RUN_ENVIRONMENT = "RUN_TEST_MQL_8_LONG_REGRESSION" class TestMql8StaticRegressionTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.manifest = load_regression_manifest(DEFAULT_MANIFEST_PATH) cls.source_path = Path(cls.manifest["_sourcePath"]) cls.source_payload = cls.source_path.read_bytes() cls.validation = validate_system_xml_document(cls.source_payload) def test_authoritative_xml_hash_and_simulation_settings_are_fixed(self) -> None: source = self.manifest["source"] self.assertEqual( hashlib.sha256(self.source_payload).hexdigest(), source["sha256"] ) self.assertEqual(len(self.source_payload), source["bytes"]) self.assertEqual( source_simulation_config(self.source_payload), source["simulation"] ) self.assertEqual(self.source_path.name, "test-mql-8.xml") companion = source["companionProject"] companion_path = Path(self.manifest["_companionPath"]) companion_payload = companion_path.read_bytes() companion_document = json.loads(companion_payload) self.assertEqual(companion_path.name, "test-mql-8.json") self.assertEqual( hashlib.sha256(companion_payload).hexdigest(), companion["sha256"] ) self.assertEqual(len(companion_payload), companion["bytes"]) self.assertFalse(companion["executionInput"]) self.assertEqual( companion_document["simulation"], { "t_start": source["simulation"]["tStart"], "t_stop": source["simulation"]["tStop"], "step": source["simulation"]["sampleStep"], "max_step": source["simulation"]["maxStep"], "method": source["simulation"]["method"], }, ) def test_authoritative_xml_validates_and_matches_structure_snapshot(self) -> None: self.assertTrue(self.validation.valid, self.validation.as_dict()) assert self.validation.document is not None network = compile_system_xml_network(self.validation.document) system = GenericFluidSystem(network) expected = self.manifest["structure"] pressure_flow = network.pressure_flow_structure_dict() causal_execution = system.pressure_flow_solver.causal_execution_diagnostics() jacobian = system.jacobian_sparsity_diagnostics() self.assertEqual(len(network.components), expected["componentCount"]) self.assertEqual(len(network.connections), expected["connectionCount"]) self.assertEqual( len(network.dynamic_components()), expected["dynamicComponentCount"] ) self.assertEqual( sum(component.state_size for component in network.dynamic_components()), expected["stateCount"], ) self.assertEqual( len(network.result_variable_metadata()), expected["resultVariableCount"] ) self.assertEqual( pressure_flow["unknownCount"], expected["pressureFlowUnknownCount"] ) self.assertEqual( pressure_flow["equationCount"], expected["pressureFlowEquationCount"] ) self.assertEqual(pressure_flow["isSquare"], expected["pressureFlowIsSquare"]) for key in ( "logicalEffortCoordinateCount", "eliminatedEffortAliasCount", "canonicalCoordinateCount", "compatibilityScatterCount", ): self.assertEqual(causal_execution[key], expected[key]) self.assertEqual( jacobian["nonzeroCount"], expected["jacobianNonzeroCount"] ) self.assertEqual( jacobian["colorGroupCount"], expected["jacobianColorGroupCount"] ) self.assertEqual( system.mechanical_state_reducer.has_state_events, expected["hasMechanicalStateEvents"], ) def test_contact_aware_mechanical_tolerance_plan_is_proof_gated(self) -> None: assert self.validation.document is not None system = GenericFluidSystem( compile_system_xml_network(self.validation.document) ) reducer = system.mechanical_state_reducer plan = reducer.absolute_tolerance_plan( 1.0e-8, mode="contact-aware-v1", ) diagnostics = plan.as_dict() self.assertEqual(len(plan.values), len(system.initial_state_vector())) self.assertEqual(diagnostics["groupCount"], 10) self.assertEqual(diagnostics["eligibleGroupCount"], 8) self.assertEqual(diagnostics["relaxedVelocityStateCount"], 8) self.assertEqual(diagnostics["relaxedPositionStateCount"], 0) by_component = { group.components[0]: group for group in plan.groups } for index in range(1, 9): group = by_component[f"amesim_mecmas21_{index}"] self.assertTrue(group.eligible) self.assertEqual(group.reason, "eligibleFlexibleContact") self.assertAlmostEqual(group.velocity_atol, 1.0e-10) self.assertEqual(group.position_atol, 1.0e-12) self.assertAlmostEqual( group.minimum_damping_strength_ratio, 10.0, ) self.assertAlmostEqual( group.minimum_force_limited_velocity_atol, 1.0e-10, ) self.assertEqual( group.contacts, (f"amesim_lstp00a_{index}",), ) for index in (9, 10): group = by_component[f"amesim_mecmas21_{index}"] self.assertFalse(group.eligible) self.assertEqual(group.reason, "discreteEndstop") self.assertEqual(group.velocity_atol, 1.0e-12) self.assertEqual(group.position_atol, 1.0e-12) legacy = reducer.absolute_tolerance_plan(1.0e-8, mode="legacy") self.assertTrue( all( group.velocity_atol == 1.0e-12 and group.position_atol == 1.0e-12 for group in legacy.groups ) ) def test_signal_event_schedule_is_fixed_for_every_horizon(self) -> None: assert self.validation.document is not None system = GenericFluidSystem( compile_system_xml_network(self.validation.document) ) for case_id in self.manifest["sequence"]: variant = self.manifest["variants"][case_id] actual = system.signal_resolver.event_times( 0.0, float(variant["stopTime"]) ) self.assertEqual( actual, tuple(variant["expectedSignalEventTimes"]), case_id, ) def test_historical_v2_golden_replays_its_source_report(self) -> None: repository_root = Path(self.manifest["_repositoryRoot"]) golden_path = ( repository_root / "tests/baselines/simulation/test_mql_8/goldens/production-0.2s-v1.json" ) golden = load_regression_golden( golden_path, expected_sha256="99fa7b3631a89f59f86551847175f3701a1567d8d0bd10d87545734f17515d72", repository_root=repository_root, ) source_report = golden["provenance"]["sourceReport"] report = json.loads( (repository_root / source_report["path"]).read_text(encoding="utf-8") ) case = next(item for item in report["cases"] if item["caseId"] == "0.2s") audit = evaluate_regression_golden(case["worker"]["summary"], golden) self.assertNotEqual( golden["sourceXmlSha256"], self.manifest["source"]["sha256"], ) self.assertTrue(audit["passed"], audit) self.assertEqual(audit["comparedValueCount"], 402) self.assertEqual(audit["maxAbsoluteError"], 0.0) self.assertEqual(audit["maxToleranceRatio"], 0.0) def test_approved_production_golden_replays_its_source_report(self) -> None: repository_root = Path(self.manifest["_repositoryRoot"]) reference = self.manifest["variants"]["0.2s"]["goldens"]["production"] golden = load_regression_golden( repository_root / reference["path"], expected_sha256=reference["sha256"], repository_root=repository_root, ) source_report = golden["provenance"]["sourceReport"] report = json.loads( (repository_root / source_report["path"]).read_text(encoding="utf-8") ) case = next(item for item in report["cases"] if item["caseId"] == "0.2s") audit = evaluate_regression_golden(case["worker"]["summary"], golden) self.assertTrue(audit["passed"], audit) self.assertEqual(audit["comparedValueCount"], 426) self.assertEqual(audit["maxAbsoluteError"], 0.0) self.assertEqual(audit["maxToleranceRatio"], 0.0) def test_historical_v2_extension_decision_remains_auditable(self) -> None: repository_root = Path(self.manifest["_repositoryRoot"]) runs = repository_root / "tests/baselines/simulation/test_mql_8/runs" decision = json.loads( (runs / "2026-08-17-production-v2-extension-decision.json").read_text( encoding="utf-8" ) ) report_path = runs / decision["sourceReport"] report_payload = report_path.read_bytes() report = json.loads(report_payload) source_case = next( item for item in report["cases"] if item["caseId"] == "0.2s" ) first_decision = decision["decisions"][0] self.assertEqual( decision["sourceXmlSha256"], "170463d65d074da01f0f9e9dab730b3815c94c1cc80b5190ec2e3fe623da74d3", ) self.assertNotEqual( decision["sourceXmlSha256"], self.manifest["source"]["sha256"] ) self.assertEqual( hashlib.sha256(report_payload).hexdigest(), decision["sourceReportSha256"], ) self.assertTrue(source_case["acceptance"]["passed"]) self.assertEqual( decision["observedCase"]["workerWallSeconds"], source_case["worker"]["wallSeconds"], ) expected_prediction = ( source_case["worker"]["wallSeconds"] * self.manifest["variants"]["1s"]["stopTime"] / source_case["stopTime"] * self.manifest["execution"]["predictionSafetyFactor"] ) self.assertAlmostEqual( first_decision["predictedWallSeconds"], expected_prediction ) self.assertGreater( first_decision["predictedWallSeconds"], first_decision["softTimeoutSeconds"], ) self.assertEqual(first_decision["outcome"], "deferred") self.assertTrue(decision["simulationWasNotStartedForDeferredCases"]) def test_extension_decision_is_bound_to_the_current_report_and_budget(self) -> None: repository_root = Path(self.manifest["_repositoryRoot"]) evidence = self.manifest["acceptanceEvidence"] decision_reference = evidence["extensionDecision"] decision_path = repository_root / decision_reference["path"] decision_payload = decision_path.read_bytes() decision = json.loads(decision_payload) report_reference = evidence["production0.2ApprovedReplay"] report_path = repository_root / report_reference["path"] report_payload = report_path.read_bytes() report = json.loads(report_payload) source_case = next( item for item in report["cases"] if item["caseId"] == "0.2s" ) first_decision = decision["decisions"][0] self.assertEqual( decision["sourceXmlSha256"], self.manifest["source"]["sha256"] ) self.assertEqual(decision["sourceReport"], report_path.name) self.assertEqual(len(report_payload), report_reference["bytes"]) self.assertEqual( hashlib.sha256(report_payload).hexdigest(), report_reference["sha256"] ) self.assertEqual(len(decision_payload), decision_reference["bytes"]) self.assertEqual( hashlib.sha256(decision_payload).hexdigest(), decision_reference["sha256"], ) self.assertEqual( hashlib.sha256(report_payload).hexdigest(), decision["sourceReportSha256"], ) self.assertTrue(source_case["acceptance"]["passed"]) self.assertEqual( decision["observedCase"]["workerWallSeconds"], source_case["worker"]["wallSeconds"], ) expected_prediction = ( source_case["worker"]["wallSeconds"] * self.manifest["variants"]["1s"]["stopTime"] / source_case["stopTime"] * self.manifest["execution"]["predictionSafetyFactor"] ) self.assertAlmostEqual( first_decision["predictedWallSeconds"], expected_prediction ) self.assertLessEqual( first_decision["predictedWallSeconds"], first_decision["softTimeoutSeconds"], ) self.assertEqual(first_decision["outcome"], "eligible") self.assertEqual(first_decision["reason"], "predictedWallWithinSoftBudget") self.assertEqual( [item["outcome"] for item in decision["decisions"]], ["eligible", "notEvaluated", "notEvaluated"], ) self.assertTrue(decision["extensionWasNotStarted"]) self.assertEqual( decision["scopeReason"], "OPT-00 closes at the approved production 0.2 s gate; longer progressive horizons remain separately staged work.", ) def test_periodic_main_lane_exercises_the_approved_production_golden(self) -> None: workflow = ( Path(self.manifest["_repositoryRoot"]) / ".github/workflows/solver-regression.yml" ).read_text(encoding="utf-8") self.assertIn("default: production", workflow) self.assertIn("inputs.lane || 'production'", workflow) @unittest.skipUnless( os.getenv(LONG_RUN_ENVIRONMENT, "").strip().lower() in {"1", "true", "yes"}, f"Set {LONG_RUN_ENVIRONMENT}=1 to run bounded 0.01/0.2/1/5/10 s integration.", ) class TestMql8ProgressiveLongRegressionTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.manifest = load_regression_manifest(DEFAULT_MANIFEST_PATH) def test_all_horizons_complete_or_stop_at_the_first_bounded_failure(self) -> None: lane = os.getenv("TEST_MQL_8_REGRESSION_LANE", "production") report = run_regression_suite(DEFAULT_MANIFEST_PATH, lane=lane) outcomes = [case["outcome"] for case in report["cases"]] self.assertEqual(outcomes[0], "completed", report["cases"][0]) if "deferred" in outcomes: first_deferred = outcomes.index("deferred") self.assertTrue( all(outcome == "deferred" for outcome in outcomes[first_deferred:]) ) predecessor = report["cases"][first_deferred - 1] self.fail( "Progressive run stopped within its configured safety budget; " f"optimize before resuming. Predecessor: {predecessor}" ) self.assertEqual( outcomes, ["completed"] * len(self.manifest["sequence"]), report["cases"], ) if __name__ == "__main__": unittest.main()