from __future__ import annotations import json from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from cognitive_agent.application.assistant_capabilities import ( AssistantCapabilityCatalog, ) from cognitive_agent.application.codex_assistant import ( CodexAssistantService, _catalog_exact_scene_template_plan, ) from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.scene_composer import ( normalize_scene_output, scene_context_for_mode_creation, scene_required_native_fallbacks, scene_required_peripheral_refs, scene_semantic_guidance, ) from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService from cognitive_agent.application.self_extension import ( ExtensionIntegrationError, SelfExtensionService, SimulationGateway, ) from cognitive_agent.server import SeatDemoHandler, VehicleDisplaySync def _learned_light() -> dict[str, object]: return { "id": "ir-slot:romance-light:7", "status": "learned", "enabled": True, "device_id": "romance-light", "device_name": "氛围灯带", "object_names": ["氛围灯带", "灯带"], "device_type": "light", "card_uid": "2CC8B103", "action": "设置为红色", "action_names": ["设置为红色", "红色"], "risk": "low", "slot": 7, } def _learned_light_power_on() -> dict[str, object]: return { "id": "ir-slot:romance-light:1", "status": "learned", "enabled": True, "device_id": "romance-light", "device_name": "氛围灯带", "object_names": ["氛围灯带", "灯带"], "device_type": "light", "card_uid": "2CC8B103", "action": "打开", "action_names": ["打开", "开启", "点亮"], "risk": "low", "slot": 1, } def _received_cinema_scene() -> dict[str, object]: actions = [ { "type": "window_position", "at_ms": 0, "domain": "window", "capability": "window.set_position", "target_scope": "front_left", "operation": "set", "parameters": {"position_pct": 0}, }, { "type": "climate_temperature", "at_ms": 0, "domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": {"temperature_c": 25}, }, { "type": "seat_preset", "at_ms": 0, "domain": "local_vehicle", "capability": "local_scene.seat_preset", "target_scope": "vehicle", "operation": "set", "parameters": {"targets_pulse": [246, 112, 405, 342]}, }, { "type": "android_app_launch", "at_ms": 3000, "domain": "android", "capability": "android.launch_app", "target_scope": "headunit", "operation": "launch", "parameters": { "package_name": "com.byd.videoplay", "activity_name": "com.byd.videoplay.activity.WelComeActivity", }, }, ] return { "id": "cinema", "label": "影院模式", "name": "影院模式", "description": "接收到的车机影院模式。", "icon": "film", "aliases": ["观影模式"], "origin": "headunit", "deletable": True, "behavior": "static", "domains": ["window", "climate", "local_vehicle", "android"], "operation_count": len(actions), "scope": {"common": False, "vehicle_ids": ["denza_n7"]}, "profile_revision": "received-cinema-v1", "actions": actions, } def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light( tmp_path: Path, ) -> None: service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = tmp_path / "runs" service.mode_profiles = ModeProfileRegistry( REPOSITORY_ROOT / "config" / "modes", user_root=tmp_path / "modes", ) extension = SelfExtensionService( SimulationGateway(), catalog_db_path=tmp_path / "infrared.sqlite3", ) extension.sync_peripheral_catalog([_learned_light_power_on()]) received_cinema = _received_cinema_scene() service.import_n7_mode_scene(received_cinema) stale_profile = service.mode_profiles.save_mode( "浪漫影院模式", [ { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": {"temperature_c": 30}, } ], vehicle_id="denza_n7", instruction="进入浪漫影院模式", ) catalog = AssistantCapabilityCatalog( service, peripheral_provider=extension.list_peripheral_routing_capabilities, ) handler = object.__new__(SeatDemoHandler) handler.service = service handler.extension = extension handler.nfc = SimpleNamespace(snapshot=lambda: {"card": None}) handler.vehicle_display_sync = VehicleDisplaySync() client = MagicMock() client.available = True assistant = CodexAssistantService(client) snapshot = catalog.snapshot(vehicle_id="denza_n7") result: dict[str, object] = {} def execute_control( instruction: str, progress, assistant_plan: dict[str, object], ) -> dict[str, object]: nonlocal result arguments = { "utterance": instruction, "provider": "local", "conversation_id": "conv-romantic-cinema-template", "execution_mode": "hardware", "vehicle_id": "denza_n7", "simulation_fault": "normal", "hardware_access_bypass": True, "reasoning_effort": "low", "control_transport": "wired_can", "assistant_plan": assistant_plan, } result = handler._preview_mode_with_scene_peripherals( "/api/requests/preview", arguments, assistant_plan, on_commands_validated=progress, ) return result try: with patch.object( extension, "execute_gateway", side_effect=AssertionError("mode creation must not emit infrared"), ): response = assistant.stream( "创建浪漫影院模式", conversation_id="conv-romantic-cinema-template", principal="romantic-cinema-template-test", history=[], emit=lambda _event: True, execute_control=execute_control, capability_snapshot=snapshot, capability_context=catalog.compact_context(snapshot), scene_capability_context=catalog.scene_composition_context(snapshot), control_terms=catalog.control_terms(snapshot), ) assert response["success"] is True assert response["route"] == "scene_template" client.start_structured_turn.assert_not_called() client.start_turn.assert_not_called() assert result["status"] == "succeeded" assert result["mode_profile_draft"]["generation_source"] == "scene_template_copy" assert result["mode_profile_draft"]["existing_mode_id"] == stale_profile.id assert result["mode_profile_draft"]["operation_count"] == 3 native_actions = { ( action["capability"], action["target"].get("instance") or action["target"].get("zone"), ): action["parameters"] for action in result["intent_ir"]["spec"]["actions"] if action["domain"] != "self_extension" } assert native_actions == { (action["capability"], action["target_scope"]): action["parameters"] for action in received_cinema["actions"] if action["domain"] in {"window", "climate"} } assert result["n7_transfer_candidates"] == received_cinema["actions"][2:] assert [action["slot"] for action in result["infrared_actions"]] == [1] draft = result["mode_profile_draft"] saved = service.save_mode_profile_card( stale_profile.id, draft["name"], draft["instruction"], draft["source_request_id"], draft["vehicle_id"], ) profile = service.get_mode_profile_card(saved["mode_profile"]["id"], "denza_n7") saved_selectors = { (operation.capability, operation.target_scope) for operation in profile.operations } assert saved_selectors == { (action["capability"], action["target_scope"]) for action in received_cinema["actions"] } | {("infrared.emit", "romance-light")} _local_requests, transfer_actions = service._mode_requests_and_n7_transfers(profile) assert transfer_actions == received_cinema["actions"][2:] finally: service.close() def test_scene_composer_mixed_can_ir_preview_save_and_revalidate( tmp_path: Path, ) -> None: service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = tmp_path / "runs" service.mode_profiles = ModeProfileRegistry( REPOSITORY_ROOT / "config" / "modes", user_root=tmp_path / "modes", ) extension = SelfExtensionService( SimulationGateway(), catalog_db_path=tmp_path / "infrared.sqlite3", ) extension.sync_peripheral_catalog( [_learned_light_power_on(), _learned_light()] ) catalog = AssistantCapabilityCatalog( service, peripheral_provider=extension.list_peripheral_routing_capabilities, ) context = catalog.scene_composition_context(catalog.snapshot()) capability_ref = next( item["ref"] for item in context["peripheral_actions"] if "红色" in str(item.get("action_label") or "") ) encoded_context = json.dumps(context, ensure_ascii=False) assert "2CC8B103" not in encoded_context assert '"slot":7' not in encoded_context plan = normalize_scene_output( { "requests": [ { "capability": "climate.set_temperature", "target_scope": "vehicle", "parameters": {"temperature_c": 24}, }, {"capability_ref": capability_ref}, ], "ambiguities": [], "confidence": 0.94, }, context, ) handler = object.__new__(SeatDemoHandler) handler.service = service handler.extension = extension handler.nfc = SimpleNamespace(snapshot=lambda: {"card": None}) handler.vehicle_display_sync = VehicleDisplaySync() message = "创建一个星夜外设模式" arguments: dict[str, object] = { "utterance": message, "provider": "local", "conversation_id": "conv-scene-peripheral-flow", # The scene path must override this and never emit while creating. "execution_mode": "hardware", "vehicle_id": "denza_n7", "simulation_fault": "normal", "hardware_access_bypass": True, "reasoning_effort": "low", "control_transport": "wired_can", "assistant_plan": plan, } with patch.object( extension, "execute_gateway", side_effect=AssertionError("mode creation must not emit infrared"), ): preview = handler._preview_mode_with_scene_peripherals( "/api/requests/preview", arguments, plan, on_commands_validated=None, ) assert preview["status"] == "succeeded" assert preview["execution_mode"] == "dry_run" assert preview["hardware_execution_allowed"] is False assert preview["mode_profile_draft"]["domains"] == [ "climate", "self_extension", ] assert preview["mode_profile_draft"]["operation_count"] == 3 assert set(preview["function_configs"]) == {"native_climate", "self_extension"} infrared_commands = preview["command_plans"]["self_extension"]["spec"][ "commands" ] assert [command["message"] for command in infrared_commands] == [ "infrared.emit", "infrared.emit", ] assert [command["signals"] for command in infrared_commands] == [ {"slot": 1}, {"slot": 7}, ] assert preview["infrared_actions"][0]["parse_source"] == ( "scene_composer_capability_ref" ) assert "2CC8B103" not in json.dumps( CodexAssistantService._safe_vehicle_result(preview), ensure_ascii=False, ) assert "2CC8B103" not in json.dumps( CodexAssistantService._validated_control_payload(preview), ensure_ascii=False, ) draft = preview["mode_profile_draft"] saved = service.create_mode_profile_card( draft["name"], draft["instruction"], preview["request_id"], "denza_n7", ) profile = service.get_mode_profile_card(saved["mode_profile"]["id"], "denza_n7") assert [operation.capability for operation in profile.operations] == [ "climate.set_temperature", "infrared.emit", "infrared.emit", ] entries = handler._saved_mode_infrared_entries(profile) assert [entry["capability"]["slot"] for entry in entries] == [1, 7] native_execution = service.execute_mode_profile_card( profile.id, conversation_id="conv-saved-scene-peripheral", execution_mode="dry_run", vehicle_id="denza_n7", ) infrared_execution = handler._execute_local_peripheral_actions( profile.name, entries, provider="saved_mode_command_flow", conversation_id="conv-saved-scene-peripheral", execution_mode="dry_run", ) executed = handler._merge_can_and_peripheral_results( profile.name, native_execution, infrared_execution, entries, ) assert executed["status"] == "succeeded" assert [action["slot"] for action in executed["infrared_actions"]] == [1, 7] invoked = handler._execute_request_with_peripherals( "/api/requests/preview", { "utterance": f"进入{profile.name}", "provider": "local", "conversation_id": "conv-invoke-saved-peripheral-mode", "execution_mode": "hardware", "vehicle_id": "denza_n7", "simulation_fault": "normal", "hardware_access_bypass": True, "control_transport": "wired_can", }, ) assert invoked is not None assert invoked["status"] == "succeeded" assert invoked["execution_mode"] == "dry_run" assert [action["slot"] for action in invoked["infrared_actions"]] == [1, 7] template_plan = _catalog_exact_scene_template_plan( catalog.snapshot(), "浪漫模式", ) assert template_plan is not None template_external = normalize_scene_output( { "requests": [{"capability_ref": capability_ref}], "ambiguities": [], "confidence": 0.95, }, context, ) augmented_template = { **template_plan, "requests": [ *template_plan["requests"], *template_external["requests"], ], } template_context = scene_context_for_mode_creation( context, "创建一个浪漫模式", ) template_semantics = scene_semantic_guidance("浪漫模式") template_native_fallbacks = scene_required_native_fallbacks( "浪漫模式", template_context, include_default_bundle=True, ) template_semantics["required_native_domains"] = [ str(item["domain"]) for item in template_native_fallbacks ] template_semantics["native_fallback_requests"] = template_native_fallbacks template_semantics["peripheral_policy"] = { "required_refs": scene_required_peripheral_refs( "浪漫模式", template_context, include_default_bundle=True, ), } template_context["scene_semantics"] = template_semantics augmented_template = CodexAssistantService._with_required_scene_peripherals( augmented_template, template_context, ) template_preview = handler._preview_mode_with_scene_peripherals( "/api/requests/preview", { **arguments, "utterance": "创建一个浪漫模式", "conversation_id": "conv-romantic-template-peripheral", "assistant_plan": augmented_template, }, augmented_template, on_commands_validated=None, ) assert template_preview["status"] == "succeeded" assert template_preview["mode_profile_draft"]["generation_source"] == ( "codex_scene_composition" ) template_actions = template_preview["intent_ir"]["spec"]["actions"] assert template_preview["mode_profile_draft"]["operation_count"] == ( len(template_actions) ) assert { "climate", "ambient_light", "roof", "self_extension", }.issubset(template_preview["mode_profile_draft"]["domains"]) template_capabilities = { action["capability"] for action in template_actions } assert any( capability.startswith("climate.") for capability in template_capabilities ) assert "ambient_light.set_power" in template_capabilities assert "roof.set_shade_position" in template_capabilities assert "infrared.emit" in template_capabilities explicit = handler._execute_request_with_peripherals( "/api/requests/preview", { "utterance": ( "创建一个外设办公模式，空调温度设置为25度，" "氛围灯带设置为红色" ), "provider": "local", "conversation_id": "conv-explicit-peripheral-mode", "execution_mode": "hardware", "vehicle_id": "denza_n7", "simulation_fault": "normal", "hardware_access_bypass": True, "control_transport": "wired_can", }, ) assert explicit is not None assert explicit["status"] == "succeeded" assert explicit["execution_mode"] == "dry_run" assert explicit["mode_profile_draft"]["operation_count"] == 3 assert [action["slot"] for action in explicit["infrared_actions"]] == [1, 7] # Removing the durable NFC whitelist invalidates both cached refs and saved # mode execution before any transport can be reached. extension.sync_peripheral_catalog([]) with pytest.raises(ExtensionIntegrationError, match="未授权"): handler._saved_mode_infrared_entries(profile)