"""Contract tests for the system-simulation optimization helper. The optimizer is exercised with deterministic in-process simulation results. No test starts the FastAPI service or submits a real, potentially long simulation. """ from __future__ import annotations import argparse import contextlib import copy import dataclasses import hashlib import importlib.util import json import math import sys import tempfile import unittest import xml.etree.ElementTree as ET from pathlib import Path from unittest import mock REPOSITORY_ROOT = Path(__file__).resolve().parents[1] SCRIPTS_DIRECTORY = ( REPOSITORY_ROOT / "skills" / "system-simulation" / "scripts" ) SCRIPT_PATH = SCRIPTS_DIRECTORY / "optimization_skill.py" MODULE_NAME = "system_optimization_skill_under_test" MODULE_SPEC = importlib.util.spec_from_file_location(MODULE_NAME, SCRIPT_PATH) if MODULE_SPEC is None or MODULE_SPEC.loader is None: # pragma: no cover raise RuntimeError(f"Cannot load optimization helper from {SCRIPT_PATH}") optimization = importlib.util.module_from_spec(MODULE_SPEC) sys.modules[MODULE_NAME] = optimization MODULE_SPEC.loader.exec_module(optimization) BASE_URL = "http://127.0.0.1:8000" ABSENT = object() RESULT_VARIABLES = [ { "key": "sensor.output", "componentId": "sensor", "name": "output", "label": "Output", "quantity": "displacement", "unit": "m", }, { "key": "sensor.limit", "componentId": "sensor", "name": "limit", "label": "Limit response", "quantity": "displacement", "unit": "m", }, ] PROJECT = { "projectSchemaVersion": 1, "name": "optimization-fixture", "nodes": [ { "id": "component-a", "type": "component", "position": {"x": 10, "y": 20}, "data": { "componentType": "FIXTURE", "label": "A", "parameters": {"gain": "1 + 1", "fixed": 99.0}, "custom": {"preserve": True}, }, }, { "id": "component-b", "type": "component", "data": { "componentType": "FIXTURE", "parameters": {"gain": 7.0}, }, }, ], "edges": [{"id": "edge-1", "source": "component-a", "target": "component-b"}], "simulation": {"startTime": 0.0, "endTime": 1.0, "sampleStep": 0.1}, "customRoot": {"preserve": [1, 2, 3]}, } BASELINE_XML = b""" """ def _json_bytes(value: object) -> bytes: return (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8") def _valid_spec( *, seed: int = 12345, max_simulation_runs: int = 6, lower: float = 0.0, upper: float = 4.0, ) -> dict[str, object]: return { "optimizationSchemaVersion": 1, "objective": { "resultKey": "sensor.output", "expectedUnit": "m", "statistic": {"kind": "final", "window": None}, "goal": {"kind": "minimize"}, }, "designVariables": [ { "id": "gain", "componentId": "component-a", "parameter": "gain", "unit": "m", "lower": lower, "upper": upper, } ], "constraints": [ { "id": "limit", "resultKey": "sensor.limit", "expectedUnit": "m", "statistic": {"kind": "maximum", "window": None}, "lower": None, "upper": 4.0, "tolerance": 0.0, "scale": 1.0, } ], "algorithm": { "name": "differentialEvolution", "seed": seed, "populationSize": 4, "mutationFactor": 0.8, "crossoverProbability": 0.7, }, "budget": { "maxSimulationRuns": max_simulation_runs, "maxWallSeconds": 60.0, }, "validation": { "relativeTolerance": 1e-12, "absoluteTolerance": 1e-12, }, } def _inspection( *, optimization_eligible: object = ABSENT, editor: object = ABSENT, options: object = ABSENT, unit: object = "m", value: object = 2.0, minimum: object = 0.0, maximum: object = 4.0, minimum_exclusive: object = False, ) -> dict[str, object]: contract = { "name": "gain", "label": "Gain", "quantity": "displacement", "unit": unit, "value": value, "minimum": minimum, "maximum": maximum, "minimumExclusive": minimum_exclusive, } if optimization_eligible is not ABSENT: contract["optimizationEligible"] = optimization_eligible if editor is not ABSENT: contract["editor"] = editor if options is not ABSENT: contract["options"] = options return { "ok": True, "system": { "name": "optimization-fixture", "componentDetails": [ { "id": "component-a", "compiled": {"parameters": [contract]}, "source": copy.deepcopy(PROJECT["nodes"][0]), } ], "resultVariables": copy.deepcopy(RESULT_VARIABLES), }, } def _make_plan( directory: Path, *, output_name: str = "optimization-output", seed: int = 12345, max_simulation_runs: int = 6, lower: float = 0.0, upper: float = 4.0, inspection: dict[str, object] | None = None, ) -> optimization.RuntimePlan: directory.mkdir(parents=True, exist_ok=True) source_path = directory / "source.json" source_path.write_bytes(_json_bytes(PROJECT)) spec_payload = _valid_spec( seed=seed, max_simulation_runs=max_simulation_runs, lower=lower, upper=upper, ) spec_path = directory / "optimization-spec.json" spec_path.write_bytes(_json_bytes(spec_payload)) source = optimization.simulation.load_source(str(source_path), "json") spec_source = optimization.simulation.load_source(str(spec_path), "json") spec = optimization.parse_optimization_spec(spec_source.parsed) resolved_inspection = inspection if inspection is not None else _inspection() variables = optimization.simulation._available_variables(resolved_inspection) resolved = optimization._resolve_design_variables(spec, resolved_inspection) output_directory = directory / output_name token = optimization._confirmation_token( source_sha256=source.sha256, spec_sha256=spec_source.sha256, baseline_xml_sha256=hashlib.sha256(BASELINE_XML).hexdigest(), output_directory=output_directory, base_url=BASE_URL, timeout=10.0, ) return optimization.RuntimePlan( source=source, spec_source=spec_source, spec=spec, inspection=resolved_inspection, baseline_xml=BASELINE_XML, variables=variables, resolved_design_variables=resolved, output_directory=output_directory, base_url=BASE_URL, timeout=10.0, confirmation_token=token, ) def _xml_parameter_values(xml: bytes) -> dict[tuple[str, str], str]: root = ET.fromstring(xml) values: dict[tuple[str, str], str] = {} for component in root.findall("./Components/Component"): component_id = component.get("id") if component_id is None: continue for parameter in component.findall("./Parameter"): name = parameter.get("name") value = parameter.get("value") if name is not None and value is not None: values[(component_id, name)] = value return values class OptimizationSpecTests(unittest.TestCase): def test_spec_rejects_unknown_fields_at_every_nested_contract(self) -> None: parsed = optimization.parse_optimization_spec(_valid_spec()) self.assertEqual(parsed.algorithm.name, "differentialEvolution") cases: list[tuple[str, dict[str, object]]] = [] root_unknown = _valid_spec() root_unknown["surprise"] = True cases.append(("root", root_unknown)) objective_unknown = _valid_spec() objective = objective_unknown["objective"] assert isinstance(objective, dict) objective["label"] = "not part of schema 1" cases.append(("objective", objective_unknown)) variable_unknown = _valid_spec() design_variables = variable_unknown["designVariables"] assert isinstance(design_variables, list) assert isinstance(design_variables[0], dict) design_variables[0]["logScale"] = True cases.append(("design variable", variable_unknown)) window_unknown = _valid_spec() objective = window_unknown["objective"] assert isinstance(objective, dict) statistic = objective["statistic"] assert isinstance(statistic, dict) statistic["window"] = {"start": 0.0, "end": 1.0, "closed": True} cases.append(("window", window_unknown)) algorithm_unknown = _valid_spec() algorithm = algorithm_unknown["algorithm"] assert isinstance(algorithm, dict) algorithm["workers"] = 2 cases.append(("algorithm", algorithm_unknown)) for label, payload in cases: with self.subTest(label=label): with self.assertRaises(optimization.simulation.InputError) as caught: optimization.parse_optimization_spec(payload) self.assertEqual(caught.exception.code, "OPTIMIZATION_SPEC_INVALID") self.assertTrue(caught.exception.details["unknown"]) def test_spec_rejects_boolean_numeric_values_and_invalid_bounds(self) -> None: boolean_bound = _valid_spec() variables = boolean_bound["designVariables"] assert isinstance(variables, list) and isinstance(variables[0], dict) variables[0]["lower"] = False with self.assertRaises(optimization.simulation.InputError) as caught: optimization.parse_optimization_spec(boolean_bound) self.assertEqual(caught.exception.code, "OPTIMIZATION_SPEC_INVALID") reversed_bounds = _valid_spec() variables = reversed_bounds["designVariables"] assert isinstance(variables, list) and isinstance(variables[0], dict) variables[0]["lower"] = 4.0 with self.assertRaises(optimization.simulation.InputError) as caught: optimization.parse_optimization_spec(reversed_bounds) self.assertEqual(caught.exception.code, "OPTIMIZATION_BOUNDS_INVALID") overflowing_span = _valid_spec(lower=-1e308, upper=1e308) with self.assertRaises(optimization.simulation.InputError) as caught: optimization.parse_optimization_spec(overflowing_span) self.assertEqual(caught.exception.code, "OPTIMIZATION_BOUNDS_INVALID") class StatisticTests(unittest.TestCase): def test_statistics_use_linearly_interpolated_window_boundaries(self) -> None: times = [0.0, 1.0, 2.0] values = [0.0, 2.0, 0.0] window = optimization.TimeWindow(0.5, 1.5) expected = { "final": 1.0, "minimum": 1.0, "maximum": 2.0, "timeMean": 1.5, "rms": math.sqrt(2.5), "integral": 1.5, "absoluteIntegral": 1.5, "peakAbsolute": 2.0, } for kind, expected_value in expected.items(): with self.subTest(kind=kind): actual = optimization.statistic_value( times, values, optimization.StatisticSpec(kind, window), ) self.assertAlmostEqual(actual, expected_value) def test_statistic_rejects_uncovered_windows_and_nonmonotonic_time(self) -> None: with self.assertRaises(optimization.OptimizationError) as caught: optimization.statistic_value( [0.0, 1.0], [1.0, 2.0], optimization.StatisticSpec( "final", optimization.TimeWindow(-0.1, 0.5) ), ) self.assertEqual(caught.exception.code, "OPTIMIZATION_WINDOW_NOT_COVERED") with self.assertRaises(optimization.OptimizationError) as caught: optimization.statistic_value( [0.0, 1.0, 1.0], [1.0, 2.0, 3.0], optimization.StatisticSpec("maximum", None), ) self.assertEqual(caught.exception.code, "OPTIMIZATION_RESULT_TIME_INVALID") def test_completed_metrics_reject_result_unit_drift(self) -> None: spec = optimization.parse_optimization_spec(_valid_spec()) variables = copy.deepcopy(RESULT_VARIABLES) variables[0]["unit"] = "cm" result = { "status": "completed", "success": True, "variables": variables, "series": { "time": [0.0, 1.0], "sensor.output": [1.0, 1.0], "sensor.limit": [1.0, 1.0], }, } with self.assertRaises(optimization.OptimizationError) as caught: optimization._completed_metrics(result, spec) self.assertEqual( caught.exception.code, "OPTIMIZATION_RESULT_METADATA_MISMATCH" ) self.assertEqual(caught.exception.details["expected"], "m") self.assertEqual(caught.exception.details["received"], "cm") def test_metric_overflow_is_a_structured_error(self) -> None: with self.assertRaises(optimization.OptimizationError) as caught: optimization.statistic_value( [0.0, 1.0], [1e308, 1e308], optimization.StatisticSpec("rms", None), ) self.assertEqual(caught.exception.code, "OPTIMIZATION_METRIC_OVERFLOW") def test_signed_integral_fsum_value_error_is_structured_overflow(self) -> None: with self.assertRaises(optimization.OptimizationError) as caught: optimization.statistic_value( [0.0, 1.0, 2.0, 3.0], [1e308, 1e308, -1e308, -1e308], optimization.StatisticSpec("integral", None), ) self.assertEqual(caught.exception.code, "OPTIMIZATION_METRIC_OVERFLOW") def test_integral_metrics_report_series_unit_times_seconds(self) -> None: payload = _valid_spec() objective = payload["objective"] constraints = payload["constraints"] assert isinstance(objective, dict) assert isinstance(constraints, list) and isinstance(constraints[0], dict) objective["statistic"] = {"kind": "integral", "window": None} constraints[0]["statistic"] = { "kind": "absoluteIntegral", "window": None, } spec = optimization.parse_optimization_spec(payload) result = { "status": "completed", "success": True, "variables": copy.deepcopy(RESULT_VARIABLES), "series": { "time": [0.0, 1.0], "sensor.output": [1.0, 1.0], "sensor.limit": [2.0, 2.0], }, } metrics = optimization._completed_metrics(result, spec) self.assertEqual(spec.objective.as_dict()["metricUnit"], "(m)*s") self.assertEqual(metrics[2][0]["seriesUnit"], "m") self.assertEqual(metrics[2][0]["unit"], "(m)*s") class ObjectiveEndpointTrendTests(unittest.TestCase): @staticmethod def _objective(*, statistic_kind: str = "final") -> optimization.ObjectiveSpec: payload = _valid_spec() objective = payload["objective"] assert isinstance(objective, dict) statistic = objective["statistic"] assert isinstance(statistic, dict) statistic["kind"] = statistic_kind return optimization.parse_optimization_spec(payload).objective def test_final_objective_detects_material_terminal_change(self) -> None: times = [float(index) for index in range(101)] values = [100.0] * 95 + [100.0, 98.0, 96.0, 94.0, 92.0, 90.0] trend = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": {"time": times, "sensor.output": values}, }, self._objective(), expected_end=100.0, ) self.assertTrue(trend["applicable"]) self.assertEqual(trend["status"], "materialChangeDetected") self.assertTrue(trend["materialChangeDetected"]) self.assertEqual(trend["direction"], "decreasing") self.assertFalse(trend["steadyStateProven"]) def test_final_objective_reports_no_material_terminal_change(self) -> None: times = [float(index) for index in range(101)] values = [100.0] * len(times) trend = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": {"time": times, "sensor.output": values}, }, self._objective(), expected_end=100.0, ) self.assertTrue(trend["applicable"]) self.assertEqual(trend["status"], "noMaterialChangeDetected") self.assertFalse(trend["materialChangeDetected"]) self.assertEqual(trend["direction"], "flat") self.assertFalse(trend["steadyStateProven"]) def test_final_objective_labels_range_only_signal_as_fluctuation(self) -> None: times = [float(index) for index in range(101)] values = [100.0] * 95 + [100.0, 104.0, 96.0, 104.0, 96.0, 100.0] trend = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": {"time": times, "sensor.output": values}, }, self._objective(), expected_end=100.0, ) self.assertEqual(trend["status"], "materialChangeDetected") self.assertEqual(trend["direction"], "fluctuating") self.assertFalse(trend["directionalChangeDetected"]) self.assertTrue(trend["tailVariabilityDetected"]) self.assertEqual(trend["detectionReasons"], ["tailVariability"]) self.assertFalse(trend["steadyStateProven"]) def test_flat_tail_followed_by_step_is_directional_not_fluctuating(self) -> None: times = [float(index) for index in range(101)] values = [100.0] * 100 + [103.0] trend = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": {"time": times, "sensor.output": values}, }, self._objective(), expected_end=100.0, ) self.assertEqual(trend["status"], "materialChangeDetected") self.assertEqual(trend["direction"], "increasing") self.assertTrue(trend["directionalChangeDetected"]) self.assertEqual(trend["directionalConsistency"], 1.0) self.assertEqual(trend["nonzeroIncrementCount"], 1) def test_final_objective_reports_insufficient_terminal_samples(self) -> None: trend = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": { "time": [0.0, 1.0, 2.0, 3.0, 4.0], "sensor.output": [5.0, 4.0, 3.0, 2.0, 1.0], } }, self._objective(), expected_end=4.0, ) self.assertTrue(trend["applicable"]) self.assertEqual(trend["status"], "insufficientData") self.assertEqual(trend["availableSamples"], 5) self.assertFalse(trend["steadyStateProven"]) def test_non_final_objective_is_not_applicable(self) -> None: trend = optimization._objective_endpoint_trend( {}, self._objective(statistic_kind="maximum"), ) self.assertFalse(trend["applicable"]) self.assertEqual(trend["status"], "notApplicable") self.assertFalse(trend["steadyStateProven"]) def test_incomplete_or_short_of_planned_endpoint_is_unavailable(self) -> None: incomplete = optimization._objective_endpoint_trend( { "status": "stopped", "success": False, "series": { "time": [0.0, 0.5], "sensor.output": [2.0, 1.0], }, }, self._objective(), expected_end=1.0, ) short = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": { "time": [index / 10 for index in range(10)], "sensor.output": [1.0] * 10, }, }, self._objective(), expected_end=1.0, ) self.assertEqual(incomplete["status"], "unavailable") self.assertEqual(incomplete["reason"], "freshVerificationNotCompleted") self.assertEqual(short["status"], "unavailable") self.assertEqual(short["reason"], "plannedEndpointNotCovered") self.assertFalse(incomplete["steadyStateProven"]) self.assertFalse(short["steadyStateProven"]) def test_terminal_trend_numeric_overflow_degrades_to_unavailable(self) -> None: times = [float(index) for index in range(101)] values = [1e308] * 95 + [1e308, -1e308, 1e308, -1e308, 1e308, -1e308] trend = optimization._objective_endpoint_trend( { "status": "completed", "success": True, "series": {"time": times, "sensor.output": values}, }, self._objective(), expected_end=100.0, ) self.assertEqual(trend["status"], "unavailable") self.assertEqual(trend["reason"], "terminalTrendNumericOverflow") self.assertFalse(trend["steadyStateProven"]) class ReportSafetyTests(unittest.TestCase): def test_report_escapes_markdown_link_and_image_delimiters(self) -> None: with tempfile.TemporaryDirectory() as directory_text: plan = _make_plan(Path(directory_text)) injected_unit = "![x](https://example.invalid/pixel)" objective = dataclasses.replace( plan.spec.objective, expected_unit=injected_unit, ) plan = dataclasses.replace( plan, spec=dataclasses.replace(plan.spec, objective=objective), ) runner = optimization.OptimizationRunner(plan, "report-safety-test") report = runner._report_markdown( { "solutionStatus": "noFeasibleCandidate", "terminationReason": "simulationBudgetExhausted", "counts": {}, } ) self.assertNotIn(injected_unit, report) self.assertNotRegex(report, r"!\[[^]]*\]\([^)]*\)") self.assertNotRegex(report, r"\[[^]]*\]\([^)]*\)") class ParameterContractTests(unittest.TestCase): def test_user_selected_si_parameter_resolves_with_source_provenance(self) -> None: spec = optimization.parse_optimization_spec(_valid_spec()) resolved = optimization._resolve_design_variables(spec, _inspection()) self.assertEqual(len(resolved), 1) self.assertEqual(resolved[0].initial, 2.0) self.assertEqual(resolved[0].spec.unit, "m") self.assertTrue(resolved[0].was_explicit) self.assertEqual(resolved[0].original_value, "1 + 1") self.assertIsNone(resolved[0].catalog_optimization_eligible) def test_discrete_metadata_unit_and_bounds_are_authoritative(self) -> None: cases = [ ( "explicit backend opt-out", _valid_spec(), _inspection(optimization_eligible=False), "OPTIMIZATION_PARAMETER_INELIGIBLE", ), ( "editor", _valid_spec(), _inspection(editor="choice"), "OPTIMIZATION_PARAMETER_DISCRETE_METADATA", ), ( "discrete options", _valid_spec(), _inspection(options=[{"value": 1.0, "label": "one"}]), "OPTIMIZATION_PARAMETER_DISCRETE_METADATA", ), ( "unit", _valid_spec(), _inspection(unit="cm"), "OPTIMIZATION_PARAMETER_UNIT_MISMATCH", ), ( "initial", _valid_spec(), _inspection(value=5.0, maximum=10.0), "OPTIMIZATION_INITIAL_OUTSIDE_BOUNDS", ), ( "registered minimum", _valid_spec(lower=-0.1), _inspection(), "OPTIMIZATION_BOUNDS_OUTSIDE_CONTRACT", ), ( "exclusive minimum", _valid_spec(lower=0.0), _inspection(minimum_exclusive=True), "OPTIMIZATION_BOUNDS_OUTSIDE_CONTRACT", ), ( "registered maximum", _valid_spec(upper=4.1), _inspection(maximum=4.0), "OPTIMIZATION_BOUNDS_OUTSIDE_CONTRACT", ), ] for label, payload, inspection, expected_code in cases: with self.subTest(label=label): spec = optimization.parse_optimization_spec(payload) with self.assertRaises(optimization.simulation.InputError) as caught: optimization._resolve_design_variables(spec, inspection) self.assertEqual(caught.exception.code, expected_code) def test_optional_backend_eligibility_contract_must_be_boolean(self) -> None: spec = optimization.parse_optimization_spec(_valid_spec()) malformed_inspections = [ _inspection(optimization_eligible="yes"), _inspection(optimization_eligible=None), _inspection(editor=""), _inspection(editor=1), _inspection(options=[]), _inspection(options="choice"), ] for inspection in malformed_inspections: with self.subTest(inspection=inspection): with self.assertRaises(optimization.simulation.BackendError) as caught: optimization._resolve_design_variables(spec, inspection) self.assertEqual( caught.exception.code, "BACKEND_PARAMETER_CONTRACT_INVALID", ) def test_optional_backend_eligibility_true_is_recorded(self) -> None: spec = optimization.parse_optimization_spec(_valid_spec()) resolved = optimization._resolve_design_variables( spec, _inspection(optimization_eligible=True), ) self.assertIs(resolved[0].catalog_optimization_eligible, True) class CandidateIsolationTests(unittest.TestCase): def test_reflection_maps_any_finite_coordinate_into_unit_interval(self) -> None: cases = ( (-2.25, 0.25), (-1.25, 0.75), (-0.25, 0.25), (0.0, 0.0), (0.25, 0.25), (1.0, 1.0), (1.25, 0.75), (2.0, 0.0), (2.25, 0.25), ) for coordinate, expected in cases: with self.subTest(coordinate=coordinate): self.assertEqual( optimization._reflect_unit_interval(coordinate), expected, ) for coordinate in (math.inf, -math.inf, math.nan): with self.subTest(coordinate=coordinate): with self.assertRaises(AssertionError): optimization._reflect_unit_interval(coordinate) def test_candidates_only_change_whitelisted_parameters_and_never_source(self) -> None: with tempfile.TemporaryDirectory() as directory_text: directory = Path(directory_text) plan = _make_plan(directory) original_project = copy.deepcopy(plan.source.parsed) original_source_bytes = plan.source.path.read_bytes() original_xml_values = _xml_parameter_values(plan.baseline_xml) first_project = optimization._optimized_project(plan, {"gain": 3.5}) second_project = optimization._optimized_project(plan, {"gain": 0.25}) expected_first = copy.deepcopy(original_project) expected_second = copy.deepcopy(original_project) expected_first["nodes"][0]["data"]["parameters"]["gain"] = 3.5 expected_second["nodes"][0]["data"]["parameters"]["gain"] = 0.25 self.assertEqual(first_project, expected_first) self.assertEqual(second_project, expected_second) self.assertEqual(plan.source.parsed, original_project) self.assertEqual(plan.source.path.read_bytes(), original_source_bytes) first_xml_values = _xml_parameter_values( optimization._candidate_xml(plan, {"gain": 3.5}) ) second_xml_values = _xml_parameter_values( optimization._candidate_xml(plan, {"gain": 0.25}) ) self.assertEqual(first_xml_values[("component-a", "gain")], "3.5") self.assertEqual(second_xml_values[("component-a", "gain")], "0.25") for target in ( ("component-a", "fixed"), ("component-b", "gain"), ): self.assertEqual(first_xml_values[target], original_xml_values[target]) self.assertEqual(second_xml_values[target], original_xml_values[target]) self.assertEqual(plan.baseline_xml, BASELINE_XML) class RankingAndConfirmationTests(unittest.TestCase): def test_plan_warns_when_budget_cannot_cover_a_complete_generation(self) -> None: with tempfile.TemporaryDirectory() as directory_text: minimal = _make_plan(Path(directory_text) / "minimal", max_simulation_runs=5) one_generation = _make_plan( Path(directory_text) / "one-generation", max_simulation_runs=9 ) minimal_public = minimal.public_dict() one_generation_public = one_generation.public_dict() self.assertEqual( minimal_public["execution"]["fullGenerationsWithUniqueCandidates"], 0, ) self.assertEqual( [warning["code"] for warning in minimal_public["warnings"]], ["OPTIMIZATION_BUDGET_INITIAL_POPULATION_ONLY"], ) self.assertEqual( minimal_public["parameterContinuity"][ "userAssertionRequiredFor" ], ["gain"], ) self.assertEqual( minimal_public["designVariables"][0]["continuity"], { "machineVerified": False, "userAssertionRequired": True, "catalogOptimizationEligible": None, "editorAbsent": True, "optionsAbsent": True, }, ) self.assertEqual( minimal_public["requiredAssertions"][0]["code"], "OPTIMIZATION_CONTINUITY_USER_ASSERTION", ) self.assertEqual( one_generation_public["execution"][ "fullGenerationsWithUniqueCandidates" ], 1, ) self.assertEqual(one_generation_public["warnings"], []) def test_confirmation_token_binds_all_plan_hash_inputs(self) -> None: arguments = { "source_sha256": "source-hash", "spec_sha256": "spec-hash", "baseline_xml_sha256": "xml-hash", "output_directory": Path("/tmp/optimization-output"), "base_url": BASE_URL, "timeout": 10.0, } token = optimization._confirmation_token(**arguments) self.assertEqual(len(token), 64) variants = [ {**arguments, "source_sha256": "changed-source"}, {**arguments, "spec_sha256": "changed-spec"}, {**arguments, "baseline_xml_sha256": "changed-xml"}, { **arguments, "output_directory": Path("/tmp/different-optimization-output"), }, {**arguments, "base_url": "http://localhost:8000"}, {**arguments, "timeout": 11.0}, ] for changed in variants: with self.subTest(changed=changed): self.assertNotEqual( optimization._confirmation_token(**changed), token ) with mock.patch.object( optimization, "CONTINUITY_POLICY", "different-continuity-policy", ): self.assertNotEqual( optimization._confirmation_token(**arguments), token, ) with mock.patch.object( optimization, "SEARCH_POLICY", "different-search-policy", ): self.assertNotEqual( optimization._confirmation_token(**arguments), token, ) def test_failed_or_infeasible_trial_cannot_beat_a_feasible_trial(self) -> None: feasible = optimization.Trial( evaluation_id=3, stage="search", simulation_id="opt.0003", parameters={"gain": 2.0}, status="completed", duration_seconds=0.1, objective_value=1000.0, objective_loss=1000.0, feasible=True, total_constraint_violation=0.0, ) infeasible = optimization.Trial( evaluation_id=2, stage="search", simulation_id="opt.0002", parameters={"gain": 1.0}, status="completed", duration_seconds=0.1, objective_value=-100.0, objective_loss=-100.0, feasible=False, total_constraint_violation=0.01, ) failed = optimization.Trial( evaluation_id=1, stage="search", simulation_id="opt.0001", parameters={"gain": 0.0}, status="failed", duration_seconds=0.1, objective_value=-10000.0, objective_loss=-10000.0, feasible=True, total_constraint_violation=0.0, ) self.assertLess(optimization._trial_rank(feasible), optimization._trial_rank(infeasible)) self.assertLess(optimization._trial_rank(feasible), optimization._trial_rank(failed)) with tempfile.TemporaryDirectory() as directory_text: runner = optimization.OptimizationRunner( _make_plan(Path(directory_text)), "ranking-test" ) runner.trials.extend([failed, infeasible, feasible]) self.assertIs(runner._best_trial(feasible_only=True), feasible) self.assertIs(runner._best_trial(feasible_only=False), feasible) def test_optimize_rejects_a_stale_confirmation_token_before_runner_starts(self) -> None: with tempfile.TemporaryDirectory() as directory_text: plan = _make_plan(Path(directory_text)) args = argparse.Namespace( confirmed=True, input=str(plan.source.path), spec=str(plan.spec_source.path), output_dir=str(plan.output_directory), base_url=plan.base_url, timeout=plan.timeout, expected_source_sha256=plan.source.sha256, expected_spec_sha256=plan.spec_source.sha256, confirmation_token="stale-token", optimization_id="must-not-start", ) with mock.patch.object( optimization, "build_runtime_plan", return_value=plan ), mock.patch.object(optimization, "OptimizationRunner") as runner_class: with self.assertRaises(optimization.simulation.InputError) as caught: optimization.command_optimize(args) self.assertEqual(caught.exception.code, "OPTIMIZATION_CONFIRMATION_STALE") self.assertIn("confirmationToken", caught.exception.details) runner_class.assert_not_called() class DeterministicRunnerTests(unittest.TestCase): @staticmethod def _run_mock_stagnating_search( directory: Path, population: list[list[float]], ) -> tuple[ optimization.OptimizationRunner, str, int, list[dict[str, object]], ]: plan = _make_plan(directory, max_simulation_runs=6) runner = optimization.OptimizationRunner(plan, "stagnation-test") initial_trials: list[optimization.Trial] = [] fresh_trials: dict[tuple[float, ...], optimization.Trial] = {} events: list[dict[str, object]] = [] def evaluate( normalized_vector: list[float], *, stage: str, **_kwargs: object, ) -> tuple[optimization.Trial, None]: if stage != "search": raise AssertionError("The search fixture only accepts search trials.") proposal_index = runner.search_proposals runner.optimizer_calls += 1 runner.search_proposals += 1 if proposal_index < len(population): key = tuple(normalized_vector) trial = fresh_trials.get(key) if trial is None: runner.backend_submissions += 1 runner.search_backend_submissions += 1 parameter = optimization._candidate_parameters( plan, normalized_vector )["gain"] trial = optimization.Trial( evaluation_id=runner.backend_submissions, stage="search", simulation_id=( f"stagnation-test.{runner.backend_submissions:04d}" ), parameters={"gain": parameter}, status="completed", duration_seconds=0.0, objective_value=0.0, objective_loss=0.0, feasible=True, total_constraint_violation=0.0, ) fresh_trials[key] = trial runner.trials.append(trial) else: runner.cache_hits += 1 initial_trials.append(trial) return trial, None target_index = (proposal_index - len(population)) % len(population) runner.cache_hits += 1 return initial_trials[target_index], None with mock.patch.object( runner, "_initial_population", return_value=copy.deepcopy(population), ), mock.patch.object( runner, "_evaluate", side_effect=evaluate, ), mock.patch.object( runner, "_checkpoint", ), mock.patch.object( runner, "_emit", side_effect=lambda payload: events.append(dict(payload)), ): reason, generations = runner._search() return runner, reason, generations, events def test_one_dimensional_initial_population_contains_baseline_and_endpoints( self, ) -> None: cases = ( ("interior", 0.0, 4.0), ("baseline-at-lower", 2.0, 4.0), ("baseline-at-upper", 0.0, 2.0), ) with tempfile.TemporaryDirectory() as directory_text: root = Path(directory_text) for label, lower, upper in cases: with self.subTest(label=label): plan = _make_plan( root / label, lower=lower, upper=upper, ) runner = optimization.OptimizationRunner(plan, label) population = runner._initial_population( optimization.random.Random(2026) ) self.assertEqual( population[0], optimization._normalized_initial(plan), ) self.assertEqual( len(population), plan.spec.algorithm.population_size, ) self.assertIn([0.0], population) self.assertIn([1.0], population) mapped = [ optimization._candidate_parameters(plan, vector)["gain"] for vector in population ] self.assertIn(lower, mapped) self.assertIn(upper, mapped) self.assertEqual( len({tuple(vector) for vector in population}), len(population), ) def test_de_generation_uses_a_frozen_donor_population(self) -> None: class FixedRandom: def __init__(self, _seed: int) -> None: pass @staticmethod def sample(values: list[int], count: int) -> list[int]: return values[:count] @staticmethod def randrange(_stop: int) -> int: return 0 @staticmethod def random() -> float: return 0.0 with tempfile.TemporaryDirectory() as directory_text: plan = _make_plan(Path(directory_text), max_simulation_runs=9) runner = optimization.OptimizationRunner(plan, "deferred-update-test") population = [[0.1], [0.2], [0.3], [0.4]] candidates: list[list[float]] = [] def evaluate( normalized_vector: list[float], *, stage: str, **_kwargs: object, ) -> tuple[optimization.Trial, None]: self.assertEqual(stage, "search") runner.optimizer_calls += 1 runner.search_proposals += 1 runner.backend_submissions += 1 runner.search_backend_submissions += 1 evaluation_id = runner.backend_submissions if evaluation_id > len(population): candidates.append(list(normalized_vector)) loss = 100.0 + evaluation_id if evaluation_id <= 4 else 0.0 trial = optimization.Trial( evaluation_id=evaluation_id, stage="search", simulation_id=f"deferred-update-test.{evaluation_id:04d}", parameters=optimization._candidate_parameters( plan, normalized_vector ), status="completed", duration_seconds=0.0, objective_value=loss, objective_loss=loss, feasible=True, total_constraint_violation=0.0, ) runner.trials.append(trial) return trial, None with mock.patch.object( runner, "_initial_population", return_value=copy.deepcopy(population), ), mock.patch.object( runner, "_evaluate", side_effect=evaluate, ), mock.patch.object( runner, "_checkpoint", ), mock.patch.object( runner, "_emit", ), mock.patch.object( optimization.random, "Random", FixedRandom, ): reason, generations = runner._search() self.assertEqual(reason, "simulationBudgetExhausted") self.assertEqual(generations, 1) self.assertEqual(len(candidates), 4) self.assertAlmostEqual(candidates[0][0], 0.12) self.assertAlmostEqual(candidates[1][0], 0.02) def test_three_empty_generations_report_population_collapse_as_stagnation( self, ) -> None: with tempfile.TemporaryDirectory() as directory_text: runner, reason, generations, events = self._run_mock_stagnating_search( Path(directory_text), [[0.5], [0.5], [0.5], [0.5]], ) self.assertEqual( reason, "populationCollapsedAfterDuplicateStagnation", ) self.assertEqual( generations, optimization.NO_NEW_SUBMISSION_GENERATION_LIMIT, ) counts = runner._counts_payload() self.assertEqual(counts["searchBackendSubmissions"], 1) self.assertEqual(counts["verificationBackendSubmissions"], 0) self.assertEqual(counts["cacheHits"], 15) self.assertEqual(counts["generationsWithNewBackendSubmissions"], 0) self.assertEqual(counts["generationsWithoutNewBackendSubmissions"], 3) self.assertEqual(counts["endingNoNewSubmissionGenerationStreak"], 3) self.assertEqual(counts["finalPopulationUniqueCandidates"], 1) summary = runner._search_summary(reason) self.assertFalse(summary["searchConvergenceEstablished"]) self.assertTrue(summary["populationCollapsedToSingleCandidate"]) self.assertEqual(summary["terminationCategory"], "stagnation") generation_events = [ event for event in events if event.get("event") == "optimization-generation-completed" ] self.assertEqual(len(generation_events), 3) self.assertEqual( [ event["backendSubmissionsThisGeneration"] for event in generation_events ], [0, 0, 0], ) self.assertEqual( [ event["consecutiveGenerationsWithoutNewBackendSubmissions"] for event in generation_events ], [1, 2, 3], ) def test_three_empty_generations_report_duplicate_proposal_stagnation( self, ) -> None: with tempfile.TemporaryDirectory() as directory_text: runner, reason, generations, _events = ( self._run_mock_stagnating_search( Path(directory_text), [[0.0], [0.25], [0.5], [1.0]], ) ) self.assertEqual(reason, "duplicateProposalStagnation") self.assertEqual( generations, optimization.NO_NEW_SUBMISSION_GENERATION_LIMIT, ) counts = runner._counts_payload() self.assertEqual(counts["searchBackendSubmissions"], 4) self.assertEqual(counts["verificationBackendSubmissions"], 0) self.assertEqual(counts["cacheHits"], 12) self.assertEqual(counts["generationsWithNewBackendSubmissions"], 0) self.assertEqual(counts["generationsWithoutNewBackendSubmissions"], 3) self.assertEqual(counts["maxNoNewSubmissionGenerationStreak"], 3) self.assertEqual(counts["finalPopulationUniqueCandidates"], 4) summary = runner._search_summary(reason) self.assertFalse(summary["searchConvergenceEstablished"]) self.assertFalse(summary["populationCollapsedToSingleCandidate"]) self.assertEqual( summary["terminationCategory"], "stagnation", ) @staticmethod def _run_mock_optimization( directory: Path, *, seed: int, change_source_after_artifacts: bool = False, force_infeasible: bool = False, ) -> tuple[dict[str, object], list[dict[str, object]], optimization.RuntimePlan]: plan = _make_plan(directory, seed=seed, max_simulation_runs=6) runner = optimization.OptimizationRunner(plan, "deterministic-test") submissions: list[dict[str, object]] = [] def simulation_result( _base_url: str, xml: bytes, simulation_id: str, _timeout: float, _progress_path: Path, **kwargs: object, ) -> tuple[dict[str, object], None]: parameter_values = _xml_parameter_values(xml) gain = float(parameter_values[("component-a", "gain")]) submissions.append( {"simulationId": simulation_id, "gain": gain, "xml": xml} ) event_sink = kwargs.get("event_sink") if callable(event_sink): event_sink({"event": "progress", "progress": 0.5}) objective = (gain - 0.3) ** 2 return ( { "status": "completed", "success": True, "variables": copy.deepcopy(RESULT_VARIABLES), "series": { "time": [0.0, 1.0], "sensor.output": [objective, objective], "sensor.limit": ( [10.0, 10.0] if force_infeasible else [gain, gain] ), }, }, None, ) original_write_best_artifacts = runner._write_best_artifacts def write_best_artifacts_then_change_source( *args: object, **kwargs: object ) -> dict[str, object]: artifacts = original_write_best_artifacts(*args, **kwargs) plan.source.path.write_bytes(plan.source.raw + b"\n") return artifacts artifact_context = ( mock.patch.object( runner, "_write_best_artifacts", side_effect=write_best_artifacts_then_change_source, ) if change_source_after_artifacts else contextlib.nullcontext() ) with mock.patch.object( optimization.simulation, "_read_simulation_stream", side_effect=simulation_result, ), mock.patch.object( optimization.simulation, "_download_csv", return_value=b"time,sensor.output,sensor.limit\n0,0,0\n", ), mock.patch.object(optimization.simulation, "emit_json"), artifact_context: result = runner.run() return result, submissions, plan def test_de_seed_budget_fresh_verification_and_artifacts(self) -> None: with tempfile.TemporaryDirectory() as directory_text: root = Path(directory_text) first, first_submissions, first_plan = self._run_mock_optimization( root / "first", seed=8675309 ) second, second_submissions, _ = self._run_mock_optimization( root / "second", seed=8675309 ) first_search = [ trial["parameters"] for trial in first["trials"] if trial["stage"] == "search" ] second_search = [ trial["parameters"] for trial in second["trials"] if trial["stage"] == "search" ] self.assertEqual(first_search, second_search) self.assertEqual( [item["gain"] for item in first_submissions[:-1]], [item["gain"] for item in second_submissions[:-1]], ) self.assertEqual(first["optimizationResultSchemaVersion"], 2) self.assertEqual(first["solutionStatus"], "verified") self.assertIn("proof", first["claim"]) self.assertIn("global optimality", first["claim"]) self.assertEqual( first["objectiveEndpointTrend"]["status"], "insufficientData", ) self.assertFalse( first["objectiveEndpointTrend"]["steadyStateProven"] ) self.assertEqual(first["planHash"], first_plan.confirmation_token) self.assertEqual( first["continuityPolicy"]["id"], optimization.CONTINUITY_POLICY, ) self.assertTrue( first["continuityPolicy"]["confirmedUserAssertion"] ) self.assertEqual( first["continuityPolicy"]["designVariableIds"], ["gain"], ) self.assertEqual(first["terminationReason"], "simulationBudgetExhausted") self.assertEqual(first["counts"]["generations"], 0) self.assertEqual(first["counts"]["backendSubmissions"], 6) self.assertEqual( first["counts"]["backendSubmissionSlotsConsumed"], 6, ) self.assertEqual(first["counts"]["searchBackendSubmissions"], 5) self.assertEqual( first["counts"]["searchSubmissionSlotsConsumed"], 5, ) self.assertEqual( first["counts"]["verificationBackendSubmissions"], 1, ) self.assertEqual( first["counts"]["cacheHits"], first["search"]["cacheHits"], ) self.assertEqual(first["search"]["backendSubmissions"], 5) self.assertEqual(first["search"]["budget"]["used"], 5) self.assertTrue(first["search"]["budget"]["exhausted"]) self.assertEqual(first["verification"]["backendSubmissions"], 1) self.assertEqual( first["verification"]["submissionSlotsConsumed"], 1, ) self.assertEqual(len(first_submissions), 6) self.assertEqual( sum(trial["stage"] == "search" for trial in first["trials"]), 5, ) self.assertEqual(first["trials"][-1]["stage"], "verification") self.assertTrue(first["verification"]["passed"]) self.assertTrue(first["verification"]["sourceUnchanged"]) self.assertEqual( first_submissions[-1]["gain"], first["bestSearch"]["parameters"]["gain"], ) self.assertEqual( first["verifiedBest"]["parameters"], first["bestSearch"]["parameters"], ) self.assertTrue(first["source"]["unchanged"]) self.assertEqual( first_plan.source.path.read_bytes(), first_plan.source.raw ) expected_artifacts = { "plan", "bestParameters", "bestSystemXml", "bestProject", "result", "csv", "charts", "report", "evaluations", "events", "checkpoint", } self.assertEqual(set(first["artifacts"]), expected_artifacts) for name, artifact in first["artifacts"].items(): artifacts = artifact if name == "charts" else [artifact] for item in artifacts: path = Path(item["path"]) data = path.read_bytes() self.assertEqual(item["sizeBytes"], len(data)) self.assertEqual(item["sha256"], hashlib.sha256(data).hexdigest()) for filename in ( "optimization-plan.json", "optimization-result.json", "report.md", "best-parameters.json", "best-project.json", "best-system.xml", "result.json", "results.csv", ): self.assertTrue((first_plan.output_directory / filename).is_file()) optimized_project = json.loads( (first_plan.output_directory / "best-project.json").read_text( encoding="utf-8" ) ) expected_project = copy.deepcopy(PROJECT) expected_project["nodes"][0]["data"]["parameters"]["gain"] = first[ "verifiedBest" ]["parameters"]["gain"] self.assertEqual(optimized_project, expected_project) def test_source_change_during_fresh_verification_invalidates_solution(self) -> None: with tempfile.TemporaryDirectory() as directory_text: root = Path(directory_text) plan = _make_plan(root, seed=42, max_simulation_runs=5) runner = optimization.OptimizationRunner(plan, "source-race-test") submissions = 0 def simulation_result( _base_url: str, xml: bytes, _simulation_id: str, _timeout: float, _progress_path: Path, **_kwargs: object, ) -> tuple[dict[str, object], None]: nonlocal submissions submissions += 1 gain = float(_xml_parameter_values(xml)[("component-a", "gain")]) if submissions == plan.spec.budget.max_simulation_runs: plan.source.path.write_bytes(plan.source.raw + b"\n") return ( { "status": "completed", "success": True, "variables": copy.deepcopy(RESULT_VARIABLES), "series": { "time": [0.0, 1.0], "sensor.output": [gain, gain], "sensor.limit": [gain, gain], }, }, None, ) with mock.patch.object( optimization.simulation, "_read_simulation_stream", side_effect=simulation_result, ), mock.patch.object(optimization.simulation, "emit_json"): result = runner.run() self.assertEqual(result["solutionStatus"], "verificationFailed") self.assertFalse(result["verification"]["passed"]) self.assertFalse(result["verification"]["sourceUnchanged"]) self.assertFalse(result["source"]["unchanged"]) self.assertFalse((plan.output_directory / "best-project.json").exists()) self.assertFalse((plan.output_directory / "best-system.xml").exists()) def test_source_change_while_writing_artifacts_removes_best_outputs(self) -> None: with tempfile.TemporaryDirectory() as directory_text: result, _submissions, plan = self._run_mock_optimization( Path(directory_text), seed=7, change_source_after_artifacts=True, ) self.assertEqual(result["solutionStatus"], "verificationFailed") self.assertFalse(result["verification"]["passed"]) self.assertFalse(result["source"]["unchanged"]) self.assertEqual(result["artifacts"].keys(), { "plan", "report", "evaluations", "events", "checkpoint", }) for filename in ( "best-parameters.json", "best-system.xml", "best-project.json", "result.json", "results.csv", ): self.assertFalse((plan.output_directory / filename).exists()) def test_late_best_artifact_export_failure_removes_all_best_outputs(self) -> None: failures = ( ("csv", "_download_csv", "CSV_EXPORT_FAILED"), ("charts", "_write_charts", "CHART_EXPORT_FAILED"), ) for label, failing_export, expected_code in failures: with self.subTest( export=label ), tempfile.TemporaryDirectory() as directory_text: plan = _make_plan(Path(directory_text), max_simulation_runs=5) plan.output_directory.mkdir() runner = optimization.OptimizationRunner(plan, "artifact-failure-test") trial = optimization.Trial( evaluation_id=5, stage="verification", simulation_id="artifact-failure-test.0005", parameters={"gain": 1.0}, status="completed", duration_seconds=0.1, objective_value=1.0, objective_loss=1.0, feasible=True, total_constraint_violation=0.0, ) result = { "status": "completed", "success": True, "variables": copy.deepcopy(RESULT_VARIABLES), "series": { "time": [0.0, 1.0], "sensor.output": [1.0, 1.0], "sensor.limit": [1.0, 1.0], }, } csv_patch = ( mock.patch.object( optimization.simulation, "_download_csv", side_effect=optimization.simulation.BackendError( "CSV_EXPORT_FAILED", "forced CSV export failure" ), ) if failing_export == "_download_csv" else mock.patch.object( optimization.simulation, "_download_csv", return_value=b"time,sensor.output,sensor.limit\n0,1,1\n", ) ) chart_patch = ( mock.patch.object( optimization.simulation, "_write_charts", side_effect=optimization.simulation.ArtifactError( "CHART_EXPORT_FAILED", "forced chart export failure" ), ) if failing_export == "_write_charts" else contextlib.nullcontext() ) with csv_patch, chart_patch, self.assertRaises( optimization.simulation.SkillCliError ) as caught: runner._write_best_artifacts(trial, result) self.assertEqual(caught.exception.code, expected_code) for filename in ( "best-parameters.json", "best-system.xml", "best-project.json", "result.json", "results.csv", ): self.assertFalse((plan.output_directory / filename).exists()) self.assertEqual( list(plan.output_directory.glob("curve-*.svg")), [] ) def test_no_feasible_candidate_reports_diagnostic_without_best_files(self) -> None: with tempfile.TemporaryDirectory() as directory_text: result, submissions, plan = self._run_mock_optimization( Path(directory_text), seed=9, force_infeasible=True ) self.assertEqual(result["solutionStatus"], "noFeasibleCandidate") self.assertIn("no complete feasible candidate", result["claim"]) self.assertEqual(result["counts"]["backendSubmissions"], 5) self.assertEqual(len(submissions), 5) self.assertIsNone(result["bestSearch"]) self.assertIsNotNone(result["bestDiagnosticSearch"]) self.assertFalse(result["bestDiagnosticSearch"]["feasible"]) self.assertIsNone(result["verifiedBest"]) self.assertNotIn("bestProject", result["artifacts"]) self.assertFalse((plan.output_directory / "best-project.json").exists()) report = (plan.output_directory / "report.md").read_text( encoding="utf-8" ) self.assertNotIn("搜索阶段最佳可行点", report) self.assertIn("约束违反最小的完整诊断点（不可行）", report) self.assertIn("它不是可行方案", report) def test_interrupted_stream_keeps_id_available_for_backend_cancel(self) -> None: with tempfile.TemporaryDirectory() as directory_text: runner = optimization.OptimizationRunner( _make_plan(Path(directory_text)), "cancel-test" ) with mock.patch.object( optimization.simulation, "_read_simulation_stream", side_effect=KeyboardInterrupt, ), mock.patch.object( optimization.simulation, "emit_json" ), mock.patch.object( optimization.simulation, "http_json" ) as http_json: with self.assertRaises(KeyboardInterrupt): runner._evaluate([0.5], stage="search") self.assertEqual(runner.current_simulation_id, "cancel-test.0001") runner.cancel_active() http_json.assert_called_once() self.assertIsNone(runner.current_simulation_id) def test_completed_but_unsuccessful_baseline_is_recorded_as_failed(self) -> None: with tempfile.TemporaryDirectory() as directory_text: plan = _make_plan(Path(directory_text), max_simulation_runs=5) runner = optimization.OptimizationRunner(plan, "failed-baseline-test") failed_result = { "status": "completed", "success": False, "message": "solver rejected the baseline", } with mock.patch.object( optimization.simulation, "_read_simulation_stream", return_value=(failed_result, None), ), mock.patch.object(optimization.simulation, "emit_json"): with self.assertRaises(optimization.OptimizationError) as caught: runner.run() self.assertEqual(caught.exception.code, "OPTIMIZATION_BASELINE_FAILED") self.assertEqual(len(runner.trials), 1) self.assertEqual(runner.trials[0].status, "failed") checkpoint = json.loads( (plan.output_directory / "checkpoint.json").read_text( encoding="utf-8" ) ) self.assertEqual(checkpoint["counts"]["completedTrials"], 0) self.assertEqual(checkpoint["counts"]["failedTrials"], 1) def test_completed_contract_error_is_recorded_before_abort(self) -> None: with tempfile.TemporaryDirectory() as directory_text: plan = _make_plan(Path(directory_text), max_simulation_runs=5) runner = optimization.OptimizationRunner(plan, "contract-error-test") variables = copy.deepcopy(RESULT_VARIABLES) variables[0]["unit"] = "cm" invalid_result = { "status": "completed", "success": True, "variables": variables, "series": { "time": [0.0, 1.0], "sensor.output": [1.0, 1.0], "sensor.limit": [1.0, 1.0], }, } with mock.patch.object( optimization.simulation, "_read_simulation_stream", return_value=(invalid_result, None), ), mock.patch.object(optimization.simulation, "emit_json"): with self.assertRaises(optimization.OptimizationError) as caught: runner.run() self.assertEqual( caught.exception.code, "OPTIMIZATION_RESULT_METADATA_MISMATCH" ) self.assertEqual(runner.backend_submissions, 1) self.assertEqual(len(runner.trials), 1) self.assertEqual( runner.trials[0].failure_code, "OPTIMIZATION_RESULT_METADATA_MISMATCH", ) evaluations = ( plan.output_directory / "evaluations.csv" ).read_text(encoding="utf-8") self.assertIn("OPTIMIZATION_RESULT_METADATA_MISMATCH", evaluations) if __name__ == "__main__": # pragma: no cover unittest.main()