from __future__ import annotations import os import tempfile import unittest from copy import deepcopy from pathlib import Path from unittest.mock import Mock, patch from cognitive_agent.application.native_can_runtime import NativeCanCompileError from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService from cognitive_agent.application.self_extension import ( SelfExtensionService, SimulationGateway, ) from cognitive_agent.infrastructure.can.gateway import CanFrame from cognitive_agent.testing.ch1_hil import ( ARM_ENV, CONFIRMATION_TEXT, EXPECTED_TARGET, TARGET_ENV, Ch1HilContractError, Ch1HilRunner, Ch1HilSafetyError, HttpResult, classify_failure, load_suite, observe_stationary_ch1, require_explicit_hil_arm, suite_cases, suite_rejections, validate_result_contract, validate_runtime, validate_stationary_snapshot, validate_suite, validate_ui_contract, ) SUITE_PATH = ( REPOSITORY_ROOT / "tests" / "fixtures" / "ch1_vehicle_frontend_hil_suite.json" ) DOOR_SMOKE_SUITE_PATH = ( REPOSITORY_ROOT / "tests" / "fixtures" / "ch1_vehicle_door_smoke_hil_suite.json" ) def stationary_snapshot() -> dict: return { "source": "physical_can_ch1", "channel": 1, "read_only": True, "gear": "P", "speed_kmh": 0, "evidence": { "gear_frame_id": "0x342", "speed_frame_id": "0x4EB", }, } def valid_runtime() -> dict: return { "hardware_execution_allowed": True, "vehicle_can": { "enabled": True, "configured": True, "channel": 1, "required_channel": 1, "channel_is_allowed": True, "backend": "zlgcan", "platform": "t2g_cabin_bench", "arbitration": {"bitrate": 500_000}, "data": {"bitrate": 2_000_000}, }, "multi_channel_can": {"vehicle_channel": 1}, } class ReceiveOnlyBus: def __init__(self, frames: list[CanFrame]) -> None: self.frames = list(frames) self.receive_calls: list[tuple[int, bool]] = [] def receive(self, timeout_ms: int, *, include_fd: bool = False) -> CanFrame | None: self.receive_calls.append((timeout_ms, include_fd)) return self.frames.pop(0) if self.frames else None class FailingApi: """In-memory API double; it has no CAN transport and cannot transmit frames.""" def __init__(self, successful_results: dict[str, dict] | None = None) -> None: self.calls: list[tuple[str, str, dict | None]] = [] self.successful_results = successful_results or {} def request( self, path: str, *, method: str = "GET", payload: dict | None = None, timeout_seconds: float = 10, ) -> HttpResult: del timeout_seconds self.calls.append((path, method, deepcopy(payload))) if path == "/api/runtime": return HttpResult(200, valid_runtime(), {}) if payload and payload.get("utterance") == "ac on": return HttpResult( 503, {"error": "gscu_unavailable", "message": "GSCU disconnected"}, {}, ) if payload and payload.get("utterance") in self.successful_results: return HttpResult( 200, deepcopy(self.successful_results[payload["utterance"]]), {}, ) return HttpResult(200, {"status": "succeeded"}, {}) @staticmethod def text(path: str, *, timeout_seconds: float = 10) -> str: del timeout_seconds if path == "/app.js": return " ".join( ( "/api/requests", "renderExecution", "feedback_frame", "hardware_busy", "physical_can_ch1", ) ) return " ".join( ( "/api/extension/", "operation_results", "physical_feedback_confirmed", "vehicleExecutionSucceeded", "cognitive-agent:extension-vehicle-result", ) ) class Ch1VehicleHilContractTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.temporary_directory = tempfile.TemporaryDirectory() with patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", }, clear=False, ): cls.service = SeatDemoService(REPOSITORY_ROOT) cls.service.artifact_root = Path(cls.temporary_directory.name) cls.suite = load_suite(SUITE_PATH) cls.door_smoke_suite = load_suite(DOOR_SMOKE_SUITE_PATH) @classmethod def tearDownClass(cls) -> None: cls.temporary_directory.cleanup() def _physical_result(self, case: dict) -> dict: result = self.service.process( case["command"], provider="local", conversation_id=f"conv-cleanup-{case['id'].lower()}", execution_mode="dry_run", vehicle_id=self.suite["vehicle_id"], ) result["status"] = "succeeded" feedback_id = case["expected"]["feedback_ids"][0] feedback_state = { "source": "physical_can_ch1", "channel": 1, "valid": True, "frame_id": feedback_id, "feedback_frame_id": feedback_id, } for state_key in ("domain_state", "actuator_state"): state = result.get(state_key) if isinstance(state, dict) and state: feedback_state[state_key] = deepcopy(state) result["execution_report"] = { "metadata": {"labels": {"seat_feedback": "physical_can_ch1"}}, "spec": { "mode": "hil", "status": "succeeded", "results": [ { "ref": "native-cleanup-operation", "status": "succeeded", "telemetry": { "channel": 1, "simulated": False, "feedback_required": True, "feedback_observed": True, "feedback_match": True, "feedback_frame": { "frame_id": feedback_id, "channel": 1, "payload_size": 8, "data_hex": "00 00 00 00 00 00 00 00", }, "feedback_state": feedback_state, }, } ], }, } return result def test_fixture_is_new_ch1_driver_only_suite(self) -> None: self.assertEqual(SUITE_PATH.name, "ch1_vehicle_frontend_hil_suite.json") validate_suite(self.suite) scope = self.suite["physical_scope"] self.assertEqual((scope["bus"], scope["channel"]), ("body_can", 1)) self.assertEqual(scope["seat_target"], "driver") self.assertFalse(scope["allow_ch0"]) self.assertFalse(scope["allow_iar"]) self.assertFalse(self.suite["execution_policy"]["live_negative_requests"]) orchestration = suite_cases(self.suite, "self_orchestration") extension = suite_cases(self.suite, "self_extension") self.assertEqual([case["id"] for case in orchestration], [case["id"] for case in extension]) self.assertEqual(len(orchestration), 43) self.assertEqual( len(suite_rejections(self.suite, "self_orchestration")), 14 ) self.assertEqual(sum(case.get("hil", True) for case in orchestration), 30) self.assertEqual(len(self.suite["cleanup"]), 9) self.assertTrue( all( not item["command"].startswith("door ") for item in self.suite["cleanup"] ) ) def test_door_smoke_fixture_is_inherited_rejection_only_contract( self, ) -> None: smoke = self.door_smoke_suite self.assertEqual(smoke["suite"], "ch1-vehicle-door-rejection-contract") self.assertTrue(smoke["rejection_only"]) self.assertEqual(smoke["physical_scope"], self.suite["physical_scope"]) self.assertEqual(smoke["execution_policy"], self.suite["execution_policy"]) self.assertEqual(smoke["case_catalog"], self.suite["case_catalog"]) self.assertEqual(smoke["rejection_catalog"], self.suite["rejection_catalog"]) self.assertEqual(smoke["cleanup"], self.suite["cleanup"]) self.assertEqual(len(smoke["cleanup"]), 9) for module in ("self_orchestration", "self_extension"): with self.subTest(module=module): self.assertEqual(suite_cases(smoke, module), []) self.assertEqual( [case["id"] for case in suite_rejections(smoke, module)], [case["id"] for case in self.suite["rejection_catalog"]], ) def test_retired_door_commands_are_only_rejections_and_cannot_enter_live_runner( self, ) -> None: retired = { "REJECT-VEHICLE-LOCK": "door vehicle lock", "REJECT-VEHICLE-UNLOCK": "door vehicle unlock", "REJECT-REAR-LOCK": "door rear lock", "REJECT-REAR-UNLOCK": "door rear unlock", } positive_commands = {case["command"] for case in self.suite["case_catalog"]} rejection_by_id = { case["id"]: case["command"] for case in self.suite["rejection_catalog"] } self.assertTrue(set(retired.values()).isdisjoint(positive_commands)) for rejection_id, command in retired.items(): self.assertEqual(rejection_by_id[rejection_id], command) api = FailingApi() with self.assertRaisesRegex(Ch1HilSafetyError, "rejection-only"): Ch1HilRunner( self.door_smoke_suite, api, stationary_snapshot, confirmation=CONFIRMATION_TEXT, ) self.assertEqual(api.calls, []) shrunk = deepcopy(self.door_smoke_suite) for rejection_id in retired: shrunk["rejection_catalog"] = [ case for case in shrunk["rejection_catalog"] if case["id"] != rejection_id ] for module in ("self_orchestration", "self_extension"): shrunk["modules"][module]["rejection_case_ids"].remove( rejection_id ) with self.assertRaisesRegex(Ch1HilSafetyError, "retired door command"): validate_suite(shrunk) injected = deepcopy(self.suite) retired_positive = deepcopy(injected["case_catalog"][0]) retired_positive.update( id="DOOR-RETIRED-INJECTED", command="door vehicle lock", ) injected["case_catalog"].append(retired_positive) for module in ("self_orchestration", "self_extension"): injected["modules"][module]["case_ids"].append( "DOOR-RETIRED-INJECTED" ) with self.assertRaisesRegex(Ch1HilSafetyError, "retired physical"): validate_suite(injected) def test_verified_body_can_matrix_is_pinned_in_fixture(self) -> None: by_id = {case["id"]: case for case in self.suite["case_catalog"]} expected_frames = { "LIGHT-INDOOR-ON": ["0x4F5"], "LIGHT-READING-ALL-ON": ["0x4F5"], "AMBIENT-POWER-ON": ["0x3FC"], "AMBIENT-COLOR-32": ["0x3FC"], "AMBIENT-BRIGHTNESS-3": ["0x3FC"], "WINDOW-FRONT-LEFT-50": ["0x431"], "SHADE-POSITION-10": ["0x431"], "SHADE-POSITION-0": ["0x431"], "SHADE-POSITION-50": ["0x431"], "SHADE-POSITION-100": ["0x431"], } for case_id, frame_ids in expected_frames.items(): with self.subTest(case=case_id): self.assertEqual(by_id[case_id]["expected"]["frame_ids"], frame_ids) contract_only = {case["id"] for case in by_id.values() if not case.get("hil", True)} self.assertIn("OUTLET-DRIVER-LEFT-CENTER", contract_only) self.assertEqual( len({case_id for case_id in contract_only if case_id.startswith("SEAT-DRIVER-")}), 12, ) def test_driver_seat_ch1_matrix_separates_verified_feedback_from_motion( self, ) -> None: seat_cases = [ case for case in self.suite["case_catalog"] if case["expected"]["domain"] == "seat" ] physical = {case["id"]: case for case in seat_cases if case.get("hil", True)} contract_only = { case["id"]: case for case in seat_cases if not case.get("hil", True) } self.assertEqual( set(physical), { "SEAT-DRIVER-VENT-LOW", "SEAT-DRIVER-VENT-HIGH", "SEAT-DRIVER-VENT-OFF", "SEAT-DRIVER-HEAT-LOW", "SEAT-DRIVER-HEAT-HIGH", "SEAT-DRIVER-HEAT-OFF", "SEAT-DRIVER-MASSAGE-WEAK", "SEAT-DRIVER-MASSAGE-RELAX", "SEAT-DRIVER-MASSAGE-OFF", }, ) self.assertEqual(len(contract_only), 12) self.assertEqual( { case["expected"]["capability"] for case in contract_only.values() }, {"seat.set_driver_axis_native"}, ) for case in seat_cases: with self.subTest(case=case["id"]): expected = case["expected"] self.assertEqual(expected["config_key"], "native_seat") self.assertEqual(expected["target"], "driver") self.assertNotIn("passenger", case["command"]) if case.get("hil", True): self.assertEqual(expected["feedback_ids"], ["0x4FA"]) self.assertTrue(expected["state_assertions"]) def test_negative_gate_matrix_is_semantically_pinned_for_both_modules( self, ) -> None: required = { "REJECT-CH0", "REJECT-PASSENGER", "REJECT-FRONT-RIGHT", "REJECT-T2G", "REJECT-IAR", "REJECT-RAW-CAN", "REJECT-VEHICLE-LOCK", "REJECT-VEHICLE-UNLOCK", "REJECT-REAR-LOCK", "REJECT-REAR-UNLOCK", "REJECT-WALL-LIGHT", "REJECT-READING-ZONE", "REJECT-AMBIENT-SCENE", "REJECT-AMBIENT-RGB", } for module in ("self_orchestration", "self_extension"): rejections = suite_rejections(self.suite, module) with self.subTest(module=module): self.assertEqual({case["id"] for case in rejections}, required) self.assertTrue( all( case.get("expected_category") == "safety_rejection" for case in rejections ) ) def test_suite_validation_rejects_module_and_catalog_coverage_holes( self, ) -> None: invalid_mutations = ( ( "physical case without UI state assertion", lambda fixture: fixture["case_catalog"][0]["expected"].update( state_assertions=[] ), ), ( "extension missing a positive case", lambda fixture: fixture["modules"]["self_extension"][ "case_ids" ].pop(), ), ( "duplicate positive case", lambda fixture: fixture["modules"]["self_extension"][ "case_ids" ].append(fixture["modules"]["self_extension"]["case_ids"][0]), ), ( "extension endpoint redirected", lambda fixture: fixture["modules"]["self_extension"].update( endpoint="/api/requests" ), ), ( "wrong rejection category", lambda fixture: fixture["rejection_catalog"][0].update( expected_category="unexpected_failure" ), ), ( "extension missing a rejection", lambda fixture: fixture["modules"]["self_extension"][ "rejection_case_ids" ].pop(), ), ) for label, mutate in invalid_mutations: fixture = deepcopy(self.suite) mutate(fixture) with self.subTest(invalid_coverage=label): with self.assertRaises((Ch1HilContractError, Ch1HilSafetyError)): validate_suite(fixture) def test_position_feedback_tolerances_are_pinned_in_fixture(self) -> None: by_id = {case["id"]: case for case in self.suite["case_catalog"]} expected_tolerances = { "WINDOW-FRONT-LEFT-10": 3, "WINDOW-FRONT-LEFT-50": 3, "WINDOW-FRONT-LEFT-0": 3, "SHADE-POSITION-10": 3, "SHADE-POSITION-50": 3, "SHADE-POSITION-100": 2, "SHADE-POSITION-0": 2, } for case_id, tolerance in expected_tolerances.items(): with self.subTest(case=case_id): assertions = by_id[case_id]["expected"]["state_assertions"] self.assertEqual(len(assertions), 1) self.assertEqual(assertions[0]["tolerance"], tolerance) def test_numeric_state_tolerance_is_inclusive_and_rejects_outside_values( self, ) -> None: by_id = {case["id"]: case for case in self.suite["case_catalog"]} checks = ( ( "WINDOW-FRONT-LEFT-10", "domain_state.window.positions.front_left.position_pct", 13, 14, ), ( "WINDOW-FRONT-LEFT-50", "domain_state.window.positions.front_left.position_pct", 53, 54, ), ("SHADE-POSITION-10", "domain_state.roof.shade_position_pct", 13, 14), ("SHADE-POSITION-50", "domain_state.roof.shade_position_pct", 53, 54), ("SHADE-POSITION-100", "domain_state.roof.shade_position_pct", 98, 97), ("SHADE-POSITION-0", "domain_state.roof.shade_position_pct", 2, 3), ) def set_path(payload: dict, path: str, value: int) -> None: parent = payload parts = path.split(".") for part in parts[:-1]: parent = parent[part] parent[parts[-1]] = value for case_id, path, within, outside in checks: case = by_id[case_id] result = self._physical_result(case) with self.subTest(case=case_id, position=within): set_path(result, path, within) validate_result_contract(case, result, require_physical=True) with self.subTest(case=case_id, position=outside): set_path(result, path, outside) with self.assertRaisesRegex(Ch1HilContractError, "within ±"): validate_result_contract(case, result, require_physical=True) def test_invalid_numeric_tolerance_contracts_fail_fixture_validation(self) -> None: window_case_index = next( index for index, case in enumerate(self.suite["case_catalog"]) if case["id"] == "WINDOW-FRONT-LEFT-10" ) for tolerance in (-1, True, "3", float("inf")): fixture = deepcopy(self.suite) fixture["case_catalog"][window_case_index]["expected"]["state_assertions"][0][ "tolerance" ] = tolerance with self.subTest(tolerance=tolerance): with self.assertRaisesRegex( Ch1HilContractError, "finite non-negative numeric" ): validate_suite(fixture) fixture = deepcopy(self.suite) fixture["case_catalog"][window_case_index]["expected"]["state_assertions"][0][ "equals" ] = "10" with self.assertRaisesRegex( Ch1HilContractError, "finite non-negative numeric" ): validate_suite(fixture) def test_front_left_window_fifty_is_authoritative_ch1_case(self) -> None: case = next( case for case in self.suite["case_catalog"] if case["id"] == "WINDOW-FRONT-LEFT-50" ) expected = case["expected"] self.assertIsNot(case.get("hil"), False) self.assertEqual(case["command"], "window front-left position 50") self.assertEqual(expected["frame_ids"], ["0x431"]) self.assertEqual(expected["feedback_ids"], ["0x388"]) self.assertEqual( expected["state_assertions"], [ { "path": "domain_state.window.positions.front_left.position_pct", "equals": 50, "tolerance": 3, } ], ) def test_each_module_restores_shade_to_zero(self) -> None: for module in ("self_orchestration", "self_extension"): shade_cases = [ case["id"] for case in suite_cases(self.suite, module) if case["id"].startswith("SHADE-POSITION-") ] self.assertEqual( shade_cases, [ "SHADE-POSITION-10", "SHADE-POSITION-50", "SHADE-POSITION-100", "SHADE-POSITION-0", ], ) def test_every_positive_case_compiles_offline_to_one_ch1_body_can_plan(self) -> None: cases = suite_cases(self.suite, "self_orchestration") for index, case in enumerate(cases): with self.subTest(case=case["id"], command=case["command"]): result = self.service.process( case["command"], provider="local", conversation_id=f"conv-ch1-contract-{index}", execution_mode="dry_run", vehicle_id=self.suite["vehicle_id"], ) self.assertEqual(result["status"], "succeeded") validate_result_contract(case, result, require_physical=False) def test_self_extension_exact_ac_endpoint_keeps_gpt56_and_one_operation( self, ) -> None: extension = SelfExtensionService(SimulationGateway()) extension.vehicle_provider = "openai" extension.vehicle_execution_mode = "dry_run" extension.vehicle_id = self.suite["vehicle_id"] self.service.openai.api_key = "offline-test-key" self.service.openai.model = "gpt-5.6-sol" cases_by_command = { case["command"]: case for case in self.suite["case_catalog"] } for index, (command, state) in enumerate((("ac on", "on"), ("ac off", "off"))): model_output = { "detected_domains": ["climate"], "requests": [ { "domain": "climate", "capability": "climate.set_ac", "target_scope": "vehicle", "operation": "set", "parameters": [{"name": "state", "value": state}], "reason": "GPT interpreted the exact AC request", } ], "ambiguities": [], "decision_summary": ["GPT-5.6 climate intent"], "confidence": 0.99, } captured: list[dict] = [] vehicle_service = Mock() def process(*args: object, **kwargs: object) -> dict: result = self.service.process(*args, **kwargs) captured.append(result) return result vehicle_service.process.side_effect = process with self.subTest(command=command): with patch.object( self.service.openai, "interpret", return_value=model_output, ) as interpret: response = extension.execute_vehicle( { "command": command, "conversation_id": f"conv-extension-exact-ac-{index}", }, vehicle_service, hardware_access_bypass=True, ) interpret.assert_called_once() self.assertEqual(vehicle_service.process.call_args.args[1], "openai") self.assertEqual(response["provider_used"], "openai:gpt-5.6-sol") self.assertTrue(response["success"]) self.assertEqual(len(response["operation_results"]), 1) self.assertEqual(len(captured), 1) full_result = captured[0] validate_result_contract( cases_by_command[command], full_result, require_physical=False ) config = full_result["function_configs"]["native_climate"] self.assertEqual(len(config["spec"]["operations"]), 1) self.assertEqual( config["spec"]["operations"][0]["capability"], "climate.set_power", ) self.assertEqual( len( full_result["command_plans"]["native_climate"]["spec"][ "commands" ] ), 1, ) def test_self_extension_non_exact_multi_action_keeps_model_orchestration( self, ) -> None: extension = SelfExtensionService(SimulationGateway()) extension.vehicle_provider = "openai" extension.vehicle_execution_mode = "dry_run" self.service.openai.api_key = "offline-test-key" self.service.openai.model = "gpt-5.6-sol" model_output = { "detected_domains": ["climate"], "requests": [ { "domain": "climate", "capability": "climate.set_ac", "target_scope": "vehicle", "operation": "set", "parameters": [{"name": "state", "value": "on"}], "reason": "enable AC", }, { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "driver", "operation": "set", "parameters": [{"name": "temperature_c", "value": 24}], "reason": "set driver temperature", }, ], "ambiguities": [], "decision_summary": ["GPT-5.6 multi-action climate intent"], "confidence": 0.98, } captured: list[dict] = [] vehicle_service = Mock() def process(*args: object, **kwargs: object) -> dict: result = self.service.process(*args, **kwargs) captured.append(result) return result vehicle_service.process.side_effect = process with patch.object( self.service.openai, "interpret", return_value=model_output, ) as interpret: response = extension.execute_vehicle( { "command": ( "Please turn on the air conditioner and set the driver " "temperature to 24 C." ), "conversation_id": "conv-extension-natural-multi-action", }, vehicle_service, hardware_access_bypass=True, ) interpret.assert_called_once() self.assertEqual(vehicle_service.process.call_args.args[1], "openai") self.assertEqual(response["provider_used"], "openai:gpt-5.6-sol") self.assertTrue(response["success"]) self.assertEqual(len(response["operation_results"]), 2) self.assertEqual(len(captured), 1) operations = next(iter(captured[0]["function_configs"].values()))["spec"][ "operations" ] self.assertEqual( [item["capability"] for item in operations], ["climate.set_ac", "climate.set_temperature"], ) def test_rejected_paths_never_compile_a_command_plan(self) -> None: retired_door_ids = { "REJECT-VEHICLE-LOCK", "REJECT-VEHICLE-UNLOCK", "REJECT-REAR-LOCK", "REJECT-REAR-UNLOCK", } for index, case in enumerate( suite_rejections(self.suite, "self_orchestration") ): with self.subTest(case=case["id"], command=case["command"]): try: result = self.service.process( case["command"], provider="local", conversation_id=f"conv-ch1-rejection-{index}", execution_mode="dry_run", vehicle_id=self.suite["vehicle_id"], ) except NativeCanCompileError as exc: self.assertIn(case["id"], retired_door_ids) self.assertIn("unsupported for physical execution", str(exc)) continue self.assertNotEqual(result["status"], "succeeded") self.assertEqual(result["function_configs"], {}) self.assertEqual(result["command_plans"], {}) def test_hil_is_default_closed_and_requires_all_three_exact_gates(self) -> None: armed = {ARM_ENV: "true", TARGET_ENV: EXPECTED_TARGET} require_explicit_hil_arm(armed, confirmation=CONFIRMATION_TEXT) for environment, confirmation in ( ({}, CONFIRMATION_TEXT), ({ARM_ENV: "false", TARGET_ENV: EXPECTED_TARGET}, CONFIRMATION_TEXT), ({ARM_ENV: "true", TARGET_ENV: "ch0"}, CONFIRMATION_TEXT), (armed, "ch1_driver_only"), ): with self.subTest(environment=environment, confirmation=confirmation): with self.assertRaises(Ch1HilSafetyError): require_explicit_hil_arm(environment, confirmation=confirmation) api = FailingApi() runner = Ch1HilRunner( self.suite, api, stationary_snapshot, confirmation="", ) with patch.dict( os.environ, {ARM_ENV: "false", TARGET_ENV: EXPECTED_TARGET}, clear=False, ): with self.assertRaises(Ch1HilSafetyError): runner.run() self.assertEqual(api.calls, []) def test_runtime_refuses_ch0_before_physical_execution(self) -> None: validate_runtime(valid_runtime()) runtime = valid_runtime() runtime["vehicle_can"]["channel"] = 0 runtime["multi_channel_can"]["vehicle_channel"] = 0 with self.assertRaisesRegex(Ch1HilSafetyError, "CH1"): validate_runtime(runtime) fixture = deepcopy(self.suite) fixture["physical_scope"]["channel"] = 0 with self.assertRaisesRegex(Ch1HilSafetyError, "channel 1"): validate_suite(fixture) def test_passive_preflight_requires_fresh_ch1_p_and_zero_speed_frames(self) -> None: bus = ReceiveOnlyBus( [ CanFrame(0x342, bytes((0, 0, 0, 1, 0, 0, 0, 0)), channel=0), CanFrame(0x342, bytes((0, 0, 0, 1, 0, 0, 0, 0)), channel=1), CanFrame(0x4EB, bytes((7, 0, 0, 0, 0, 0, 0, 0)), channel=1), ] ) snapshot = observe_stationary_ch1(bus, timeout_seconds=0.2) validate_stationary_snapshot(snapshot) self.assertEqual(snapshot["frame_count"], 3) self.assertTrue(all(include_fd for _, include_fd in bus.receive_calls)) self.assertFalse(hasattr(bus, "send")) moving = deepcopy(snapshot) moving["speed_kmh"] = 1.0 with self.assertRaisesRegex(Ch1HilSafetyError, "0 km/h"): validate_stationary_snapshot(moving) def test_frontend_sources_expose_feedback_and_state_synchronization_contract(self) -> None: validate_ui_contract( "self_orchestration", (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text( encoding="utf-8" ), ) validate_ui_contract( "self_extension", ( REPOSITORY_ROOT / "apps" / "web-console" / "self-extension.js" ).read_text(encoding="utf-8"), ) def test_physical_success_requires_matching_feedback_and_decoded_state(self) -> None: case = next( case for case in suite_cases(self.suite, "self_orchestration") if case["id"] == "CLIMATE-POWER-ON" ) result = self._physical_result(case) validate_result_contract(case, result, require_physical=True) for field, value in ( ("feedback_required", False), ("feedback_observed", False), ("feedback_match", False), ("feedback_state", None), ): failed = deepcopy(result) failed["execution_report"]["spec"]["results"][0]["telemetry"][field] = value with self.subTest(field=field): with self.assertRaises((Ch1HilContractError, Ch1HilSafetyError)): validate_result_contract(case, failed, require_physical=True) def test_physical_contract_rejects_non_authoritative_or_unsynchronized_evidence( self, ) -> None: case = next( case for case in suite_cases(self.suite, "self_orchestration") if case["id"] == "CLIMATE-POWER-ON" ) result = self._physical_result(case) telemetry = result["execution_report"]["spec"]["results"][0]["telemetry"] def telemetry_of(item: dict) -> dict: return item["execution_report"]["spec"]["results"][0]["telemetry"] def replace_feedback_id(item: dict) -> None: telemetry_item = telemetry_of(item) telemetry_item["feedback_frame"]["frame_id"] = "0x4FA" telemetry_item["feedback_state"].update( frame_id="0x4FA", feedback_frame_id="0x4FA" ) invalid_mutations = ( ( "intermediate vehicle mode", lambda item: item["execution_report"]["spec"].update(mode="vehicle"), ), ( "missing feedback frame", lambda item: telemetry_of(item).update(feedback_frame=None), ), ( "wrong raw frame channel", lambda item: telemetry_of(item)["feedback_frame"].update(channel=0), ), ( "incomplete raw frame", lambda item: telemetry_of(item)["feedback_frame"].update( payload_size=7 ), ), ( "wrong decoded source", lambda item: telemetry_of(item)["feedback_state"].update( source="physical_can_ch0" ), ), ( "wrong decoded channel", lambda item: telemetry_of(item)["feedback_state"].update(channel=0), ), ( "invalid decoded state", lambda item: telemetry_of(item)["feedback_state"].update(valid=False), ), ( "mismatched decoded frame", lambda item: telemetry_of(item)["feedback_state"].update( frame_id="0x4FA" ), ), ("unexpected feedback ID", replace_feedback_id), ( "feedback differs from requested state", lambda item: telemetry_of(item)["feedback_state"]["domain_state"][ "climate" ].update(power="off"), ), ( "UI state differs from feedback", lambda item: item["domain_state"]["climate"].update(power="off"), ), ) self.assertEqual(telemetry["feedback_state"]["source"], "physical_can_ch1") for label, mutate in invalid_mutations: failed = deepcopy(result) mutate(failed) with self.subTest(evidence=label): with self.assertRaises((Ch1HilContractError, Ch1HilSafetyError)): validate_result_contract(case, failed, require_physical=True) def test_timeout_disconnect_and_busy_are_distinct_failure_categories(self) -> None: self.assertEqual( classify_failure(504, {"message": "gateway timeout"}), "disconnected" ) self.assertEqual( classify_failure(422, {"message": "feedback timed out"}), "feedback_timeout", ) self.assertEqual( classify_failure(409, {"error": "hardware_busy"}), "busy_conflict" ) self.assertEqual( classify_failure(403, {"error": "hardware_safety_gate"}), "safety_rejection", ) def test_runner_stops_after_first_failure_and_runs_one_verified_cleanup(self) -> None: fixture = deepcopy(self.suite) fixture["modules"]["self_orchestration"]["case_ids"] = [ "CLIMATE-POWER-ON", "CLIMATE-POWER-OFF", ] fixture["modules"]["self_extension"]["case_ids"] = [ "CLIMATE-POWER-ON", "CLIMATE-POWER-OFF", ] catalog = {case["id"]: case for case in fixture["case_catalog"]} cleanup_results = { item["command"]: self._physical_result(catalog[item["case_id"]]) for item in fixture["cleanup"] } api = FailingApi(cleanup_results) safety_reader = Mock(side_effect=stationary_snapshot) runner = Ch1HilRunner( fixture, api, safety_reader, confirmation=CONFIRMATION_TEXT, ) with patch.dict( os.environ, {ARM_ENV: "true", TARGET_ENV: EXPECTED_TARGET}, clear=False, ): report = runner.run() self.assertEqual(report["summary"]["failed"], 1) self.assertTrue(report["summary"]["aborted_on_first_failure"]) self.assertEqual(report["summary"]["cleanup_runs"], 1) self.assertFalse(report["summary"]["cleanup_failed"]) self.assertEqual(safety_reader.call_count, 2 + len(fixture["cleanup"])) action_commands = [ payload["utterance"] for path, _, payload in api.calls if path == "/api/requests" and payload and payload.get("conversation_id") != "conv-ch1-hil-cleanup" ] self.assertEqual(action_commands, ["ac on"]) cleanup_commands = [ payload["utterance"] for _, _, payload in api.calls if payload and payload.get("conversation_id") == "conv-ch1-hil-cleanup" ] self.assertEqual( cleanup_commands, [item["command"] for item in fixture["cleanup"]], ) self.assertEqual( sum(1 for _, _, payload in api.calls if payload is None), 1, ) def test_cleanup_failure_is_reported_as_a_suite_failure(self) -> None: fixture = deepcopy(self.suite) fixture["modules"]["self_orchestration"]["case_ids"] = [ "CLIMATE-POWER-ON" ] fixture["modules"]["self_extension"]["case_ids"] = [ "CLIMATE-POWER-ON" ] api = FailingApi() runner = Ch1HilRunner( fixture, api, stationary_snapshot, confirmation=CONFIRMATION_TEXT, ) with patch.dict( os.environ, {ARM_ENV: "true", TARGET_ENV: EXPECTED_TARGET}, clear=False, ): report = runner.run() self.assertTrue(report["summary"]["cleanup_failed"]) self.assertGreaterEqual(report["summary"]["failed"], 2) def test_runner_never_submits_negative_safety_cases_to_live_endpoints(self) -> None: fixture = deepcopy(self.suite) for module in ("self_orchestration", "self_extension"): fixture["modules"][module]["case_ids"] = [ "OUTLET-DRIVER-LEFT-CENTER" ] api = FailingApi() runner = Ch1HilRunner( fixture, api, stationary_snapshot, confirmation=CONFIRMATION_TEXT, ) with patch.dict( os.environ, {ARM_ENV: "true", TARGET_ENV: EXPECTED_TARGET}, clear=False, ): report = runner.run() self.assertEqual(report["summary"]["total"], 0) self.assertEqual( report["summary"]["offline_rejection_cases"], 2 * len(fixture["rejection_catalog"]), ) self.assertEqual(len(fixture["rejection_catalog"]), 14) self.assertEqual( [(path, method) for path, method, _ in api.calls], [("/api/runtime", "GET")], ) if __name__ == "__main__": unittest.main()