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.assistant_capabilities import AssistantCapabilityCatalog from cognitive_agent.application.codex_assistant import CodexAssistantService from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.mode_profile_intent import ( is_mode_profile_query, parse_mode_profile_detail_query, parse_mode_profile_update, ) from cognitive_agent.application.native_can_runtime import NativeCanCompileError 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, } VIEWING_SCENE_PLAN = { "detected_domains": ["ambient_light", "roof", "climate", "interior_light"], "requests": [ _scene_request( "ambient_light", "ambient_light.set_mode", "vehicle", {"mode": "cinema"}, ), _scene_request( "roof", "roof.set_shade_position", "vehicle", {"position_pct": 0}, ), _scene_request( "climate", "climate.set_temperature", "vehicle", {"temperature_c": 24}, ), _scene_request( "climate", "climate.set_fan_level", "vehicle", {"level": 2}, ), _scene_request( "climate", "climate.set_cycle_mode", "vehicle", {"mode": "internal"}, ), _scene_request( "interior_light", "interior_light.set_indoor_power", "vehicle", {"state": "off"}, ), ], "ambiguities": [], "decision_summary": ["组合遮光、低照明和舒适温度的观影场景。"], "confidence": 0.94, } @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 _build_viewing_mode( service: SeatDemoService, conversation_id: str, ) -> dict[str, object]: return service.preview_native_can( "构建一个观影模式", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=VIEWING_SCENE_PLAN, ) 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 _draft_reference(result: dict[str, object]) -> dict[str, str]: draft = result["mode_profile_draft"] assert isinstance(draft, dict) return { "name": str(draft["name"]), "source_request_id": str(draft["source_request_id"]), "vehicle_id": str(draft["vehicle_id"]), } def test_every_repository_mode_can_build_a_name_only_draft_without_exception( vehicle_service: SeatDemoService, ) -> None: modes = vehicle_service.mode_profiles.public_modes("denza_n7") assert modes for index, mode in enumerate(modes): labels = [str(mode["name"]), *map(str, mode.get("aliases", []))] label = next((item for item in labels if item.endswith("模式")), None) assert label is not None, mode["id"] result = vehicle_service.preview_native_can( f"构建一个{label}", provider="local", conversation_id=f"conv-name-only-mode-{index:02d}", execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "succeeded", mode["id"] assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["mode_profile_draft"]["operation_count"] > 0 def test_name_only_mode_native_compile_exception_returns_scrubbed_failure( vehicle_service: SeatDemoService, ) -> None: with patch.object( vehicle_service.native_can_runtime, "compile", side_effect=NativeCanCompileError("synthetic compiler rejection"), ): result = vehicle_service.preview_native_can( "构建一个办公模式", provider="local", conversation_id="conv-name-only-native-compile-failure", execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "validation_failed" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["orchestration_route"] == "mode_profile_create_preview" assert result.get("mode_profile_draft") is None assert result["function_configs"] == {} assert result["command_plans"] == {} assert result["mode_creation_failure"]["code"] == ( "MODE_PROFILE_NATIVE_COMPILE_FAILED" ) 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_pending_nap_draft_temperature_can_be_revised_without_saving_or_execution( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-nap-pending-draft-revision" original_profile = vehicle_service.mode_profiles.get_profile("nap", "denza_n7") original_revision = vehicle_service.mode_profiles.profile_revision(original_profile) built = _build_nap_mode(vehicle_service, conversation_id) built_draft = built["mode_profile_draft"] assert isinstance(built_draft, dict) revised = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到25度", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) draft = revised["mode_profile_draft"] assert revised["status"] == "succeeded" assert revised["execution_mode"] == "dry_run" assert revised["hardware_execution_allowed"] is False assert revised["orchestration_route"] == "mode_profile_create_preview" assert draft["name"] == "小憩模式" assert draft["operation_count"] == built_draft["operation_count"] == 9 assert draft["generation_source"] == "draft_revision" assert draft["draft_revision"] == 2 assert draft["revises_source_request_id"] == built_draft["source_request_id"] assert draft["source_request_id"] != built_draft["source_request_id"] assert draft["existing_mode_id"] == "nap" assert draft["existing_mode_revision"] == original_revision revised_actions = _intent_actions(revised) temperatures = [ action for action in revised_actions if action["capability"] == "climate.set_temperature" ] assert len(temperatures) == 1 assert temperatures[0]["parameters"]["temperature_c"] == 25 assert ( temperatures[0]["target"].get("zone") or temperatures[0]["target"].get("instance") ) == "vehicle" def sibling_signature(action: dict[str, object]) -> str: parameters = dict(action.get("parameters") or {}) if parameters.get("position_pct") is not None: parameters.pop("angle_deg", None) return json.dumps( { "domain": action.get("domain"), "capability": action.get("capability"), "target": action.get("target"), "parameters": parameters, }, ensure_ascii=False, sort_keys=True, ) assert { sibling_signature(action) for action in revised_actions if action["capability"] != "climate.set_temperature" } == { sibling_signature(action) for action in _intent_actions(built) if action["capability"] != "climate.set_temperature" } # Generating and revising a card is configuration-only. The registered # profile remains unchanged until the card's explicit save/update action. unchanged = vehicle_service.mode_profiles.get_profile("nap", "denza_n7") assert vehicle_service.mode_profiles.profile_revision(unchanged) == original_revision assert next( operation.parameters["temperature_c"] for operation in unchanged.operations if operation.capability == "climate.set_temperature" ) != 25 @pytest.mark.parametrize( ("message", "capability", "parameter", "expected"), ( ( "请帮我 把 小憩模式 的 空调温度调整到25度", "climate.set_temperature", "temperature_c", 25, ), ( "把小憩模式中的空调温度调整到25度", "climate.set_temperature", "temperature_c", 25, ), ( "把小憩模式里空调温度设置为25度", "climate.set_temperature", "temperature_c", 25, ), ( "小憩模式，空调温度调到25度。", "climate.set_temperature", "temperature_c", 25, ), ( "把小憩模式的遮阳帘关闭", "roof.set_shade_position", "position_pct", 0, ), ( "把小憩模式里的氛围灯打开", "ambient_light.set_power", "state", "on", ), ), ) def test_pending_draft_update_variants_never_enter_physical_execution( vehicle_service: SeatDemoService, message: str, capability: str, parameter: str, expected: object, ) -> None: conversation_id = "conv-nap-draft-variant-" + str(abs(hash(message)))[:12] built = _build_nap_mode(vehicle_service, conversation_id) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("pending draft update reached hardware lease"), ): revised = vehicle_service.preview_native_can( message, provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) assert revised["status"] == "succeeded" assert revised["execution_mode"] == "dry_run" assert revised["hardware_execution_allowed"] is False assert revised["orchestration_route"] == "mode_profile_create_preview" assert next( action["parameters"][parameter] for action in _intent_actions(revised) if action["capability"] == capability ) == expected def test_pending_draft_can_be_revised_repeatedly_from_latest_server_artifact( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-nap-pending-draft-chain" built = _build_nap_mode(vehicle_service, conversation_id) second = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到25度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) third = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到26度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(second), ) second_draft = second["mode_profile_draft"] third_draft = third["mode_profile_draft"] assert third_draft["draft_revision"] == 3 assert third_draft["revises_source_request_id"] == second_draft["source_request_id"] assert next( action["parameters"]["temperature_c"] for action in _intent_actions(third) if action["capability"] == "climate.set_temperature" ) == 26 def test_viewing_draft_shade_close_then_100_percent_is_an_atomic_dry_run_chain( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-viewing-shade-revision-chain" built = _build_viewing_mode(vehicle_service, conversation_id) assert built["status"] == "succeeded" assert built["mode_profile_draft"]["operation_count"] == 6 def semantic_actions(result: dict[str, object]) -> dict[tuple[str, str, str], object]: return { ( str(action["domain"]), str(action["capability"]), str( action["target"].get("instance") or action["target"].get("zone") ), ): action["parameters"] for action in _intent_actions(result) } with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("viewing draft revision reached hardware lease"), ): closed = vehicle_service.preview_native_can( "把观影模式的遮阳帘变成关闭状态。", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) opened = vehicle_service.preview_native_can( "把观影模式的遮阳帘变成100%", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(closed), ) built_actions = semantic_actions(built) closed_actions = semantic_actions(closed) opened_actions = semantic_actions(opened) shade_key = ("roof", "roof.set_shade_position", "vehicle") assert closed["status"] == opened["status"] == "succeeded" assert closed["execution_mode"] == opened["execution_mode"] == "dry_run" assert closed["hardware_execution_allowed"] is False assert opened["hardware_execution_allowed"] is False assert closed["orchestration_route"] == "mode_profile_create_preview" assert opened["orchestration_route"] == "mode_profile_create_preview" assert len(built_actions) == len(closed_actions) == len(opened_actions) == 6 assert closed_actions[shade_key] == {"position_pct": 0} assert opened_actions[shade_key] == {"position_pct": 100} assert {key: value for key, value in closed_actions.items() if key != shade_key} == { key: value for key, value in built_actions.items() if key != shade_key } assert {key: value for key, value in opened_actions.items() if key != shade_key} == { key: value for key, value in built_actions.items() if key != shade_key } closed_draft = closed["mode_profile_draft"] opened_draft = opened["mode_profile_draft"] assert closed_draft["draft_revision"] == 2 assert closed_draft["requested_operation_count"] == 1 assert closed_draft["changed_operation_count"] == 0 assert opened_draft["draft_revision"] == 3 assert opened_draft["revises_source_request_id"] == closed_draft[ "source_request_id" ] assert opened_draft["requested_operation_count"] == 1 assert opened_draft["changed_operation_count"] == 1 @pytest.mark.parametrize( ("change", "expected_position"), ( ("遮阳帘调整为1%", 1), ("遮阳帘调为20%", 20), ("遮阳帘改为30%", 30), ("遮阳帘改成40%", 40), ("遮阳帘变为50%", 50), ("遮阳帘变成99%", 99), ("遮阳帘换成100%", 100), ("遮阳帘设置为0%", 0), ("遮阳帘变成打开状态", 100), ("遮阳帘变成半开状态", 50), ), ) def test_viewing_draft_accepts_supported_shade_absolute_expressions( vehicle_service: SeatDemoService, change: str, expected_position: int, ) -> None: conversation_id = f"conv-viewing-shade-alias-{expected_position}-{len(change)}" built = _build_viewing_mode(vehicle_service, conversation_id) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("shade draft alias reached hardware lease"), ): revised = vehicle_service.preview_native_can( f"把观影模式的{change}", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) actions = _intent_actions(revised) assert revised["status"] == "succeeded" assert revised["execution_mode"] == "dry_run" assert revised["hardware_execution_allowed"] is False assert revised["mode_profile_draft"]["operation_count"] == 6 assert next( action["parameters"]["position_pct"] for action in actions if action["capability"] == "roof.set_shade_position" ) == expected_position def test_viewing_draft_updates_percent_integer_and_state_in_one_atomic_turn( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-viewing-multi-parameter-revision" built = _build_viewing_mode(vehicle_service, conversation_id) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("multi-parameter draft update reached hardware"), ): revised = vehicle_service.preview_native_can( "把观影模式的遮阳帘变成100%，空调温度变成25度，室内灯变成打开状态", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) by_capability = { str(action["capability"]): action["parameters"] for action in _intent_actions(revised) } assert revised["status"] == "succeeded" assert revised["execution_mode"] == "dry_run" assert revised["hardware_execution_allowed"] is False assert revised["mode_profile_draft"]["operation_count"] == 6 assert revised["mode_profile_draft"]["requested_operation_count"] == 3 assert revised["mode_profile_draft"]["changed_operation_count"] == 3 assert by_capability["roof.set_shade_position"] == {"position_pct": 100} assert by_capability["climate.set_temperature"] == {"temperature_c": 25} assert by_capability["interior_light.set_indoor_power"] == {"state": "on"} assert by_capability["ambient_light.set_mode"] == {"mode": "cinema"} assert by_capability["climate.set_fan_level"] == {"level": 2} assert by_capability["climate.set_cycle_mode"] == {"mode": "internal"} @pytest.mark.parametrize("position", (-1, 101)) def test_viewing_draft_rejects_out_of_range_shade_without_partial_evidence( vehicle_service: SeatDemoService, position: int, ) -> None: conversation_id = f"conv-viewing-invalid-shade-{abs(position)}" built = _build_viewing_mode(vehicle_service, conversation_id) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("invalid shade draft update reached hardware"), ): rejected = vehicle_service.preview_native_can( f"把观影模式的遮阳帘变成{position}%", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) assert rejected["status"] == "needs_confirmation" assert rejected["execution_mode"] == "dry_run" assert rejected["hardware_execution_allowed"] is False assert rejected["orchestration_route"] == "mode_profile_create_preview" assert rejected.get("mode_profile_draft") is None assert rejected["function_configs"] == {} assert rejected["command_plans"] == {} assert rejected["mode_creation_failure"]["code"] == ( "MODE_PROFILE_DRAFT_UPDATE_REJECTED" ) @pytest.mark.parametrize( "message", ( "把观影模式的遮阳帘变成100%吗", "把观影模式的遮阳帘调整为50%好吗？", "把观影模式的遮阳帘变成关闭状态吗。", "把观影模式的遮阳帘变成100%合适么", "观影模式里的遮阳帘应该变成多少呢", ), ) def test_viewing_draft_shade_question_never_updates_or_reaches_hardware( vehicle_service: SeatDemoService, message: str, ) -> None: conversation_id = "conv-viewing-shade-question" built = _build_viewing_mode(vehicle_service, conversation_id) reference = _draft_reference(built) assert ( parse_mode_profile_update(message, expected_mode_name=reference["name"]) is None ) assert is_mode_profile_query(message, expected_mode_name=reference["name"]) with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("shade question reached hardware lease"), ): result = vehicle_service.preview_native_can( message, provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=reference, ) assert result["status"] == "needs_confirmation" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["orchestration_route"] == "mode_profile_query" assert result.get("mode_profile_draft") is None assert result["function_configs"] == {} assert result["command_plans"] == {} def test_pending_draft_reference_is_bound_to_conversation_name_and_vehicle( vehicle_service: SeatDemoService, ) -> None: built = _build_nap_mode(vehicle_service, "conv-nap-reference-owner") reference = _draft_reference(built) wrong_conversation = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到25度", provider="local", conversation_id="conv-nap-reference-attacker", vehicle_id="denza_n7", mode_profile_draft_context=reference, ) assert wrong_conversation["status"] == "needs_confirmation" assert wrong_conversation.get("mode_profile_draft") is None assert any( issue.get("code") == "MODE_PROFILE_DRAFT_UPDATE_REJECTED" for issue in wrong_conversation["validation"] ) for invalid_reference in ( {**reference, "vehicle_id": "another_vehicle"}, {**reference, "source_request_id": "req-forged-preview"}, ): rejected = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到25度", provider="local", conversation_id="conv-nap-reference-owner", vehicle_id="denza_n7", mode_profile_draft_context=invalid_reference, ) assert rejected["status"] == "needs_confirmation" assert rejected.get("mode_profile_draft") is None assert rejected["hardware_execution_allowed"] is False @pytest.mark.parametrize( "message", ( "把小憩模式的空调介绍一下", "小憩模式的空调有什么作用", "小憩模式的空调和普通空调有什么区别", "小憩模式的空调设置有哪些？", "小憩模式的空调该调几度", "把小憩模式的空调调到25度合适吗", "调整小憩模式的空调参数，列出来", ), ) def test_air_conditioner_name_alone_is_not_parsed_as_mode_profile_update( message: str, ) -> None: assert SeatDemoService._parse_mode_profile_update(message) is None @pytest.mark.parametrize( "message", ( "空调出风模式设置为吹面和吹脚", "把空调出风模式设置为吹面和吹脚", "调整空调出风模式为吹面和吹脚", "主驾座椅按摩模式设置为放松", "副驾座椅零重力模式设为开启", "进入动态摇篮模式，空调24度，打开遮阳帘", "开启女王副驾模式，天幕透光率调到20%", ), ) def test_device_property_mode_is_not_a_named_mode_profile_update( message: str, ) -> None: assert SeatDemoService._parse_mode_profile_update(message) is None @pytest.mark.parametrize( "message", ( "空调出风模式设置为吹面和吹脚，主驾空调温度设置为24度", "空调出风模式设置为吹面和吹脚；副驾空调温度设置为24度", "空调出风模式设置为吹面和吹脚，空调设置为自动控制", "空调出风模式设置为吹面和吹脚，主驾座椅按摩模式设置为放松", "主驾座椅按摩模式设置为放松，按摩强度设置为中等", "把空调出风模式的出风方向设置为吹面和吹脚", "调整主驾座椅按摩模式的强度为中等", "副驾座椅零重力模式的状态设为开启", "空调出风模式有哪些选项？", "主驾座椅按摩模式是什么？", ), ) def test_device_property_mode_commands_and_questions_are_not_saved_mode_queries( message: str, ) -> None: assert parse_mode_profile_detail_query(message) is None assert not is_mode_profile_query(message) assert parse_mode_profile_update(message) is None @pytest.mark.parametrize( "message", ( "空调出风模式设置为吹面和吹脚，副驾空调温度设置为24度", "主驾座椅按摩模式设置为放松，氛围灯亮度设置为3档", ), ) def test_device_property_mode_combinations_continue_to_execute_as_vehicle_control( vehicle_service: SeatDemoService, message: str, ) -> None: result = vehicle_service.process( message, provider="local", conversation_id="conv-device-property-mode-control", execution_mode="dry_run", vehicle_id="denza_n7", ) assert result["status"] == "succeeded" assert result.get("orchestration_route") != "mode_profile_query" assert not SeatDemoService.is_mode_profile_configuration_request(message) def test_quoted_device_property_name_can_explicitly_address_a_saved_mode() -> None: assert parse_mode_profile_detail_query("查看“空调出风模式”的配置") == ( "空调出风模式" ) assert parse_mode_profile_update( "把“空调出风模式”的温度调整到25度" ) == ("空调出风模式", "温度调整到25度") @pytest.mark.parametrize( "message", ( "办公模式的设备？", "办公模式配置？", "办公模式下的动作?", ), ) def test_terminal_question_mark_keeps_short_saved_mode_detail_queries_read_only( message: str, ) -> None: assert parse_mode_profile_detail_query(message) == "办公模式" assert is_mode_profile_query(message) @pytest.mark.parametrize( "message", ( "办公模式的设备。", "办公模式配置。", "空调出风模式设置为吹面和吹脚，温度设置为24度", ), ) def test_plain_mode_nouns_without_a_read_cue_are_not_detail_queries( message: str, ) -> None: assert parse_mode_profile_detail_query(message) is None @pytest.mark.parametrize( ("message", "expected"), ( ("把小憩模式的空调温度调整到25度", ("小憩模式", "空调温度调整到25度")), ("午休模式里的遮阳帘关闭", ("午休模式", "遮阳帘关闭")), ("修改露营模式，天幕透光率改到50%", ("露营模式", "天幕透光率改到50%")), ), ) def test_named_mode_profile_update_requires_an_explicit_member_boundary( message: str, expected: tuple[str, str], ) -> None: assert SeatDemoService._parse_mode_profile_update(message) == expected def test_name_first_comma_update_requires_the_active_draft_name() -> None: message = "小憩模式，空调温度调到25度。" assert SeatDemoService._parse_mode_profile_update(message) is None assert parse_mode_profile_update( message, expected_mode_name="小憩模式", ) == ("小憩模式", "空调温度调到25度") assert parse_mode_profile_update( message, expected_mode_name="露营模式", ) is None def test_name_first_comma_question_is_not_a_draft_update() -> None: message = "小憩模式，空调温度调到25度合适吗？" assert parse_mode_profile_update( message, expected_mode_name="小憩模式", ) is None assert is_mode_profile_query( message, expected_mode_name="小憩模式", ) @pytest.mark.parametrize( "message", ( "修改小憩模式的空调设置是什么", "小憩模式的空调该调几度", "把小憩模式的空调调到25度合适吗", "调整小憩模式的空调参数，列出来", "小憩模式的空调调到25度会怎样", ), ) def test_mode_profile_query_is_fail_closed_even_on_direct_hardware_service_call( vehicle_service: SeatDemoService, message: str, ) -> None: with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("mode-profile query reached hardware lease"), ): result = vehicle_service.preview_native_can( message, provider="local", conversation_id="conv-mode-profile-query-guard", execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "needs_confirmation" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result["orchestration_route"] == "mode_profile_query" assert result["function_configs"] == {} assert result["command_plans"] == {} def test_pending_draft_rejects_unsafe_change_without_returning_a_partial_card( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-nap-pending-invalid-change" built = _build_nap_mode(vehicle_service, conversation_id) rejected = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到99度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) assert rejected["status"] == "needs_confirmation" assert rejected["execution_mode"] == "dry_run" assert rejected["hardware_execution_allowed"] is False assert rejected.get("mode_profile_draft") is None assert rejected["function_configs"] == {} assert rejected["command_plans"] == {} def test_pending_draft_revision_fails_closed_if_full_plan_loses_any_action( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-nap-pending-incomplete-revision" built = _build_nap_mode(vehicle_service, conversation_id) with patch.object( vehicle_service, "_scene_action_signatures", return_value={}, ): rejected = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到25度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) assert rejected["status"] != "succeeded" assert rejected.get("mode_profile_draft") is None assert not _intent_actions(rejected) assert rejected["function_configs"] == {} assert rejected["command_plans"] == {} assert any( issue.get("code") == "MODE_PROFILE_SCENE_PLAN_INCOMPLETE" for issue in rejected["validation"] ) def test_pending_draft_update_rejects_partially_understood_change_atomically( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-nap-pending-partial-change" built = _build_nap_mode(vehicle_service, conversation_id) rejected = vehicle_service.preview_native_can( "把小憩模式的空调温度调整到25度，同时启动曲速引擎", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) assert rejected["status"] == "needs_confirmation" assert rejected.get("mode_profile_draft") is None assert rejected["function_configs"] == {} assert rejected["command_plans"] == {} assert "未完整识别" in rejected["mode_creation_failure"]["reason"] def test_revised_unknown_mode_draft_can_be_trialed_and_saved_with_all_actions( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-custom-pending-save-after-revision" built = vehicle_service.preview_native_can( "构建一个午休模式", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", assistant_plan=LUNCH_BREAK_SCENE_PLAN, ) revised = vehicle_service.preview_native_can( "午休模式的空调温度改为25度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) draft = revised["mode_profile_draft"] trial = vehicle_service.trial_mode_profile_draft( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) saved = vehicle_service.create_mode_profile_card( str(draft["name"]), str(draft["instruction"]), str(draft["source_request_id"]), str(draft["vehicle_id"]), ) assert trial["status"] == "succeeded" assert trial["mode_profile_draft"]["draft_revision"] == 2 assert len(_intent_actions(trial)) == draft["operation_count"] == 5 assert saved["mode_profile"]["operation_count"] == 5 profile = vehicle_service.mode_profiles.get_profile( str(saved["mode_profile"]["id"]), "denza_n7", ) assert next( operation.parameters["temperature_c"] for operation in profile.operations if operation.capability == "climate.set_temperature" ) == 25 def test_explicit_mode_with_close_motions_can_still_receive_a_temperature_revision( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-explicit-motion-draft-revision" built = vehicle_service.preview_native_can( "创建一个躺平测试模式，副驾座椅靠背向后移动到最大位置，" "遮阳帘关闭，车窗关闭，灯光全部关闭，空调调整到24度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", ) assert built["status"] == "succeeded" assert built["mode_profile_draft"]["operation_count"] == 7 revised = vehicle_service.preview_native_can( "把躺平测试模式的空调温度调整到25度", provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) actions = _intent_actions(revised) assert revised["status"] == "succeeded" assert revised["execution_mode"] == "dry_run" assert revised["hardware_execution_allowed"] is False assert revised["mode_profile_draft"]["operation_count"] == len(actions) == 7 assert next( action["parameters"]["temperature_c"] for action in actions if action["capability"] == "climate.set_temperature" ) == 25 capabilities = {str(action["capability"]) for action in actions} assert "roof.set_shade_position" in capabilities assert "window.set_position" in capabilities assert "roof.set_shade_motion" not in capabilities assert "window.set_motion" not in capabilities def test_unscoped_change_rejects_a_draft_with_multiple_targets_for_same_capability( vehicle_service: SeatDemoService, ) -> None: conversation_id = "conv-multi-zone-pending-draft" multi_zone_plan = { **LUNCH_BREAK_SCENE_PLAN, "detected_domains": ["climate"], "requests": [ _scene_request( "climate", "climate.set_temperature", "driver", {"temperature_c": 23} ), _scene_request( "climate", "climate.set_temperature", "passenger", {"temperature_c": 24} ), ], } built = vehicle_service.preview_native_can( "构建一个双区舒适模式", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", assistant_plan=multi_zone_plan, ) assert built["status"] == "succeeded" rejected = vehicle_service.preview_native_can( "把双区舒适模式的空调温度调整到25度", provider="local", conversation_id=conversation_id, vehicle_id="denza_n7", mode_profile_draft_context=_draft_reference(built), ) assert rejected["status"] == "needs_confirmation" assert rejected.get("mode_profile_draft") is None assert "多个目标" in rejected["mode_creation_failure"]["reason"] 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_name_only_scene_label_does_not_leak_device_words_into_local_parser( vehicle_service: SeatDemoService, ) -> None: external_cycle_plan = { "detected_domains": ["climate"], "requests": [ _scene_request( "climate", "climate.set_cycle_mode", "vehicle", {"mode": "external"}, ) ], "ambiguities": [], "decision_summary": ["使用外循环通风。"], "confidence": 0.96, } with patch.object( vehicle_service.hardware_execution, "reserve", side_effect=AssertionError("name-only scene reached hardware lease"), ): result = vehicle_service.preview_native_can( "构建一个外循环通风模式", provider="local", conversation_id="conv-name-only-label-is-not-an-action", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=external_cycle_plan, ) actions = _intent_actions(result) 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 result["mode_profile_draft"]["name"] == "外循环通风模式" assert result["mode_profile_draft"]["operation_count"] == 1 assert result["mode_profile_draft"]["generation_source"] == ( "codex_scene_composition" ) assert not result["intent_ir"]["spec"]["ambiguities"] assert [ ( action["domain"], action["capability"], action["target"], action["parameters"], ) for action in actions ] == [ ( "climate", "climate.set_cycle_mode", {"device": "climate", "zone": "vehicle"}, {"mode": "external"}, ) ] 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"))