from __future__ import annotations import json from concurrent.futures import ThreadPoolExecutor from pathlib import Path from threading import Barrier from unittest.mock import MagicMock, patch import pytest from cognitive_agent.application.codex_assistant import CodexAssistantService from cognitive_agent.application.assistant_capabilities import AssistantCapabilityCatalog from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.seat_service import ( REPOSITORY_ROOT, SeatDemoService, ) NAP_REQUEST = "构建一个小憩模式" INTERNAL_SCENE_PLACEHOLDER = "请根据该场景名称使用当前可控原子能力构建并执行场景" EXPECTED_NAP_CAPABILITIES = { "seat.set_longitudinal_position", "seat.set_cushion_tilt", "seat.set_backrest_recline", "seat.set_legrest_height", "climate.set_temperature", "climate.set_fan_level", "climate.set_cycle_mode", "roof.set_shade_position", "ambient_light.set_power", } def _scene_request( domain: str, capability: str, target_scope: str, parameters: dict[str, object], ) -> dict[str, object]: return { "domain": domain, "capability": capability, "target_scope": target_scope, "operation": "set", "duration_ms": None, "parameters": [ {"name": name, "value": value} for name, value in parameters.items() ], "reason": "根据午休场景组合当前工程已登记能力。", } LUNCH_BREAK_SCENE_PLAN = { "detected_domains": ["seat", "climate", "roof", "ambient_light"], "requests": [ _scene_request( "seat", "seat.set_backrest_recline", "passenger", {"position_pct": 70}, ), _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": 23}, ), _scene_request( "climate", "climate.set_fan_level", "vehicle", {"level": 1}, ), _scene_request( "roof", "roof.set_shade_position", "vehicle", {"position_pct": 0}, ), _scene_request( "ambient_light", "ambient_light.set_power", "vehicle", {"state": "off"}, ), ], "ambiguities": [], "decision_summary": ["组合安静、遮光且温度适宜的午休场景。"], "confidence": 0.92, } @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 _build_nap_mode(service: SeatDemoService, conversation_id: str) -> dict[str, object]: with patch.object( service.openai, "interpret", side_effect=AssertionError( "a repository-backed named mode must not depend on a cloud model" ), ): return service.preview_native_can( NAP_REQUEST, provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", ) def _intent_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) assert all(isinstance(action, dict) for action in actions) return actions def test_name_only_nap_request_reuses_the_verified_repository_profile( vehicle_service: SeatDemoService, tmp_path: Path, ) -> None: preset = vehicle_service.mode_profiles.get_profile("nap", "denza_n7") result = _build_nap_mode(vehicle_service, "conv-nap-name-inference") assert result["status"] == "succeeded" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["orchestration_route"] == "mode_profile_create_preview" assert result["mode_profile"] is None draft = result["mode_profile_draft"] assert isinstance(draft, dict) assert draft["name"] == "小憩模式" assert draft["operation_count"] == len(preset.operations) == 9 assert set(draft["domains"]) == {"seat", "climate", "roof", "ambient_light"} assert draft["existing_mode_id"] == "nap" assert isinstance(draft["existing_mode_revision"], str) assert len(draft["existing_mode_revision"]) == 16 assert draft["generation_source"] == "registered_mode_profile" actions = _intent_actions(result) assert {action["capability"] for action in actions} == { operation.capability for operation in preset.operations } == EXPECTED_NAP_CAPABILITIES expected_targets_and_parameters = { operation.capability: ( "passenger" if operation.domain == "seat" and operation.target_scope == "front_right" else operation.target_scope, operation.parameters, ) for operation in preset.operations } actual_targets_and_parameters = { action["capability"]: ( action["target"].get("instance") or action["target"].get("zone"), action["parameters"], ) for action in actions } assert actual_targets_and_parameters == expected_targets_and_parameters assert not list((tmp_path / "user-modes").rglob("*.yaml")) public_result = json.dumps(result, ensure_ascii=False, default=str) assert INTERNAL_SCENE_PLACEHOLDER not in public_result def test_inferred_nap_actions_are_bounded_by_the_runtime_capability_catalog( vehicle_service: SeatDemoService, ) -> None: result = _build_nap_mode(vehicle_service, "conv-nap-capability-boundary") actions = _intent_actions(result) catalog = { capability["name"]: capability for capability in vehicle_service.runtime_info()["capabilities"] } for action in actions: capability = catalog[action["capability"]] assert capability.get("execution_support") != "observational_only" assert capability.get("physical_control_supported", True) is not False assert action["domain"] in {"seat", "climate", "roof", "ambient_light"} assert "door.set_vehicle_lock" not in { action["capability"] for action in actions } assert all("media" not in str(action["capability"]) for action in actions) def test_codex_composes_an_unregistered_name_only_scene_from_runtime_capabilities( vehicle_service: SeatDemoService, tmp_path: Path, ) -> None: with patch.object( vehicle_service.openai, "interpret", side_effect=AssertionError("the embedded Codex plan must not invoke legacy GPT"), ): result = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id="conv-codex-scene-composition", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=LUNCH_BREAK_SCENE_PLAN, ) draft = result["mode_profile_draft"] assert result["status"] == "succeeded" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["provider_used"] == "local_rules+codex_app_server" assert result["orchestration_route"] == "mode_profile_create_preview" assert draft["name"] == "午休模式" assert draft["operation_count"] == 5 assert draft["generation_source"] == "codex_scene_composition" assert "existing_mode_id" not in draft assert {action["capability"] for action in _intent_actions(result)} == { "seat.set_backrest_recline", "climate.set_temperature", "climate.set_fan_level", "roof.set_shade_position", "ambient_light.set_power", } assert not list((tmp_path / "user-modes").rglob("*.yaml")) trial = vehicle_service.trial_mode_profile_draft( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) assert trial["status"] == "succeeded" assert trial["execution_mode"] == "dry_run" assert trial["hardware_execution_allowed"] is False assert trial["orchestration_route"] == "mode_profile_draft_trial" assert trial["trial_source"] == "server_validated_preview_artifact" assert _intent_actions(trial) == _intent_actions(result) with pytest.raises(ValueError, match="名称不匹配"): vehicle_service.trial_mode_profile_draft( "另一个模式", str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) result_path = vehicle_service.artifact_root / str(draft["source_request_id"]) / "result.json" tampered = json.loads(result_path.read_text(encoding="utf-8")) tampered["mode_profile_draft"]["source_request_id"] = "req-another-draft" result_path.write_text(json.dumps(tampered, ensure_ascii=False), encoding="utf-8") with pytest.raises(ValueError, match="来源请求编号不匹配"): vehicle_service.trial_mode_profile_draft( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) def test_codex_scene_composition_still_rejects_capabilities_outside_the_registry( vehicle_service: SeatDemoService, ) -> None: invalid_plan = { **LUNCH_BREAK_SCENE_PLAN, "requests": [ _scene_request( "flight", "vehicle.fly", "vehicle", {"altitude_m": 100}, ) ], } result = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id="conv-codex-scene-unknown-capability", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=invalid_plan, ) assert result["status"] != "succeeded" assert result.get("mode_profile_draft") is None assert not _intent_actions(result) def test_codex_scene_composition_rejects_out_of_range_parameters_atomically( vehicle_service: SeatDemoService, ) -> None: invalid_plan = { **LUNCH_BREAK_SCENE_PLAN, "requests": [ _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": 99}, ), _scene_request( "ambient_light", "ambient_light.set_power", "vehicle", {"state": "off"}, ), ], } result = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id="conv-codex-scene-range-rejected", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=invalid_plan, ) assert result["status"] != "succeeded" assert result.get("mode_profile_draft") is None assert not _intent_actions(result) assert any("32" in str(issue.get("message") or "") for issue in result["validation"]) def test_codex_scene_rejects_fan_zero_before_native_compile_and_clears_all_evidence( vehicle_service: SeatDemoService, ) -> None: invalid_plan = { **LUNCH_BREAK_SCENE_PLAN, "detected_domains": ["climate", "ambient_light"], "requests": [ _scene_request( "climate", "climate.set_fan_level", "vehicle", {"level": 0}, ), _scene_request( "ambient_light", "ambient_light.set_power", "vehicle", {"state": "off"}, ), ], } result = vehicle_service.preview_native_can( "构建一个静享模式", provider="local", conversation_id="conv-codex-scene-fan-zero", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=invalid_plan, ) assert result["status"] != "succeeded" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result.get("mode_profile_draft") is None assert not _intent_actions(result) assert result["function_config"] is None assert result["function_configs"] == {} assert result["command_plan"] is None assert result["command_plans"] == {} assert result["scene_config"] is None assert result["execution_report"] is None assert not result["deferred_requests"] assert any("1-7" in str(issue.get("message") or "") for issue in result["validation"]) @pytest.mark.parametrize( ("case_id", "invalid_request"), [ ( "wrong-type", _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": "温暖"}, ), ), ( "unknown-parameter", _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": 24, "invented": 1}, ), ), ( "missing-required", _scene_request( "climate", "climate.set_temperature", "vehicle", {}, ), ), ( "invalid-target", _scene_request( "climate", "climate.set_temperature", "trunk", {"temperature_c": 24}, ), ), ( "relative-seat-delta", _scene_request( "seat", "seat.set_backrest_recline", "passenger", {"angle_delta_deg": 3}, ), ), ( "native-missing-required-axis", _scene_request( "outlet", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 50}, ), ), ( "conflicting-seat-representations", _scene_request( "seat", "seat.set_backrest_recline", "passenger", {"position_pct": 0, "angle_deg": 163}, ), ), ], ) def test_codex_scene_adapter_rejects_one_invalid_action_without_partial_draft( vehicle_service: SeatDemoService, case_id: str, invalid_request: dict[str, object], ) -> None: plan = { **LUNCH_BREAK_SCENE_PLAN, "requests": [ invalid_request, _scene_request( "ambient_light", "ambient_light.set_power", "vehicle", {"state": "off"}, ), ], } result = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id=f"conv-codex-scene-adapter-{case_id}", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=plan, ) assert result["status"] != "succeeded" assert result.get("mode_profile_draft") is None assert not _intent_actions(result) @pytest.mark.parametrize( "native_request", [ _scene_request( "interior_light", "interior_light.set_reading_power", "all", {"state": "off"}, ), _scene_request( "outlet", "outlet.set_switch_state", "driver_left", {"state": "close"}, ), _scene_request( "climate", "climate.set_control_mode", "vehicle", {"mode": "auto"}, ), ], ) def test_codex_scene_keeps_valid_native_can_actions_in_the_complete_draft( vehicle_service: SeatDemoService, native_request: dict[str, object], ) -> None: plan = { **LUNCH_BREAK_SCENE_PLAN, "requests": [ native_request, _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": 24}, ), ], } result = vehicle_service.preview_native_can( "构建一个静享模式", provider="local", conversation_id=( "conv-codex-scene-native-" + str(native_request["capability"]).replace(".", "-").replace("_", "-") ), execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=plan, ) draft = result["mode_profile_draft"] assert result["status"] == "succeeded" assert draft["operation_count"] == 2 assert {action["capability"] for action in _intent_actions(result)} == { str(native_request["capability"]), "climate.set_temperature", } assert not result["deferred_requests"] def test_codex_scene_promotes_rear_temperature_to_verified_native_can( vehicle_service: SeatDemoService, ) -> None: plan = { **LUNCH_BREAK_SCENE_PLAN, "detected_domains": ["climate"], "requests": [ _scene_request( "climate", "climate.set_temperature", "rear", {"temperature_c": 24}, ) ], } result = vehicle_service.preview_native_can( "构建一个后排舒适模式", provider="local", conversation_id="conv-codex-scene-rear-climate", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=plan, ) draft = result["mode_profile_draft"] assert result["status"] == "succeeded" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert draft["operation_count"] == 1 assert result["deferred_requests"] == [] assert set(result["function_configs"]) == {"native_climate"} actions = _intent_actions(result) assert [(action["capability"], action["target"]["zone"]) for action in actions] == [ ("climate.set_temperature", "rear") ] def test_codex_composed_scene_can_be_saved_and_replayed_without_reinterpretation( vehicle_service: SeatDemoService, tmp_path: Path, ) -> None: built = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id="conv-codex-scene-save", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=LUNCH_BREAK_SCENE_PLAN, ) draft = built["mode_profile_draft"] saved = vehicle_service.create_mode_profile_card( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) mode_id = str(saved["mode_profile"]["id"]) assert saved["status"] == "succeeded" assert saved["mode_profile"]["name"] == "午休模式" assert saved["mode_profile"]["operation_count"] == 5 assert list((tmp_path / "user-modes").rglob(f"{mode_id}.yaml")) with patch.object( vehicle_service, "_interpret", side_effect=AssertionError("saved scene must replay trusted command flow"), ): replayed = vehicle_service.execute_mode_profile_card( mode_id, conversation_id="conv-codex-scene-replay", execution_mode="dry_run", vehicle_id="denza_n7", ) assert replayed["status"] == "succeeded" assert replayed["provider_used"] == "saved_mode_command_flow" assert {action["capability"] for action in _intent_actions(replayed)} == { request["capability"] for request in LUNCH_BREAK_SCENE_PLAN["requests"] } def test_inferred_nap_draft_can_be_confirmed_saved_and_replayed_without_reinterpretation( vehicle_service: SeatDemoService, tmp_path: Path, ) -> None: built = _build_nap_mode(vehicle_service, "conv-nap-build-save") draft = built["mode_profile_draft"] assert isinstance(draft, dict) saved = vehicle_service.save_mode_profile_card( "nap", str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) profile = saved["mode_profile"] assert saved["status"] == "succeeded" assert profile["id"] == "nap" assert profile["name"] == "小憩模式" assert profile["origin"] == "customized" assert profile["operation_count"] == 9 assert ( tmp_path / "user-modes" / "vehicles" / "denza_n7" / "user" / "nap.yaml" ).is_file() with patch.object( vehicle_service, "_interpret", side_effect=AssertionError("saved mode must replay its trusted command flow"), ): executed = vehicle_service.execute_mode_profile_card( "nap", conversation_id="conv-nap-saved-execution", execution_mode="dry_run", vehicle_id="denza_n7", ) assert executed["status"] == "succeeded" assert executed["provider_used"] == "saved_mode_command_flow" assert executed["orchestration_route"] == "saved_mode_command_flow" assert executed["execution_report"]["spec"]["mode"] == "dry_run" replayed_capabilities = { operation["capability"] for config in executed["function_configs"].values() for operation in config["spec"]["operations"] } assert replayed_capabilities == EXPECTED_NAP_CAPABILITIES def test_assistant_name_only_nap_request_stays_useful_without_codex( vehicle_service: SeatDemoService, tmp_path: Path, ) -> None: codex_client = MagicMock() codex_client.available = False assistant = CodexAssistantService(codex_client) events: list[dict[str, object]] = [] def execute_control( instruction: str, progress: object, ) -> dict[str, object]: callback = progress if callable(progress) else None return vehicle_service.preview_native_can( instruction, provider="local", conversation_id="conv-nap-assistant-route", execution_mode="hardware", vehicle_id="denza_n7", on_commands_validated=callback, ) response = assistant.stream( NAP_REQUEST, conversation_id="conv-nap-assistant-route", principal="mode-regression-test", history=[], emit=lambda event: events.append(event) or True, execute_control=execute_control, capability_snapshot=AssistantCapabilityCatalog(vehicle_service).snapshot(), ) assert response["route"] == "vehicle_control" assert response["success"] is True assert response["result"]["mode_profile_draft"]["operation_count"] == 9 assert response["result"]["mode_profile_draft"]["existing_mode_id"] == "nap" assert ( response["result"]["mode_profile_draft"]["generation_source"] == "registered_mode_profile" ) assert "小憩模式" in response["answer"] assert "预览" in response["answer"] assert "更新现有模式" in response["answer"] assert "功能模式未生成" not in response["answer"] assert INTERNAL_SCENE_PLACEHOLDER not in json.dumps( events, ensure_ascii=False, default=str, ) assert not list((tmp_path / "user-modes").rglob("*.yaml")) codex_client.start_turn.assert_not_called() def test_existing_mode_draft_is_bound_to_its_update_endpoint( vehicle_service: SeatDemoService, ) -> None: built = _build_nap_mode(vehicle_service, "conv-nap-update-binding") draft = built["mode_profile_draft"] assert isinstance(draft, dict) with pytest.raises(ValueError): vehicle_service.create_mode_profile_card( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) with pytest.raises(ValueError): vehicle_service.save_mode_profile_card( "office", str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) def test_concurrent_existing_mode_saves_use_revision_compare_and_swap( vehicle_service: SeatDemoService, ) -> None: drafts = [] for index, temperature in enumerate((24, 25), start=1): result = vehicle_service.preview_native_can( f"创建一个小憩模式，空调调整到{temperature}度", provider="local", conversation_id=f"conv-mode-cas-update-{index}", execution_mode="hardware", vehicle_id="denza_n7", ) drafts.append(result["mode_profile_draft"]) assert drafts[0]["existing_mode_revision"] == drafts[1]["existing_mode_revision"] barrier = Barrier(2) original_save = vehicle_service.mode_profiles.save_mode def synchronized_save(*args: object, **kwargs: object) -> object: barrier.wait(timeout=5) return original_save(*args, **kwargs) def persist(draft: dict[str, object]) -> tuple[str, str]: try: result = vehicle_service.save_mode_profile_card( "nap", str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) return "saved", str(result["mode_profile"]["id"]) except ValueError as exc: return "rejected", str(exc) with patch.object( vehicle_service.mode_profiles, "save_mode", side_effect=synchronized_save, ): with ThreadPoolExecutor(max_workers=2) as executor: outcomes = list(executor.map(persist, drafts)) assert [status for status, _detail in outcomes].count("saved") == 1 assert [status for status, _detail in outcomes].count("rejected") == 1 assert any("已被其他会话更新" in detail for _status, detail in outcomes) def test_concurrent_same_name_creates_are_create_if_absent( vehicle_service: SeatDemoService, ) -> None: drafts = [] for index, temperature in enumerate((24, 25), start=1): plan = { **LUNCH_BREAK_SCENE_PLAN, "requests": [ _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": temperature}, ) ], } result = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id=f"conv-mode-cas-create-{index}", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=plan, ) drafts.append(result["mode_profile_draft"]) barrier = Barrier(2) original_save = vehicle_service.mode_profiles.save_mode def synchronized_save(*args: object, **kwargs: object) -> object: barrier.wait(timeout=5) return original_save(*args, **kwargs) def persist(draft: dict[str, object]) -> tuple[str, str]: try: result = vehicle_service.create_mode_profile_card( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) return "saved", str(result["mode_profile"]["id"]) except ValueError as exc: return "rejected", str(exc) with patch.object( vehicle_service.mode_profiles, "save_mode", side_effect=synchronized_save, ): with ThreadPoolExecutor(max_workers=2) as executor: outcomes = list(executor.map(persist, drafts)) assert [status for status, _detail in outcomes].count("saved") == 1 assert [status for status, _detail in outcomes].count("rejected") == 1 assert any("已由其他会话创建" in detail for _status, detail in outcomes) def test_name_only_mode_does_not_bypass_exact_resolution_with_substring_match( vehicle_service: SeatDemoService, ) -> None: result = vehicle_service.preview_native_can( "构建一个我的小憩模式", provider="local", conversation_id="conv-mode-substring-rejected", execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "needs_confirmation" assert result["matched_modes"] == [] assert result.get("mode_profile_draft") is None assert not _intent_actions(result) def test_name_only_mode_does_not_combine_profiles_when_exact_name_is_ambiguous( vehicle_service: SeatDemoService, ) -> None: nap = vehicle_service.mode_profiles.get_profile("nap", "denza_n7") vehicle_service.mode_profiles.save_mode( "午休模式", list(nap.operations), mode_id="duplicate-nap-name", aliases=["小憩模式"], vehicle_id="denza_n7", ) result = vehicle_service.preview_native_can( NAP_REQUEST, provider="local", conversation_id="conv-mode-ambiguous-rejected", execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "needs_confirmation" assert result["matched_modes"] == [] assert result.get("mode_profile_draft") is None assert not _intent_actions(result) def test_explicit_actions_for_existing_name_are_bound_to_that_mode_update( vehicle_service: SeatDemoService, ) -> None: result = vehicle_service.preview_native_can( "创建一个小憩模式，空调调整到24度", provider="local", conversation_id="conv-existing-mode-explicit-action", execution_mode="hardware", vehicle_id="denza_n7", ) draft = result["mode_profile_draft"] assert result["status"] == "succeeded" assert draft["existing_mode_id"] == "nap" assert draft["generation_source"] == "explicit_scene_requirements" assert draft["operation_count"] == 1 answer = CodexAssistantService._mode_profile_preview_answer(result, True) assert "按你提供的设置" in answer assert "载入预览" not in answer saved = vehicle_service.save_mode_profile_card( "nap", str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) assert saved["status"] == "succeeded" assert saved["mode_profile"]["id"] == "nap" assert saved["mode_profile"]["operation_count"] == 1 def test_new_mode_draft_cannot_overwrite_an_unrelated_existing_mode( vehicle_service: SeatDemoService, ) -> None: result = vehicle_service.preview_native_can( "创建一个清凉测试模式，空调调整到24度", provider="local", conversation_id="conv-new-mode-update-rejected", execution_mode="hardware", vehicle_id="denza_n7", ) draft = result["mode_profile_draft"] assert result["status"] == "succeeded" assert "existing_mode_id" not in draft with pytest.raises(ValueError, match="新功能模式预览不能"): vehicle_service.save_mode_profile_card( "nap", str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) def test_exact_mode_resolver_supports_aliases_but_rejects_substrings_and_ambiguity( vehicle_service: SeatDemoService, ) -> None: registry = vehicle_service.mode_profiles nap = registry.get_profile("nap", "denza_n7") assert registry.resolve_exact_name("小憩模式", "denza_n7") == nap assert registry.resolve_exact_name("小憩", "denza_n7") == nap assert registry.resolve_exact_name("NAP", "denza_n7") == nap assert registry.resolve_exact_name("我的小憩模式", "denza_n7") is None assert registry.resolve_exact_name("小憩模式", "unsupported_vehicle") is None registry.save_mode( "午休模式", list(nap.operations), mode_id="duplicate-nap-alias", aliases=["小憩"], vehicle_id="denza_n7", ) assert registry.resolve_exact_name("小憩", "denza_n7") is None assert registry.resolve_exact_name("小憩模式", "denza_n7").id == "nap" def test_existing_mode_alias_keeps_the_canonical_name_when_updated( vehicle_service: SeatDemoService, ) -> None: result = vehicle_service.preview_native_can( "构建一个摇篮模式", provider="local", conversation_id="conv-mode-alias-canonical-name", execution_mode="hardware", vehicle_id="denza_n7", ) draft = result["mode_profile_draft"] assert result["status"] == "succeeded" assert draft["existing_mode_id"] == "dynamic_cradle" assert draft["name"] == "动态摇篮模式" saved = vehicle_service.save_mode_profile_card( "dynamic_cradle", str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) assert saved["mode_profile"]["name"] == "动态摇篮模式" def test_internal_scene_builder_placeholder_is_never_returned_to_the_user() -> None: result = { "status": "needs_confirmation", "mode_creation_failure": { "code": "MODE_PROFILE_DRAFT_INCOMPLETE", "reason": ( f"设置“{INTERNAL_SCENE_PLACEHOLDER}”未被识别为当前支持的座舱控制动作" ), "next_step": "请按设备、动作和目标值重新说明", "unresolved_clauses": [INTERNAL_SCENE_PLACEHOLDER], }, "unresolved_clauses": [INTERNAL_SCENE_PLACEHOLDER], } answer = CodexAssistantService._mode_profile_preview_answer(result, False) safe_result = CodexAssistantService._safe_vehicle_result(result) assert INTERNAL_SCENE_PLACEHOLDER not in answer assert INTERNAL_SCENE_PLACEHOLDER not in json.dumps( safe_result, ensure_ascii=False, ) assert "内部" not in answer assert "失败原因" in answer assert "下一步" in answer def test_model_cannot_create_a_mode_with_an_unregistered_capability( vehicle_service: SeatDemoService, tmp_path: Path, ) -> None: hallucinated_plan = { "detected_domains": ["flight"], "requests": [ { "domain": "flight", "capability": "vehicle.fly", "target_scope": "vehicle", "operation": "set", "parameters": [{"name": "altitude_m", "value": 100}], "reason": "模型臆造的飞行能力。", } ], "ambiguities": [], "decision_summary": ["切换飞行模式。"], "confidence": 0.99, } with patch.object( vehicle_service.openai, "interpret", return_value=hallucinated_plan, ): result = vehicle_service.process( "构建一个飞行模式", "openai", "conv-hallucinated-mode", execution_mode="dry_run", vehicle_id="denza_n7", ) assert result["status"] != "succeeded" assert result.get("mode_profile_draft") is None assert result.get("mode_profile") is None assert not _intent_actions(result) assert any( issue["severity"] == "error" for issue in result.get("validation", []) ) assert not list((tmp_path / "user-modes").rglob("*.yaml"))