from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch import pytest from cognitive_agent.application.assistant_capabilities import ( AssistantCapabilityCatalog, ) from cognitive_agent.application.codex_assistant import CodexAssistantService from cognitive_agent.application.mode_profile import ModeProfile, ModeProfileRegistry from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService READING_OPERATIONS: list[dict[str, object]] = [ { "domain": "climate", "capability": "climate.set_power", "target_scope": "vehicle", "operation": "set", "parameters": {"state": "on"}, "reason": "阅读时保持空调开启。", }, { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": {"temperature_c": 24}, "reason": "阅读模式默认温度。", }, { "domain": "roof", "capability": "roof.set_shade_position", "target_scope": "vehicle", "operation": "set", "parameters": {"position_pct": 30}, "reason": "减少阅读眩光。", }, { "domain": "interior_light", "capability": "interior_light.set_reading_power", "target_scope": "all", "operation": "set", "parameters": {"state": "on"}, "reason": "打开阅读灯。", }, ] @pytest.fixture def vehicle_service(tmp_path: Path) -> SeatDemoService: service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = tmp_path / "runs" service.mode_profiles = ModeProfileRegistry( REPOSITORY_ROOT / "config" / "modes", user_root=tmp_path / "user-modes", ) return service def _save_reading_mode(service: SeatDemoService) -> ModeProfile: return service.mode_profiles.save_mode( "阅读模式", READING_OPERATIONS, mode_id="user-reading", aliases=["看书模式", "阅读空间模式"], vehicle_id="denza_n7", description="用户保存的阅读场景。", instruction="空调24度，遮阳帘30%，打开阅读灯", ) def _save_office_override(service: SeatDemoService) -> ModeProfile: preset = service.mode_profiles.get_profile("office", "denza_n7") return service.mode_profiles.save_mode( preset.name, list(preset.operations), mode_id=preset.id, aliases=list(preset.aliases), vehicle_id="denza_n7", description=preset.description, icon=preset.icon, instruction=preset.instruction, command_flow=preset.command_flow, timeline=preset.timeline, ) def _temperature(profile: ModeProfile, scope: str) -> object: return next( operation.parameters["temperature_c"] for operation in profile.operations if operation.capability == "climate.set_temperature" and operation.target_scope == scope ) @pytest.mark.parametrize( ("message", "profile_factory", "mode_id", "target_scope", "temperature"), ( ( "把阅读模式中空调温度调整到26度", _save_reading_mode, "user-reading", "vehicle", 26, ), ( "调整办公模式，把办公模式的副驾空调温度调整为26度。", _save_office_override, "office", "passenger", 26, ), ( "将看书模式里的空调温度改成25度", _save_reading_mode, "user-reading", "vehicle", 25, ), ( "请帮我把“阅读模式”里的空调温度改成25度吧。", _save_reading_mode, "user-reading", "vehicle", 25, ), ), ) def test_saved_mode_natural_language_update_is_control_plane_only( vehicle_service: SeatDemoService, message: str, profile_factory: object, mode_id: str, target_scope: str, temperature: int, ) -> None: assert callable(profile_factory) profile_factory(vehicle_service) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("saved-mode update acquired the hardware lease"), ): result = vehicle_service.preview_native_can( message, provider="local", conversation_id=f"conv-saved-mode-update-{temperature}-{mode_id}", execution_mode="hardware", vehicle_id="denza_n7", ) updated = vehicle_service.mode_profiles.get_profile(mode_id, "denza_n7") assert result["status"] == "succeeded" assert result["orchestration_route"] == "mode_profile_update" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["mode_profile"]["id"] == mode_id assert result["command_plans"] == {} assert result["can_commands"] == [] assert _temperature(updated, target_scope) == temperature assert sum( operation.capability == "climate.set_temperature" and operation.target_scope == target_scope for operation in updated.operations ) == 1 def test_builtin_mode_update_is_copy_on_write_and_delete_reveals_preset( vehicle_service: SeatDemoService, ) -> None: preset = vehicle_service.mode_profiles.get_profile("office", "denza_n7") preset_path = Path(preset.source) original_bytes = preset_path.read_bytes() original_temperature = _temperature(preset, "passenger") updated = vehicle_service.preview_native_can( "把办公模式中的副驾空调温度调整到26度", provider="local", conversation_id="conv-office-copy-on-write", execution_mode="hardware", vehicle_id="denza_n7", ) effective = vehicle_service.mode_profiles.get_profile("office", "denza_n7") public = next( item for item in vehicle_service.mode_profiles.public_modes("denza_n7") if item["id"] == "office" ) assert updated["status"] == "succeeded" assert updated["mode_profile"]["origin"] == "customized" assert effective.user_defined is True assert _temperature(effective, "passenger") == 26 assert preset_path.read_bytes() == original_bytes assert public["origin"] == "customized" deleted = vehicle_service.preview_native_can( "删除办公模式", provider="local", conversation_id="conv-office-copy-on-write-delete", execution_mode="hardware", vehicle_id="denza_n7", ) restored = vehicle_service.mode_profiles.get_profile("office", "denza_n7") assert deleted["status"] == "succeeded" assert restored.user_defined is False assert _temperature(restored, "passenger") == original_temperature assert preset_path.read_bytes() == original_bytes @pytest.mark.parametrize( ("message", "capability", "target", "parameter", "expected"), ( ( "把阅读模式中的阅读灯关闭", "interior_light.set_reading_power", "all", "state", "off", ), ( "把阅读模式中的遮阳帘调整到80%", "roof.set_shade_position", "vehicle", "position_pct", 80, ), ( "把阅读模式中的左前车窗调整到20%", "window.set_position", "front_left", "position_pct", 20, ), ( "把阅读模式中的主驾座椅通风调整到2档", "seat.set_driver_ventilation_native", "driver", "level", 2, ), ), ) def test_saved_mode_update_persists_native_can_capability_families( vehicle_service: SeatDemoService, message: str, capability: str, target: str, parameter: str, expected: object, ) -> None: _save_reading_mode(vehicle_service) result = vehicle_service.preview_native_can( message, provider="local", conversation_id=f"conv-native-mode-update-{parameter.replace('_', '-')}", execution_mode="hardware", vehicle_id="denza_n7", ) updated = vehicle_service.mode_profiles.get_profile( "user-reading", "denza_n7", ) operation = next( item for item in updated.operations if item.capability == capability and item.target_scope == target ) assert result["status"] == "succeeded" assert result["deferred_requests"] == [] assert operation.parameters[parameter] == expected def test_failed_saved_mode_update_is_atomic_and_reports_the_missing_mode( vehicle_service: SeatDemoService, ) -> None: _save_reading_mode(vehicle_service) original = vehicle_service.mode_profiles.get_profile( "user-reading", "denza_n7" ) original_revision = vehicle_service.mode_profiles.profile_revision(original) missing = vehicle_service.preview_native_can( "把不存在模式中的空调温度调整到26度", provider="local", conversation_id="conv-saved-mode-update-missing", execution_mode="hardware", vehicle_id="denza_n7", ) invalid = vehicle_service.preview_native_can( "把阅读模式中的曲速引擎调整到100%", provider="local", conversation_id="conv-saved-mode-update-invalid", execution_mode="hardware", vehicle_id="denza_n7", ) unchanged = vehicle_service.mode_profiles.get_profile( "user-reading", "denza_n7" ) assert missing["status"] == "needs_confirmation" assert missing["orchestration_route"] == "mode_profile_update" assert any( item.get("code") == "MODE_PROFILE_NOT_FOUND" for item in missing["validation"] ) assert invalid["status"] != "succeeded" assert invalid["orchestration_route"] == "mode_profile_update" assert invalid["mode_profile"] is None assert vehicle_service.mode_profiles.profile_revision(unchanged) == original_revision @pytest.mark.parametrize( "message", ( "删除阅读模式", "请帮我删除一下看书模式吧", "把“阅读模式”删掉。", ), ) def test_saved_user_mode_can_be_deleted_by_name_or_alias_without_hardware( vehicle_service: SeatDemoService, message: str, ) -> None: saved = _save_reading_mode(vehicle_service) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("saved-mode deletion acquired the hardware lease"), ): result = vehicle_service.preview_native_can( message, provider="local", conversation_id="conv-saved-mode-delete-user", execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "succeeded" assert result["orchestration_route"] == "mode_profile_delete" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["deleted_mode_profile"]["id"] == saved.id assert result["deleted_mode_profile"]["name"] == "阅读模式" assert result["command_plans"] == {} assert result["can_commands"] == [] assert vehicle_service.mode_profiles.resolve_exact_name( "阅读模式", "denza_n7" ) is None def test_delete_missing_and_builtin_modes_fails_closed_without_changing_registry( vehicle_service: SeatDemoService, ) -> None: preset = vehicle_service.mode_profiles.get_profile("office", "denza_n7") preset_revision = vehicle_service.mode_profiles.profile_revision(preset) missing = vehicle_service.preview_native_can( "删除不存在模式", provider="local", conversation_id="conv-saved-mode-delete-missing", execution_mode="hardware", vehicle_id="denza_n7", ) protected = vehicle_service.preview_native_can( "删除办公模式", provider="local", conversation_id="conv-saved-mode-delete-protected", execution_mode="hardware", vehicle_id="denza_n7", ) assert missing["status"] == "needs_confirmation" assert missing["orchestration_route"] == "mode_profile_delete" assert any( item.get("code") == "MODE_PROFILE_NOT_FOUND" for item in missing["validation"] ) assert protected["status"] == "rejected" assert protected["orchestration_route"] == "mode_profile_delete" assert any( item.get("code") == "MODE_PROFILE_DELETE_PROTECTED" for item in protected["validation"] ) remaining = vehicle_service.mode_profiles.get_profile("office", "denza_n7") assert vehicle_service.mode_profiles.profile_revision(remaining) == preset_revision @pytest.mark.parametrize( "message", ( "阅读模式中会控制哪些设备，执行哪些动作？", "看书模式里有哪些设备和参数", "阅读空间模式会做什么？", ), ) def test_saved_mode_details_are_answered_from_the_live_registry_without_model( vehicle_service: SeatDemoService, message: str, ) -> None: _save_reading_mode(vehicle_service) catalog = AssistantCapabilityCatalog(vehicle_service) snapshot = catalog.snapshot("denza_n7") answer = catalog.fast_answer(message, snapshot) assert answer is not None assert answer.topic == "specific_mode_detail" assert "阅读模式" in answer.answer assert "空调" in answer.answer assert "24" in answer.answer assert "遮阳帘" in answer.answer assert "阅读灯" in answer.answer assert "未下发车辆控制" in answer.answer def test_saved_mode_detail_query_reports_an_unknown_exact_name_without_guessing( vehicle_service: SeatDemoService, ) -> None: _save_reading_mode(vehicle_service) catalog = AssistantCapabilityCatalog(vehicle_service) snapshot = catalog.snapshot("denza_n7") answer = catalog.fast_answer("火星模式包含哪些设备和动作？", snapshot) assert answer is not None assert answer.topic == "specific_mode_detail_not_found" assert "火星模式" in answer.answer assert "未找到" in answer.answer assert "未下发车辆控制" in answer.answer assert "阅读模式" not in answer.answer @pytest.mark.parametrize( "message", ( "把阅读模式中空调温度调整到26度", "删除阅读模式", "执行阅读模式", "把全车空调温度调整到26度", ), ) def test_saved_mode_catalog_answer_never_intercepts_mutation_or_execution( vehicle_service: SeatDemoService, message: str, ) -> None: _save_reading_mode(vehicle_service) catalog = AssistantCapabilityCatalog(vehicle_service) snapshot = catalog.snapshot("denza_n7") assert catalog.fast_answer(message, snapshot) is None def _assistant_turn( service: SeatDemoService, message: str, conversation_id: str, ) -> tuple[dict[str, object], list[dict[str, object]], MagicMock]: client = MagicMock() client.available = False client.status = "unavailable" assistant = CodexAssistantService(client) catalog = AssistantCapabilityCatalog(service) snapshot = catalog.snapshot("denza_n7") events: list[dict[str, object]] = [] def execute_control( instruction: str, progress: object, _assistant_plan: object = None, ) -> dict[str, object]: return service.preview_native_can( instruction, provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", on_commands_validated=progress if callable(progress) else None, ) response = assistant.stream( message, conversation_id=conversation_id, principal="saved-mode-management-test", history=[], emit=lambda event: events.append(event) or 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), query_capabilities=catalog.query, ) return response, events, client def test_assistant_updates_saved_mode_without_waiting_for_codex_tool_call( vehicle_service: SeatDemoService, ) -> None: _save_reading_mode(vehicle_service) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("assistant update acquired the hardware lease"), ): response, events, client = _assistant_turn( vehicle_service, "把阅读模式中空调温度调整到26度", "conv-assistant-saved-mode-update", ) assert response["route"] == "mode_profile_update" assert response["success"] is True assert "阅读模式" in response["answer"] assert "26" in response["answer"] assert "车辆控制未完成" not in response["answer"] assert events[0]["route"] == "mode_profile_update" assert events[0]["engine"] == "n7web-mode-profile-manager" assert events[-1]["route"] == "mode_profile_update" assert next(event for event in events if event["type"] == "action.completed")[ "success" ] is True assert _temperature( vehicle_service.mode_profiles.get_profile("user-reading", "denza_n7"), "vehicle", ) == 26 client.start_thread.assert_not_called() client.start_turn.assert_not_called() def test_assistant_deletes_saved_user_mode_without_waiting_for_codex_tool_call( vehicle_service: SeatDemoService, ) -> None: _save_reading_mode(vehicle_service) response, events, client = _assistant_turn( vehicle_service, "删除阅读模式", "conv-assistant-saved-mode-delete", ) assert response["route"] == "mode_profile_delete" assert response["success"] is True assert "阅读模式" in response["answer"] assert "删除" in response["answer"] assert "仍保留" not in response["answer"] assert events[0]["route"] == "mode_profile_delete" assert events[0]["engine"] == "n7web-mode-profile-manager" assert events[-1]["route"] == "mode_profile_delete" assert vehicle_service.mode_profiles.resolve_exact_name( "阅读模式", "denza_n7" ) is None client.start_thread.assert_not_called() client.start_turn.assert_not_called()