from __future__ import annotations import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch from cognitive_agent.application.native_can_intent import ( parse_exact_native_can_command, ) from cognitive_agent.application.native_can_runtime import ( NativeCanCompileError, NativeCanRuntime, ) from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService HARDWARE_DISABLED_ENV = { "COGNITIVE_AGENT_CH1_HIL_ENABLED": "false", "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", } class Can1SemanticWhitelistTests(unittest.TestCase): def setUp(self) -> None: self.temporary_directory = tempfile.TemporaryDirectory() self.addCleanup(self.temporary_directory.cleanup) temporary_root = Path(self.temporary_directory.name) memory_path = temporary_root / "driver-seat-memory.json" memory_path.write_text( json.dumps( { "version": 1, "positions": { str(slot): { "payload_15": "15 00 00 00 00 00 00 00", "payload_16": "16 00 00 00 00 00 00 00", } for slot in (1, 2, 3) }, } ), encoding="utf-8", ) self.environment = patch.dict( os.environ, { **HARDWARE_DISABLED_ENV, "COGNITIVE_AGENT_DRIVER_SEAT_MEMORY_PATH": str(memory_path), }, clear=False, ) self.environment.start() self.addCleanup(self.environment.stop) self.runtime = NativeCanRuntime(REPOSITORY_ROOT) self.service = SeatDemoService(REPOSITORY_ROOT) self.service.artifact_root = temporary_root / "artifacts" @staticmethod def _valid_exact_commands() -> list[str]: commands = [f"door rear {state}" for state in ("lock", "unlock")] commands.extend( f"window {target} {motion}" for target in ( "all", "front", "rear", "front-left", "front-right", "rear-left", "rear-right", ) for motion in ("open", "close", "stop") ) commands.extend( f"window {target} position {position}" for target in ( "front-left", "front-right", "rear-left", "rear-right", ) for position in (0, 100) ) commands.extend(("ac on", "ac off")) commands.extend(f"ac fan {level}" for level in range(1, 8)) commands.extend( f"ac mode {mode}" for mode in ( "face", "face-feet", "feet", "feet-defrost", "defrost", "face-feet-defrost", "face-defrost", ) ) commands.extend( f"ac temp {target} {temperature}" for target in ("driver", "passenger", "rear") for temperature in (18, 32) ) commands.extend(("ac control manual", "ac control auto")) commands.extend(("ac cycle internal", "ac cycle external")) commands.extend( f"outlet {target} {state}" for target in ( "driver-left", "driver-right", "passenger-left", "passenger-right", ) for state in ("open", "close") ) commands.extend( f"outlet {target} percent {horizontal} {vertical}" for target in ( "driver-left", "driver-right", "passenger-left", "passenger-right", ) for horizontal, vertical in ((0, 0), (100, 100)) ) commands.extend( ( "outlet driver-left position -25 -40", "outlet driver-right position 30 30", "outlet passenger-left position -10 -40", "outlet passenger-right position 5 30", ) ) commands.extend( f"seat driver {feature} {level}" for feature in ("vent", "heat") for level in ("off", "low", "high") ) commands.append("seat driver massage off") commands.extend( f"seat driver massage strength {strength}" for strength in ("weak", "medium", "strong") ) commands.extend( f"seat driver massage mode {mode}" for mode in ("wave", "pulse", "stretch", "slack", "relax") ) commands.extend( ( "seat driver horizontal forward 100", "seat driver horizontal backward 2000", "seat driver horizontal stop", "seat driver height up 100", "seat driver height down 2000", "seat driver height stop", "seat driver backrest forward 100", "seat driver backrest backward 2000", "seat driver backrest stop", "seat driver basin up 100", "seat driver basin down 2000", "seat driver basin stop", ) ) commands.extend(f"seat driver position {slot}" for slot in (1, 2, 3)) commands.extend( f"seat driver memory recall {slot}" for slot in (1, 2, 3) ) commands.extend(f"light indoor {state}" for state in ("on", "off")) commands.extend( f"light reading all {state}" for state in ("on", "off") ) commands.extend(f"light ambient {state}" for state in ("on", "off")) commands.extend(f"light ambient color {value}" for value in (1, 127)) commands.extend( f"light ambient brightness {value}" for value in (1, 5) ) commands.extend( f"shade {motion}" for motion in ("open", "close", "half", "stop") ) commands.extend( f"shade position {position}" for position in (0, 1, 50, 99, 100) ) return commands def _compile_exact(self, command: str) -> dict: actions = parse_exact_native_can_command(command) self.assertEqual(len(actions), 1, command) action = actions[0] config = self.runtime.build_function_config( action["domain"], actions, "req-can-one-whitelist", "local_rules", ) return self.runtime.compile(config) @staticmethod def _debug_commands(result: dict) -> list[str]: return [ command["signals"]["debug_command"] for plan in result.get("command_plans", {}).values() for command in plan["spec"]["commands"] ] def test_every_documented_exact_command_parses_and_compiles(self) -> None: commands = self._valid_exact_commands() self.assertEqual(len(commands), 129) for command in commands: with self.subTest(command=command): plan = self._compile_exact(command) self.assertEqual( [ item["signals"]["debug_command"] for item in plan["spec"]["commands"] ], [command], ) def test_out_of_whitelist_exact_values_are_rejected(self) -> None: commands = ( "door rear open", "window vehicle open", "window all position 50", "window front-left position -1", "window front-left position 101", "ac fan 0", "ac fan 8", "ac mode auto", "ac temp driver 17", "ac temp driver 33", "ac temp driver 24.5", "ac control off", "outlet driver-left toggle", "outlet driver-left percent -1 50", "outlet driver-left percent 101 50", "seat driver vent medium", "seat driver horizontal forward 99", "seat driver horizontal forward 2001", "seat driver horizontal stop 500", "seat driver position 0", "seat driver position 4", "light ambient color 0", "light ambient color 128", "light ambient brightness 0", "light ambient brightness 6", "shade position -1", "shade position 101", ) for command in commands: with self.subTest(command=command): self.assertEqual(parse_exact_native_can_command(command), []) def test_calibrated_outlet_values_must_fit_the_selected_outlet(self) -> None: action = parse_exact_native_can_command( "outlet driver-left position -26 -40" )[0] config = self.runtime.build_function_config( action["domain"], [action], "req-can-one-calibration", "local_rules", ) with self.assertRaisesRegex( NativeCanCompileError, "horizontal_value exceeds the calibrated range" ): self.runtime.compile(config) def test_individual_outlet_switches_use_observed_byte_exact_triplets(self) -> None: cases = { "outlet driver-left open": "02 FF FF FF FF 01 FF FF", "outlet driver-left close": "02 FF FF FF FF 02 FF FF", "outlet driver-right open": "02 FF FF FF FF 04 FF FF", "outlet driver-right close": "02 FF FF FF FF 08 FF FF", "outlet passenger-left open": "03 FF FF FF FF 01 00 00", "outlet passenger-left close": "03 FF FF FF FF 02 00 00", "outlet passenger-right open": "03 FF FF FF FF 04 00 00", "outlet passenger-right close": "03 FF FF FF FF 08 00 00", } for command, expected_payload in cases.items(): with self.subTest(command=command): plan = self._compile_exact(command) compiled = plan["spec"]["commands"][0] self.assertEqual(compiled["signals"]["frame_id"], "0x4EF") self.assertEqual(compiled["signals"]["data_hex"], expected_payload) self.assertTrue(compiled["signals"]["is_fd"]) self.assertTrue(compiled["signals"]["bitrate_switch"]) self.assertEqual( compiled["repeat"], {"count": 3, "period_ms": 100} ) self.assertEqual( compiled["signals"]["feedback_ids"], "0x4BB,0x37B" ) def test_representative_chinese_commands_cover_every_whitelist_family(self) -> None: cases = ( ("后排车门上锁", ["door rear lock"]), ("后排门锁解锁", ["door rear unlock"]), ("打开全部车窗", ["window all open"]), ("关闭前排车窗", ["window front close"]), ("停止后排车窗", ["window rear stop"]), ("把副驾车窗开到35%", ["window front-right position 35"]), ("打开空调", ["ac on"]), ("关闭空调", ["ac off"]), ("风量调到7档", ["ac fan 7"]), ("空调出风模式设为吹脸", ["ac mode face"]), ("空调出风模式设为吹面和吹脚", ["ac mode face-feet"]), ("空调出风模式设为吹脚", ["ac mode feet"]), ("空调出风模式设为吹脚除霜", ["ac mode feet-defrost"]), ("空调出风模式设为除霜", ["ac mode defrost"]), ("空调出风模式设为吹脸和吹脚除霜", ["ac mode face-feet-defrost"]), ("空调出风模式设为吹脸除霜", ["ac mode face-defrost"]), ("主驾空调温度设为18度", ["ac temp driver 18"]), ("副驾空调温度设为32度", ["ac temp passenger 32"]), ("后排空调温度设为24度", ["ac temp rear 24"]), ("空调设置为自动控制", ["ac control auto"]), ("空调设置为手动控制", ["ac control manual"]), ("打开主驾左风口", ["outlet driver-left open"]), ("关闭主驾右风口", ["outlet driver-right close"]), ("打开副驾左吹面出风口", ["outlet passenger-left open"]), ("关闭副驾右吹面出风口", ["outlet passenger-right close"]), ("主驾左风口水平25%上下75%", ["outlet driver-left percent 25 75"]), ("副驾右风口左右25%垂直75%", ["outlet passenger-right percent 25 75"]), ("主驾左风口标定坐标左右-25上下-40", ["outlet driver-left position -25 -40"]), ("打开主驾座椅通风", ["seat driver vent low"]), ("关闭主驾座椅加热", ["seat driver heat off"]), ("主驾座椅加热高档", ["seat driver heat high"]), ("关闭主驾座椅按摩", ["seat driver massage off"]), ("主驾座椅按摩强度调为中等", ["seat driver massage strength medium"]), ("主驾座椅按摩模式设为放松", ["seat driver massage mode relax"]), ("主驾座椅向前移动500毫秒", ["seat driver horizontal forward 500"]), ("主驾座椅高度向下调500毫秒", ["seat driver height down 500"]), ("主驾座椅靠背后仰800毫秒", ["seat driver backrest backward 800"]), ("把主驾座椅坐盆抬高300毫秒", ["seat driver basin up 300"]), ("调用主驾座椅记忆位置2", ["seat driver position 2"]), ("打开室内灯", ["light indoor on"]), ("关闭全部阅读灯", ["light reading all off"]), ("打开氛围灯", ["light ambient on"]), ("关闭氛围灯", ["light ambient off"]), ("把氛围灯颜色编号调到70", ["light ambient color 70"]), ("氛围灯亮度调整为3档", ["light ambient brightness 3"]), ("打开遮阳帘", ["shade open"]), ("关闭遮阳帘", ["shade close"]), ("遮阳帘半开", ["shade half"]), ("停止遮阳帘", ["shade stop"]), ("遮阳帘开到20%", ["shade position 20"]), ("遮阳帘打开30%", ["shade position 30"]), ) for index, (utterance, expected) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-one-natural-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(self._debug_commands(result), expected) rear_context = self.service.conversations.snapshot( "conv-can-one-natural-18" ) self.assertEqual( rear_context["domain_devices"]["climate"]["zones"]["rear_row"][ "temperature_c" ], 24, ) def test_chinese_compressor_request_is_not_collapsed_into_hvac_power(self) -> None: cases = ( ("打开空调压缩机", "on"), ("空调压缩机设为开启状态", "on"), ("关闭空调压缩机", "off"), ("空调压缩机设置为关闭状态", "off"), ) for index, (utterance, expected_state) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-one-compressor-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") actions = result["intent_ir"]["spec"]["actions"] self.assertEqual(len(actions), 1) self.assertEqual(actions[0]["capability"], "climate.set_ac") self.assertEqual(actions[0]["parameters"], {"state": expected_state}) def test_location_aliases_never_redirect_to_the_wrong_default_target(self) -> None: recognized = ( ("后座车窗打开", ["window rear open"]), ("二排车窗打开", ["window rear open"]), ("后排左侧车窗打开", ["window rear-left open"]), ("后排右侧车窗打开", ["window rear-right open"]), ("后座空调温度设为25度", ["ac temp rear 25"]), ("二排空调温度设为25度", ["ac temp rear 25"]), ) for index, (utterance, expected) in enumerate(recognized): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-location-alias-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(self._debug_commands(result), expected) unsupported = ( "第三排车窗打开", "左侧车窗打开", "左侧空调温度设为25度", "打开主驾阅读灯", "打开驾驶员阅读灯", "打开前排阅读灯", "打开后排阅读灯", "打开二排座椅通风", ) for index, utterance in enumerate(unsupported): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-location-reject-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) self.assertEqual(result["intent_ir"]["spec"]["actions"], []) passenger = self.service.process( "乘员座椅加热2档", provider="local", conversation_id="conv-location-passenger-alias", execution_mode="dry_run", ) passenger_actions = passenger["intent_ir"]["spec"]["actions"] self.assertEqual(passenger["status"], "succeeded") self.assertEqual( [ (action["capability"], action["target"]["zone"]) for action in passenger_actions ], [("seat.set_heating", "front_right")], ) def test_multi_zone_temperature_values_remain_bound_to_their_clauses(self) -> None: cases = ( ( "主驾温度22度，副驾温度24度", ["ac temp driver 22", "ac temp passenger 24"], ), ( "副驾温度24度，主驾温度22度", ["ac temp passenger 24", "ac temp driver 22"], ), ( "主驾风量3档，副驾温度24度", ["ac temp passenger 24"], ), ) for index, (utterance, expected) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-climate-clause-binding-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(self._debug_commands(result), expected) def test_narrow_unsupported_climate_targets_never_widen_to_a_valid_zone(self) -> None: unsupported = ( "左后空调温度22度", "后排左侧空调温度22度", "后排风量3档", "后排内循环", "主驾氛围灯打开", ) for index, utterance in enumerate(unsupported): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-climate-scope-reject-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) self.assertEqual(result["intent_ir"]["spec"]["actions"], []) def test_explicit_outlet_locations_never_fall_back_to_driver_left(self) -> None: unsupported = ( "副驾出风口打开", "副驾出风口左右50%上下50%", "副驾出风口标定坐标左右20上下30", "后排出风口打开", "后排出风口左右50%上下50%", "后排出风口标定坐标左右20上下30", "驾驶员出风口打开", "驾驶员出风口左右50%上下50%", "驾驶员出风口标定坐标左右20上下30", ) for index, utterance in enumerate(unsupported): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-outlet-location-reject-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) self.assertEqual(result["intent_ir"]["spec"]["actions"], []) def test_meta_language_never_authorizes_embedded_vehicle_commands(self) -> None: utterances = ( "请复述：打开遮阳帘", "把‘打开空调’翻译成英文", "写一个故事，主角打开空调", "用‘打开主驾车窗’造句", ) for index, utterance in enumerate(utterances): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-meta-language-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) self.assertEqual(result["intent_ir"]["spec"]["actions"], []) self.assertIn( "meta_language_is_not_vehicle_authorization", { item["reason"] for item in result["intent_ir"]["spec"]["unresolved_spans"] }, ) def test_zoned_hvac_requests_never_expand_to_vehicle_wide_can(self) -> None: for index, utterance in enumerate( ( "把副驾空调设为吹脚", "后排空调设为吹面", "主驾内循环", "副驾风量设为2档", "后排空调设为自动模式", ) ): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-zoned-hvac-reject-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) self.assertEqual(result["intent_ir"]["spec"]["actions"], []) def test_localized_requests_never_expand_to_whole_vehicle_devices(self) -> None: utterances = ( "左后门上锁", "右后门解锁", "主驾车门上锁", "后排遮阳帘打开", "左侧遮阳帘打开", "主驾室内灯打开", "后排车内灯打开", "副驾氛围灯打开", "后排氛围灯亮度设为3档", ) for index, utterance in enumerate(utterances): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-whole-target-reject-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) self.assertEqual(result["intent_ir"]["spec"]["actions"], []) def test_disabling_one_outlet_feature_never_enables_that_mode(self) -> None: for index, utterance in enumerate( ("关闭吹面出风口", "把吹面出风口关闭") ): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-one-close-outlet-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "needs_confirmation") self.assertEqual(self._debug_commands(result), []) ambiguities = result["intent_ir"]["spec"]["ambiguities"] self.assertEqual( ambiguities[0]["reason"], "outlet_feature_disable_requires_absolute_mode", ) preview = self.service.preview_native_can( "关闭吹面出风口", provider="openai", conversation_id="conv-can-one-close-outlet-preview", execution_mode="dry_run", ) self.assertFalse(preview["supplement_required"]) self.assertEqual(preview["unresolved_clauses"], []) self.assertEqual(self._debug_commands(preview), []) explicit_replacement = self.service.process( "关闭吹面出风口，切换到吹脚模式", provider="local", conversation_id="conv-can-one-replace-outlet", execution_mode="dry_run", ) self.assertEqual(explicit_replacement["status"], "succeeded") self.assertEqual(self._debug_commands(explicit_replacement), ["ac mode feet"]) def test_driver_seat_natural_language_levels_and_motion_aliases(self) -> None: cases = ( ("打开主驾座椅按摩", ["seat driver massage mode relax"]), ( "主驾座椅按摩强度设置为最大", ["seat driver massage strength strong"], ), ( "主驾座椅按摩强度最小", ["seat driver massage strength weak"], ), ( "主驾座椅按摩强度调中", ["seat driver massage strength medium"], ), ("主驾座椅朝前挪一秒", ["seat driver horizontal forward 1000"]), ("主驾座椅后退两秒", ["seat driver horizontal backward 2000"]), ("主驾座椅椅背放倒", ["seat driver backrest backward 500"]), ("主驾座椅座垫放低", ["seat driver basin down 500"]), ("主驾座椅往上移动", ["seat driver height up 500"]), ("主驾座椅通风最低", ["seat driver vent low"]), ("主驾座椅通风最小", ["seat driver vent low"]), ("主驾座椅通风最高", ["seat driver vent high"]), ("主驾座椅通风最大", ["seat driver vent high"]), ("空调风量设置为最低", ["ac fan 1"]), ("空调风量设置为中等", ["ac fan 4"]), ("空调风量设置为最大", ["ac fan 7"]), ("氛围灯亮度最低", ["light ambient brightness 1"]), ("氛围灯亮度中等", ["light ambient brightness 3"]), ("氛围灯亮度最高", ["light ambient brightness 5"]), ( "打开主驾座椅通风并关闭按摩", ["seat driver vent low", "seat driver massage off"], ), ( "关闭主驾座椅通风并打开按摩", ["seat driver vent off", "seat driver massage mode relax"], ), ) for index, (utterance, expected) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-one-seat-alias-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(self._debug_commands(result), expected) stop_all = self.service.process( "停止主驾座椅移动", provider="local", conversation_id="conv-can-one-seat-stop-all", execution_mode="dry_run", ) self.assertEqual(stop_all["status"], "succeeded") self.assertEqual( self._debug_commands(stop_all), [ "seat driver horizontal stop", "seat driver height stop", "seat driver backrest stop", "seat driver basin stop", ], ) def test_unsupported_driver_seat_middle_thermal_level_is_explicit(self) -> None: for index, feature in enumerate(("通风", "加热")): utterance = f"主驾座椅{feature}调到中档" with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-one-seat-middle-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "needs_confirmation") self.assertEqual(self._debug_commands(result), []) self.assertIn( "driver_seat_thermal_middle_level_unsupported", { item.get("reason") for item in result["intent_ir"]["spec"]["ambiguities"] }, ) preview = self.service.preview_native_can( "主驾座椅通风设置为中等", provider="openai", conversation_id="conv-can-one-seat-middle-preview", execution_mode="dry_run", ) self.assertFalse(preview["supplement_required"]) self.assertEqual(preview["unresolved_clauses"], []) self.assertEqual(self._debug_commands(preview), []) def test_vehicle_position_aliases_stay_scoped_to_the_device_domain(self) -> None: cases = ( ("打开左前车窗", ["window front-left open"]), ("打开右前车窗", ["window front-right open"]), ("打开左后车窗", ["window rear-left open"]), ("打开右后车窗", ["window rear-right open"]), ("打开主驾车窗", ["window front-left open"]), ("打开副驾车窗", ["window front-right open"]), ("打开左前座椅通风", ["seat driver vent low"]), ("打开主驾座椅通风", ["seat driver vent low"]), ("主驾空调温度设置为24度", ["ac temp driver 24"]), ("副驾空调温度设置为24度", ["ac temp passenger 24"]), ) for index, (utterance, expected) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-position-alias-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(self._debug_commands(result), expected) for index, utterance in enumerate( ("打开右前座椅通风", "打开副驾座椅通风") ): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-passenger-alias-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") operations = [ operation for config in result.get("function_configs", {}).values() for operation in config.get("spec", {}).get("operations", []) ] self.assertNotIn( "seat.set_driver_ventilation_native", {item["capability"] for item in operations}, ) self.assertEqual(operations[0]["capability"], "seat.set_ventilation") self.assertEqual(operations[0]["target"]["zone"], "front_right") def test_fractional_values_are_not_silently_truncated(self) -> None: utterances = ( "空调温度设为24.5度", "风量设为2.5档", "氛围灯颜色编号调到70.5", "氛围灯亮度调到3.5档", ) for index, utterance in enumerate(utterances): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-can-one-fraction-{index}", execution_mode="dry_run", ) self.assertEqual(self._debug_commands(result), []) if __name__ == "__main__": unittest.main()