from __future__ import annotations import unittest from pathlib import Path from typing import Any from cognitive_agent.application.native_can_intent import merge_native_can_intent from cognitive_agent.application.native_can_runtime import NativeCanRuntime from cognitive_agent.infrastructure.speech.capability import CapabilityResolver from cognitive_agent.infrastructure.speech.understanding import ( UnderstandingTranscriptProcessor, ) from cognitive_agent.infrastructure.speech.vocabulary import DeviceTerm REPOSITORY_ROOT = Path(__file__).resolve().parents[2] def _valid_value(spec: dict[str, Any]) -> Any: allowed = spec.get("allowed_values") if allowed: return allowed[0] value_type = str(spec.get("type", "string")) if value_type in {"integer", "number"}: minimum = spec.get("minimum", 0) return int(minimum) if value_type == "integer" else float(minimum) if value_type == "boolean": return True if value_type == "object": return {} if value_type == "array": return [] if str(spec.get("name")) == "color_hex": return "#FF0000" if str(spec.get("name")) == "preset_name": return "回归测试姿态" return "test" def _sample_parameters(definition: Any) -> dict[str, Any]: specs = list(definition.parameters) selected = [item for item in specs if item.get("required")] if not selected and specs: selected = [specs[0]] return {str(item["name"]): _valid_value(item) for item in selected} class SpeechCapabilityAlignmentTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) cls.resolver = CapabilityResolver(REPOSITORY_ROOT) cls.native_runtime = NativeCanRuntime(REPOSITORY_ROOT) def test_eight_reported_cases_preserve_structure_and_resolution(self) -> None: cases = ( { "text": "把遮阳帘打开", "status": "recognized_executable", "clean": "打开遮阳帘。", "commands": ( ( "roof.set_shade_motion", "vehicle", "roof.shade_motion", "open", {"motion": "open"}, ), ), }, { "text": "打开前大灯", "status": "recognized_unsupported", "clean": "打开前大灯。", "commands": ((None, "vehicle", "front_light.power", "on", {"state": "on"}),), }, { "text": "打开主驾座椅通风", "status": "recognized_executable", "clean": "打开主驾座椅通风。", "commands": ( ( "seat.set_driver_ventilation_native", "driver", "seat.ventilation", "on", {"state": "on", "level": 2}, ), ), }, { "text": "把主驾座椅调整为零重力模式", "status": "recognized_unsupported", "clean": "将主驾座椅调整为零重力模式。", "commands": ( (None, "driver", "seat.mode", "apply", {"mode_profile_id": "zero_gravity"}), ), }, { "text": "打开一半车窗", "status": "recognized_executable", "clean": "将车窗打开到50%。", "commands": ( ( "window.set_position", "front_left", "window.position", "set", {"position_pct": 50}, ), ), }, { "text": "打开氛围灯并调整为红色", "status": "incomplete", "clean": "打开氛围灯，然后将氛围灯颜色调整为红色。", "commands": ( ("ambient_light.set_power", "vehicle", "ambient.power", "on", {"state": "on"}), ( "ambient_light.set_illumination", "vehicle", "ambient.color", "set", {"color_hex": "#FF0000"}, ), ), }, { "text": "打开主驾座椅加热", "status": "recognized_executable", "clean": "打开主驾座椅加热。", "commands": ( ( "seat.set_driver_heating_native", "driver", "seat.heating", "on", {"state": "on", "level": 2}, ), ), }, { "text": "打开副驾座椅通风", "status": "recognized_executable", "clean": "打开副驾座椅通风。", "commands": ( ( "seat.set_ventilation", "passenger", "seat.ventilation", "on", {"state": "on"}, ), ), }, ) for case in cases: with self.subTest(text=case["text"]): result = self.processor.process(case["text"]) self.assertEqual(result.raw_text, case["text"]) self.assertEqual(result.clean_text, case["clean"]) self.assertEqual(result.semantic_status, case["status"]) self.assertEqual(len(result.commands), len(case["commands"])) for command, expected in zip(result.commands, case["commands"], strict=True): capability, target, property_name, operation, parameters = expected self.assertEqual(command.capability, capability) self.assertEqual(command.target_scope, target) self.assertEqual(command.property_name, property_name) self.assertEqual(command.operation, operation) self.assertEqual(command.parameters, parameters) if command.resolution_status == "executable": self.assertIsNotNone(command.intent_ir_action) self.assertIsNotNone(command.executor_payload) self.assertEqual(command.intent_ir_action["target"]["zone"], target) self.assertEqual(command.executor_payload["target"], target) self.assertEqual(command.executor_payload["parameters"], parameters) else: self.assertIsNone(command.executor_payload) def test_semantic_trace_covers_every_required_boundary(self) -> None: result = self.processor.process("打开主驾座椅通风") self.assertEqual( [item["stage"] for item in result.semantic_trace], [ "raw_text", "normalized_text", "clause_segmentation", "semantic_commands", "intent_ir", "capability_resolution", "executor_payload", "clean_text", ], ) def test_explicit_control_slots_cannot_disappear_from_clean_text(self) -> None: cases = ( ("打开主驾座椅加热", ("主驾", "座椅加热")), ("打开副驾座椅通风", ("副驾", "座椅通风")), ("打开右后车窗", ("右后", "车窗")), ("打开一半车窗", ("50%", "车窗")), ("把氛围灯调整为红色", ("氛围灯", "红色")), ("把主驾座椅调整为零重力模式", ("主驾", "零重力模式")), ) for raw, preserved in cases: with self.subTest(raw=raw): result = self.processor.process(raw) for token in preserved: self.assertIn(token, result.clean_text) def test_compound_shared_target_is_preserved_and_gated_atomically(self) -> None: result = self.processor.process("打开氛围灯并调整为红色") self.assertEqual( [command.property_name for command in result.commands], ["ambient.power", "ambient.color"], ) self.assertEqual(result.commands[1].value, "#FF0000") self.assertEqual(result.commands[1].resolution_status, "incomplete") self.assertIn("brightness_pct", "".join(result.commands[1].validation_issues)) self.assertFalse(result.execution_eligible) complete = self.processor.process("把氛围灯调整为红色并将亮度设置为50%") self.assertEqual(len(complete.commands), 1) self.assertEqual( complete.commands[0].parameters, {"color_hex": "#FF0000", "brightness_pct": 50}, ) self.assertEqual(complete.semantic_status, "recognized_executable") self.assertIn("红色", complete.clean_text) self.assertIn("50%", complete.clean_text) def test_reported_compound_and_extreme_position_phrases_preserve_all_slots(self) -> None: compound = self.processor.process( "打开阅读灯，并且打开氛围灯，并把氛围灯变成红色" ) self.assertEqual( [command.property_name for command in compound.commands], ["reading_light.power", "ambient.power", "ambient.color"], ) self.assertEqual(compound.commands[-1].parameters, {"color_hex": "#FF0000"}) self.assertEqual(compound.semantic_status, "incomplete") self.assertFalse(compound.execution_eligible) for token in ("阅读灯", "氛围灯", "红色"): self.assertIn(token, compound.clean_text) legrest = self.processor.process("把副驾腿托向上调整到最大位置") self.assertEqual(legrest.semantic_status, "recognized_executable") self.assertTrue(legrest.execution_eligible) self.assertEqual(legrest.commands[0].target_scope, "passenger") self.assertEqual(legrest.commands[0].capability, "seat.set_legrest_height") self.assertEqual(legrest.commands[0].parameters, {"position_pct": 100}) for token in ("副驾", "腿托", "向上", "100%"): self.assertIn(token, legrest.clean_text) def test_latest_reported_compounds_preserve_every_mutation(self) -> None: passenger = self.processor.process( "打开副驾座椅通风和加热，，然后将副驾座椅靠背调整到最大位置" ) self.assertEqual( [command.property_name for command in passenger.commands], ["seat.ventilation", "seat.heating", "seat.backrest_recline"], ) self.assertEqual( [command.target_scope for command in passenger.commands], ["passenger", "passenger", "passenger"], ) self.assertEqual(passenger.commands[-1].parameters, {"position_pct": 100}) self.assertTrue(passenger.execution_eligible) for token in ("副驾座椅通风", "副驾座椅加热", "副驾座椅靠背", "100%"): self.assertIn(token, passenger.clean_text) driver = self.processor.process( "打开主驾座椅通风和加热，并且把空调调整为24度" ) self.assertEqual( [command.property_name for command in driver.commands], ["seat.ventilation", "seat.heating", "hvac.temperature"], ) self.assertEqual( [command.target_scope for command in driver.commands[:2]], ["driver", "driver"], ) self.assertEqual(driver.commands[-1].parameters, {"temperature_c": 24}) self.assertTrue(driver.execution_eligible) for token in ("主驾座椅通风", "主驾座椅加热", "24度"): self.assertIn(token, driver.clean_text) def test_coordinated_seat_properties_cover_order_connectors_and_actions(self) -> None: cases = ( ("打开主驾座椅通风与加热", "driver", ("seat.ventilation", "seat.heating"), ("on", "on")), ("打开副驾座椅加热以及通风", "passenger", ("seat.heating", "seat.ventilation"), ("on", "on")), ("关闭主驾座椅通风、加热", "driver", ("seat.ventilation", "seat.heating"), ("off", "off")), ("打开主驾座椅通风和关闭加热", "driver", ("seat.ventilation", "seat.heating"), ("on", "off")), ) for text, scope, properties, operations in cases: with self.subTest(text=text): result = self.processor.process(text) self.assertEqual( tuple(command.property_name for command in result.commands), properties, ) self.assertEqual( tuple(command.operation for command in result.commands), operations, ) self.assertEqual( {command.target_scope for command in result.commands}, {scope}, ) self.assertTrue(result.execution_eligible) self.assertTrue(all(command.executor_payload for command in result.commands)) def test_coordinated_targets_keep_local_actions_scopes_and_properties(self) -> None: cases = ( ( "关闭空调和打开氛围灯", (("hvac.power", "off", "vehicle"), ("ambient.power", "on", "vehicle")), ), ( "打开主驾座椅通风和副驾座椅加热", (("seat.ventilation", "on", "driver"), ("seat.heating", "on", "passenger")), ), ( "打开前排和后排空调", (("hvac.power", "on", "front_row"), ("hvac.power", "on", "rear_row")), ), ( "打开主驾和副驾车窗", (("window.motion", "open", "front_left"), ("window.motion", "open", "front_right")), ), ) for text, expected in cases: with self.subTest(text=text): result = self.processor.process(text) self.assertEqual( tuple( (command.property_name, command.operation, command.target_scope) for command in result.commands ), expected, ) self.assertTrue(result.execution_eligible) for command in result.commands: self.assertEqual( command.executor_payload["target"], command.target_scope ) def test_ratio_semantics_are_shared_across_continuous_capabilities(self) -> None: cases = ( ("打开四分之一车窗", "window.position", {"position_pct": 25}, "25%"), ("打开四分之三车窗", "window.position", {"position_pct": 75}, "75%"), ("打开百分之三十车窗", "window.position", {"position_pct": 30}, "30%"), ("打开1/4车窗", "window.position", {"position_pct": 25}, "25%"), ("把遮阳帘打开到四分之一", "roof.shade_position", {"position_pct": 25}, "25%"), ("把副驾腿托调整到四分之三", "seat.legrest_height", {"position_pct": 75}, "75%"), ("把副驾座椅向后调整50%", "seat.longitudinal_position", {"position_pct": 50}, "50%"), ) for text, property_name, parameters, display in cases: with self.subTest(text=text): result = self.processor.process(text) self.assertEqual(len(result.commands), 1) self.assertEqual(result.commands[0].property_name, property_name) self.assertEqual(result.commands[0].parameters, parameters) self.assertIn(display, result.clean_text) self.assertTrue(result.execution_eligible) def test_invalid_unit_keeps_all_spoken_slots_and_fails_closed(self) -> None: raw = "打开遮阳帘，然后把副驾座椅向后调整50度" result = self.processor.process(raw) self.assertEqual(result.semantic_status, "incomplete") self.assertFalse(result.execution_eligible) self.assertEqual(result.commands[1].target_scope, "passenger") self.assertEqual(result.commands[1].value, 50) self.assertEqual(result.commands[1].parameters, {}) self.assertIsNone(result.commands[1].executor_payload) for token in ("遮阳帘", "副驾", "座椅", "向后", "50度"): self.assertIn(token, result.clean_text) self.assertNotIn("将副驾座椅向后调整。", result.clean_text) def test_shared_value_and_coordination_grammar_also_feed_one_line_parser(self) -> None: window = merge_native_can_intent({}, "打开四分之一车窗") self.assertEqual( [ (action["capability"], action["target"], action["parameters"]) for action in window["native_can_actions"] ], [("window.set_position", "front_left", {"position_pct": 25})], ) thermal = merge_native_can_intent({}, "打开主驾座椅通风和加热") self.assertEqual( [action["capability"] for action in thermal["native_can_actions"]], [ "seat.set_driver_ventilation_native", "seat.set_driver_heating_native", ], ) self.assertEqual( [action["parameters"] for action in thermal["native_can_actions"]], [ {"state": "on", "level": 2}, {"state": "on", "level": 2}, ], ) def test_shared_target_inherits_resolved_driver_scope_and_level_semantics(self) -> None: result = self.processor.process("打开座椅通风并调到三档") self.assertEqual( [command.capability for command in result.commands], [ "seat.set_driver_ventilation_native", "seat.set_driver_ventilation_native", ], ) self.assertEqual( [command.target_scope for command in result.commands], ["driver", "driver"], ) self.assertIn("3档", result.clean_text) self.assertEqual(result.semantic_status, "recognized_unsupported") self.assertFalse(result.execution_eligible) passenger = self.processor.process("打开副驾座椅通风并调到三档") self.assertEqual( [command.target_scope for command in passenger.commands], ["passenger", "passenger"], ) self.assertTrue(passenger.execution_eligible) graded = self.processor.process("打开主驾座椅加热并调到高档") self.assertEqual(graded.commands[-1].parameters["level"], 1) self.assertIn("高档", graded.clean_text) def test_colloquial_seat_aliases_do_not_turn_pragmatic_one_into_level(self) -> None: cases = ( ("把驾驶座通风打开", "driver", "seat.set_driver_ventilation_native"), ("把乘客座通风打开", "passenger", "seat.set_ventilation"), ("给副驾驶开一下座椅通风", "passenger", "seat.set_ventilation"), ) for text, target, capability in cases: with self.subTest(text=text): result = self.processor.process(text) self.assertTrue(result.execution_eligible) self.assertEqual(len(result.commands), 1) self.assertEqual(result.commands[0].target_scope, target) self.assertEqual(result.commands[0].capability, capability) self.assertEqual(result.commands[0].parameters, {"state": "on"}) explicit_level = self.processor.process("主驾座椅通风开到一档") self.assertEqual(explicit_level.commands[0].parameters["level"], 1) def test_discourse_connectors_preserve_actions_without_leaking_scope(self) -> None: cases = ( "打开主驾座椅通风，然后打开副驾座椅通风，然后将空调设置为24度。", "先打开主驾座椅通风，再打开副驾座椅通风，最后把空调调成二十四度", ) expected = ( ("seat.set_driver_ventilation_native", "driver", {"state": "on", "level": 2}), ("seat.set_ventilation", "passenger", {"state": "on"}), ("climate.set_temperature", "driver", {"temperature_c": 24}), ) for text in cases: with self.subTest(text=text): result = self.processor.process(text) self.assertEqual( tuple( (command.capability, command.target_scope, command.parameters) for command in result.commands ), expected, ) self.assertTrue(result.execution_eligible) for text in ( "打开主驾座椅通风还有副驾座椅通风", "打开主驾座椅通风之后打开副驾座椅通风", "打开主驾通风，再给副驾也打开", ): with self.subTest(text=text): result = self.processor.process(text) self.assertEqual( [(command.capability, command.target_scope) for command in result.commands], [ ("seat.set_driver_ventilation_native", "driver"), ("seat.set_ventilation", "passenger"), ], ) self.assertTrue(result.execution_eligible) def test_elliptical_values_inherit_only_an_authoritative_property(self) -> None: temperature = self.processor.process("空调二十四度") self.assertEqual(len(temperature.commands), 1) self.assertEqual(temperature.commands[0].capability, "climate.set_temperature") self.assertEqual(temperature.commands[0].target_scope, "driver") self.assertEqual(temperature.commands[0].parameters, {"temperature_c": 24}) self.assertTrue(temperature.execution_eligible) continued = self.processor.process("打开空调，然后二十四度") self.assertEqual( [(command.capability, command.target_scope) for command in continued.commands], [("climate.set_power", "vehicle"), ("climate.set_temperature", "driver")], ) self.assertTrue(continued.execution_eligible) ambient = self.processor.process("氛围灯打开，调成蓝色，亮度百分之五十") self.assertEqual( [command.capability for command in ambient.commands], ["ambient_light.set_power", "ambient_light.set_illumination"], ) self.assertEqual( ambient.commands[-1].parameters, {"color_hex": "#0000FF", "brightness_pct": 50}, ) self.assertTrue(ambient.execution_eligible) def test_non_control_continuation_never_replays_previous_action(self) -> None: for text in ("打开空调，然后开玩笑", "打开空调，然后开会"): with self.subTest(text=text): result = self.processor.process(text) self.assertEqual( [command.capability for command in result.commands], ["climate.set_power"], ) self.assertTrue(result.uncertain) self.assertFalse(result.execution_eligible) def test_unsupported_semantics_never_create_fake_executor_payloads(self) -> None: for text in ("打开前大灯", "把主驾座椅调整为零重力模式"): with self.subTest(text=text): result = self.processor.process(text) self.assertEqual(result.semantic_status, "recognized_unsupported") self.assertTrue(result.clean_text) self.assertTrue(result.commands) self.assertTrue( all(command.executor_payload is None for command in result.commands) ) def test_zero_gravity_is_real_passenger_mode_but_not_driver_capability(self) -> None: passenger = self.processor.process("打开零重力模式") self.assertEqual(passenger.semantic_status, "recognized_executable") self.assertEqual(passenger.commands[0].mode_profile_id, "zero_gravity") payload = passenger.commands[0].executor_payload self.assertIsNotNone(payload) self.assertEqual(len(payload["operations"]), 4) self.assertEqual( {operation["target"] for operation in payload["operations"]}, {"passenger"}, ) driver = self.processor.process("把主驾座椅调整为零重力模式") self.assertEqual(driver.semantic_status, "recognized_unsupported") self.assertIsNone(driver.commands[0].executor_payload) def test_native_executor_can_compile_reported_executable_cases(self) -> None: for text in ( "把遮阳帘打开", "打开主驾座椅通风", "打开一半车窗", "打开主驾座椅加热", ): with self.subTest(text=text): result = self.processor.process(text) payloads = [command.executor_payload for command in result.commands] self.assertTrue(all(payload is not None for payload in payloads)) config = self.native_runtime.build_function_config( payloads[0]["domain"], payloads, "speech-alignment", "test" ) plan = self.native_runtime.compile(config) self.assertTrue(plan["spec"]["commands"]) def test_catalog_driven_resolution_matrix_covers_every_definition(self) -> None: definitions = self.resolver.definitions() self.assertEqual(len(definitions), 44) projections = ( "canonical_action", "target_preservation", "parameter_preservation", "intent_ir_projection", "executor_projection", ) for definition in definitions: target = definition.target_scopes[0] if definition.target_scopes else "vehicle" parameters = _sample_parameters(definition) expected = ( "unsupported" if definition.availability == "catalog_only" or ( definition.availability == "native_can" and definition.name not in self.native_runtime.supported_capabilities ) else "executable" ) for projection in projections: with self.subTest(capability=definition.name, projection=projection): result = self.resolver.resolve( domain=definition.domain, semantic_property=definition.name, target_scope=target, target_device=None, operation="set", parameters=parameters, source_text=f"{projection}:{definition.name}:{target}", preferred_capability=definition.name, ) self.assertEqual(result.definition.name, definition.name) self.assertEqual(result.status, expected) if expected == "executable": self.assertEqual(result.executor_payload["target"], target) self.assertEqual(result.executor_payload["parameters"], parameters) self.assertEqual(result.intent_ir_action["target"]["zone"], target) else: self.assertIsNone(result.executor_payload) def test_vocabulary_does_not_define_vehicle_capabilities(self) -> None: self.assertNotIn("capability", DeviceTerm.__dataclass_fields__) if __name__ == "__main__": unittest.main()