from __future__ import annotations import os import tempfile from pathlib import Path from unittest.mock import patch import pytest from cognitive_agent.application.seat_service import SeatDemoService from cognitive_agent.infrastructure.speech.understanding import ( UnderstandingTranscriptProcessor, ) REPOSITORY_ROOT = Path(__file__).resolve().parents[2] def _canonical_actions(result: dict[str, object]) -> list[dict[str, object]]: intent_ir = result.get("intent_ir") assert isinstance(intent_ir, dict) spec = intent_ir.get("spec") assert isinstance(spec, dict) actions = spec.get("actions") assert isinstance(actions, list) return [ { "domain": action.get("domain"), "capability": action.get("capability"), "target": action.get("target"), "operation": action.get("operation"), "parameters": action.get("parameters"), } for action in actions if isinstance(action, dict) ] def _sorted_canonical_actions(result: dict[str, object]) -> list[dict[str, object]]: return sorted( _canonical_actions(result), key=lambda action: ( str(action["domain"]), str(action["capability"]), repr(action["target"]), repr(action["parameters"]), ), ) def _mode_service(directory: str) -> SeatDemoService: service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = Path(directory) / "artifacts" return service @pytest.mark.parametrize( ("utterance", "capability", "target", "parameters"), ( ( "打开一半车窗", "window.set_position", {"device": "window", "zone": "front_left"}, {"position_pct": 50}, ), ( "把副驾腿托向上调整到最大位置", "seat.set_legrest_height", {"device": "seat", "zone": "front_right", "instance": "passenger"}, {"position_pct": 100}, ), ( "把遮阳帘打开", "roof.set_shade_motion", {"device": "roof", "zone": "vehicle"}, {"motion": "open"}, ), ( "遮阳帘打开一半", "roof.set_shade_motion", {"device": "roof", "zone": "vehicle"}, {"motion": "half"}, ), ( "把遮阳帘打开到四分之一", "roof.set_shade_position", {"device": "roof", "zone": "vehicle"}, {"position_pct": 25}, ), ), ) def test_server_owned_speech_semantics_bypass_second_language_parse( utterance: str, capability: str, target: dict[str, str], parameters: dict[str, int], ) -> None: processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) understanding = processor.process(utterance).to_dict() understanding["utterance_id"] = "bridge-regression" with tempfile.TemporaryDirectory() as directory, patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false", }, clear=False, ): service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = Path(directory) / "artifacts" with patch.object( service, "_interpret", side_effect=AssertionError("cleanText must not be parsed a second time"), ): result = service.process( understanding["clean_text"], provider="local", execution_mode="dry_run", _speech_understanding=understanding, ) assert result["status"] == "succeeded" assert result["provider_used"] == "speech_semantic" assert result["speech_understanding"]["raw_text"] == utterance actions = result["intent_ir"]["spec"]["actions"] assert len(actions) == 1 assert actions[0]["capability"] == capability assert actions[0]["target"] == target assert actions[0]["parameters"] == parameters assert actions[0]["provenance"]["parser"] == "speech_semantic" def test_non_executable_speech_cannot_enter_orchestration_bridge() -> None: processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) understanding = processor.process("打开前大灯").to_dict() understanding["utterance_id"] = "bridge-unsupported" with patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", }, clear=False, ): service = SeatDemoService(REPOSITORY_ROOT) context = service._context_for_vehicle( service.conversations.snapshot("conv-speech-bridge"), service.vehicle_profiles.get(None), ) with pytest.raises(ValueError, match="not execution eligible"): service._adapt_speech_understanding( understanding, context, service.vehicle_profiles.get(None), ) def test_every_registered_native_runtime_capability_keeps_native_bridge_binding() -> None: def valid_value(spec: dict[str, object]) -> object: allowed = spec.get("allowed_values") if isinstance(allowed, list) and allowed: return allowed[0] value_type = str(spec.get("type", "string")) if value_type in {"integer", "number"}: return spec.get("minimum", 0) if value_type == "boolean": return True if value_type == "object": return {} if value_type == "array": return [] return "test" with patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", }, clear=False, ): service = SeatDemoService(REPOSITORY_ROOT) context = service._context_for_vehicle( service.conversations.snapshot("conv-native-speech-matrix"), service.vehicle_profiles.get(None), ) registered = { definition.name: definition for definition in service.capability_registry.definitions() if definition.name in service.native_can_runtime.supported_capabilities } assert registered for capability, definition in registered.items(): parameters = { str(spec["name"]): valid_value(spec) for spec in definition.parameters if spec.get("required") } target = definition.target_scopes[0] understanding = { "raw_text": f"native bridge matrix: {capability}", "execution_eligible": True, "commands": [ { "status": "active", "negated": False, "resolution_status": "executable", "validation_issues": [], "confidence": 1.0, "source_text": capability, "executor_payload": { "domain": definition.domain, "capability": capability, "target": target, "operation": "set", "parameters": parameters, }, } ], } parsed = service._adapt_speech_understanding( understanding, context, service.vehicle_profiles.get(None), ) assert parsed["deferred_requests"] == [], capability assert parsed["actions"] == [], capability assert parsed["domain_actions"] == [], capability assert [ action["capability"] for action in parsed["native_can_actions"] ] == [capability] def test_text_and_speech_share_implicit_mode_target_and_start_command() -> None: processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) understanding = processor.process("将座椅调整为木马模式。").to_dict() understanding["utterance_id"] = "mode-start-equivalence" with tempfile.TemporaryDirectory() as directory, patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false", "COGNITIVE_AGENT_MODE_PROFILE_ROOT": str(Path(directory) / "modes"), }, clear=False, ): service = _mode_service(directory) text_result = service.process( "将座椅调整为木马模式。", provider="local", conversation_id="conv-text-mode-start", ) speech_result = service.process( understanding["clean_text"], provider="local", conversation_id="conv-speech-mode-start", _speech_understanding=understanding, ) assert understanding["execution_eligible"] is True assert text_result["status"] == speech_result["status"] == "succeeded" assert _canonical_actions(text_result) == _canonical_actions(speech_result) assert _canonical_actions(text_result) == [ { "domain": "seat", "capability": "seat.start_dynamic_mode", "target": { "device": "seat", "zone": "front_right", "instance": "passenger", }, "operation": "activate", "parameters": {"mode_id": 1}, } ] def test_text_and_speech_share_dynamic_mode_stop_command() -> None: processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) understanding = processor.process("关闭木马模式").to_dict() understanding["utterance_id"] = "mode-stop-equivalence" with tempfile.TemporaryDirectory() as directory, patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false", "COGNITIVE_AGENT_MODE_PROFILE_ROOT": str(Path(directory) / "modes"), }, clear=False, ): service = _mode_service(directory) text_result = service.process( "关闭木马模式", provider="local", conversation_id="conv-text-mode-stop", ) speech_result = service.process( understanding["clean_text"], provider="local", conversation_id="conv-speech-mode-stop", _speech_understanding=understanding, ) assert understanding["clean_text"] == "关闭木马模式。" assert understanding["execution_eligible"] is True assert text_result["status"] == speech_result["status"] == "succeeded" assert _canonical_actions(text_result) == _canonical_actions(speech_result) assert _canonical_actions(text_result)[0]["capability"] == "seat.stop_dynamic_mode" assert _canonical_actions(text_result)[0]["parameters"] == {} def test_explicit_incompatible_mode_target_fails_closed_for_both_inputs() -> None: processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) understanding = processor.process("将主驾座椅调整为木马模式。").to_dict() with tempfile.TemporaryDirectory() as directory, patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false", "COGNITIVE_AGENT_MODE_PROFILE_ROOT": str(Path(directory) / "modes"), }, clear=False, ): service = _mode_service(directory) text_result = service.process( "将主驾座椅调整为木马模式。", provider="local", conversation_id="conv-text-mode-rejected", ) assert understanding["semantic_status"] == "recognized_unsupported" assert understanding["execution_eligible"] is False assert text_result["status"] == "needs_confirmation" assert _canonical_actions(text_result) == [] def test_text_and_speech_share_colloquial_multidomain_commands() -> None: utterance = "先打开主驾座椅通风，再打开副驾座椅通风，最后把空调调成二十四度" processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT) understanding = processor.process(utterance).to_dict() understanding["utterance_id"] = "multidomain-equivalence" with tempfile.TemporaryDirectory() as directory, patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false", }, clear=False, ): service = _mode_service(directory) text_result = service.process( utterance, provider="local", conversation_id="conv-text-multidomain", ) speech_result = service.process( understanding["clean_text"], provider="local", conversation_id="conv-speech-multidomain", _speech_understanding=understanding, ) assert understanding["clean_text"] == ( "打开主驾座椅通风，然后打开副驾座椅通风，然后将空调设置为24度。" ) assert understanding["execution_eligible"] is True assert text_result["status"] == speech_result["status"] == "succeeded" assert _sorted_canonical_actions(text_result) == _sorted_canonical_actions( speech_result ) assert { action["capability"] for action in _canonical_actions(text_result) } == { "seat.set_driver_ventilation_native", "seat.set_ventilation", "climate.set_temperature", }