from __future__ import annotations from copy import deepcopy from dataclasses import dataclass from pathlib import Path from typing import Any from unittest.mock import patch import pytest from cognitive_agent.application.assistant_capabilities import AssistantCapabilityCatalog from cognitive_agent.application.assistant_plan_guard import MAX_PLAN_REQUESTS from cognitive_agent.application.codex_assistant import _authorize_vehicle_request from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService @dataclass(frozen=True) class SceneCase: case_id: str mode_name: str capability: str target: str parameters: dict[str, object] @property def domain(self) -> str: return self.capability.split(".", 1)[0] @dataclass(frozen=True) class DraftChangeCase: case_id: str message: str capability: str target: str parameters: dict[str, object] def _scene_request( capability: str, target: str, parameters: dict[str, object], *, operation: str = "set", duration_ms: int | None = None, ) -> dict[str, object]: return { "domain": capability.split(".", 1)[0], "capability": capability, "target_scope": target, "operation": operation, "duration_ms": duration_ms, "parameters": [ {"name": name, "value": value} for name, value in parameters.items() ], "reason": "按用户场景要求设置当前工程已登记的绝对目标。", } def _scene_plan(*requests: dict[str, object]) -> dict[str, object]: return { "detected_domains": list( dict.fromkeys(str(request["domain"]) for request in requests) ) or ["seat"], "requests": [deepcopy(request) for request in requests], "ambiguities": [], "decision_summary": ["只组合当前工程注册表允许的稳定目标。"], "confidence": 0.95, } @pytest.fixture() def mode_service(tmp_path: Path) -> tuple[SeatDemoService, Path]: service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = tmp_path / "artifacts" user_root = tmp_path / "user-modes" service.mode_profiles = ModeProfileRegistry( REPOSITORY_ROOT / "config" / "modes", user_root=user_root, ) return service, user_root def _preview_without_hardware( service: SeatDemoService, message: str, *, conversation_id: str, assistant_plan: dict[str, object] | None = None, draft_context: dict[str, str] | None = None, ) -> dict[str, Any]: with ( patch.object( service.hardware_execution, "reserve", side_effect=AssertionError("mode draft route reached a hardware lease"), ), patch.object( service.openai, "interpret", side_effect=AssertionError("the deterministic matrix called an external model"), ), ): return service.preview_native_can( message, provider="local", conversation_id=conversation_id, execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=deepcopy(assistant_plan), mode_profile_draft_context=deepcopy(draft_context), ) def _intent_actions(result: dict[str, Any]) -> list[dict[str, Any]]: intent_ir = result.get("intent_ir") if not isinstance(intent_ir, dict): return [] spec = intent_ir.get("spec") if not isinstance(spec, dict): return [] actions = spec.get("actions") if not isinstance(actions, list): return [] return [action for action in actions if isinstance(action, dict)] def _target(action: dict[str, Any]) -> str: raw = action.get("target") if isinstance(raw, dict): return str( raw.get("instance") or raw.get("zone") or raw.get("scope") or "" ) return str(raw or "") def _action_map( result: dict[str, Any], ) -> dict[tuple[str, str], dict[str, object]]: return { (str(action.get("capability") or ""), _target(action)): dict( action.get("parameters") or {} ) for action in _intent_actions(result) } def _assert_parameter_subset( result: dict[str, Any], capability: str, target: str, parameters: dict[str, object], ) -> None: action = _action_map(result).get((capability, target)) assert action is not None, (capability, target, _action_map(result)) for name, value in parameters.items(): assert action.get(name) == value def _draft_reference(result: dict[str, Any]) -> dict[str, str]: draft = result.get("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 _assert_successful_preview( result: dict[str, Any], *, name: str, operation_count: int, ) -> dict[str, Any]: 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.get("mode_profile") is None assert not any( isinstance(issue, dict) and issue.get("severity") == "error" for issue in result.get("validation", []) ) draft = result.get("mode_profile_draft") assert isinstance(draft, dict) assert draft["name"] == name assert draft["operation_count"] == operation_count assert len(_intent_actions(result)) == operation_count return draft def _assert_atomic_failure(result: dict[str, Any], user_root: Path) -> None: 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 _intent_actions(result) == [] assert result.get("function_config") is None assert result.get("function_configs") == {} assert result.get("command_plan") is None assert result.get("command_plans") == {} assert result.get("scene_config") is None assert result.get("execution_report") is None assert not list(user_root.rglob("*.yaml")) TYPED_SCENE_CASES = ( SceneCase( "integer-temperature", "整数恒温模式", "climate.set_temperature", "vehicle", {"temperature_c": 25}, ), SceneCase( "percentage-shade", "柔光遮阳模式", "roof.set_shade_position", "vehicle", {"position_pct": 37}, ), SceneCase( "binary-state", "迎宾氛围模式", "ambient_light.set_power", "vehicle", {"state": "on"}, ), SceneCase( "color-and-percentage", "蓝光阅读模式", "ambient_light.set_illumination", "vehicle", {"color_hex": "#2040FF", "brightness_pct": 60}, ), SceneCase( "fan-level", "清风通勤模式", "climate.set_fan_level", "vehicle", {"level": 4}, ), SceneCase( "two-axis-percentage", "定向送风模式", "outlet.set_position_percent", "passenger_right", {"horizontal_pct": 20, "vertical_pct": 80}, ), SceneCase( "scoped-window", "后排换气模式", "window.set_position", "rear_right", {"position_pct": 45}, ), SceneCase( "state-and-level", "副驾暖座模式", "seat.set_heating", "passenger", {"state": "on", "level": 3}, ), SceneCase( "cycle-enum", "外循环通风模式", "climate.set_cycle_mode", "vehicle", {"mode": "external"}, ), SceneCase( "control-enum", "自动温控模式", "climate.set_control_mode", "vehicle", {"mode": "auto"}, ), SceneCase( "all-reading-lights", "全舱暗光模式", "interior_light.set_reading_power", "all", {"state": "off"}, ), SceneCase( "seat-angle", "副驾舒展模式", "seat.set_backrest_recline", "passenger", {"angle_deg": 120}, ), ) @pytest.mark.parametrize("case", TYPED_SCENE_CASES, ids=lambda case: case.case_id) def test_name_only_scene_parameter_types_are_registry_bounded_and_compile( mode_service: tuple[SeatDemoService, Path], case: SceneCase, ) -> None: service, user_root = mode_service catalog = AssistantCapabilityCatalog(service).snapshot() registered = { str(item["id"]): item for item in catalog["capabilities"] if isinstance(item, dict) } assert case.capability in registered assert registered[case.capability]["assistant_callable"] is True result = _preview_without_hardware( service, f"构建一个{case.mode_name}", conversation_id=f"conv-typed-scene-{case.case_id}", assistant_plan=_scene_plan( _scene_request(case.capability, case.target, case.parameters) ), ) draft = _assert_successful_preview(result, name=case.mode_name, operation_count=1) assert draft["generation_source"] == "codex_scene_composition" _assert_parameter_subset( result, case.capability, case.target, case.parameters, ) assert not list(user_root.rglob("*.yaml")) BOUNDARY_SCENE_CASES = ( SceneCase("temperature-min", "温度下界模式", "climate.set_temperature", "vehicle", {"temperature_c": 18}), SceneCase("temperature-max", "温度上界模式", "climate.set_temperature", "vehicle", {"temperature_c": 32}), SceneCase("fan-min", "微风下界模式", "climate.set_fan_level", "vehicle", {"level": 1}), SceneCase("fan-max", "强风上界模式", "climate.set_fan_level", "vehicle", {"level": 7}), SceneCase("shade-min", "遮阳下界模式", "roof.set_shade_position", "vehicle", {"position_pct": 0}), SceneCase("shade-max", "遮阳上界模式", "roof.set_shade_position", "vehicle", {"position_pct": 100}), SceneCase("window-min", "车窗下界模式", "window.set_position", "front_left", {"position_pct": 0}), SceneCase("window-max", "车窗上界模式", "window.set_position", "front_left", {"position_pct": 100}), SceneCase("brightness-min", "暗光下界模式", "ambient_light.set_brightness_level", "vehicle", {"brightness_level": 1}), SceneCase("brightness-max", "亮光上界模式", "ambient_light.set_brightness_level", "vehicle", {"brightness_level": 5}), SceneCase("color-index-min", "色号下界模式", "ambient_light.set_color_index", "vehicle", {"color_index": 1}), SceneCase("color-index-max", "色号上界模式", "ambient_light.set_color_index", "vehicle", {"color_index": 127}), SceneCase("outlet-min", "风口下界模式", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 0, "vertical_pct": 0}), SceneCase("outlet-max", "风口上界模式", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 100, "vertical_pct": 100}), SceneCase("heating-min", "暖座下界模式", "seat.set_heating", "passenger", {"state": "on", "level": 1}), # The passenger CAN0 reusable chain is physically bounded to levels 1-3. # Levels 4-5 remain declared by the broad domain catalog but currently do # not compile into a reusable Denza N7 mode. SceneCase("heating-max", "暖座上界模式", "seat.set_heating", "passenger", {"state": "on", "level": 3}), ) @pytest.mark.parametrize("case", BOUNDARY_SCENE_CASES, ids=lambda case: case.case_id) def test_scene_parameter_inclusive_boundaries_compile_without_hardware( mode_service: tuple[SeatDemoService, Path], case: SceneCase, ) -> None: service, user_root = mode_service result = _preview_without_hardware( service, f"构建一个{case.mode_name}", conversation_id=f"conv-boundary-{case.case_id}", assistant_plan=_scene_plan( _scene_request(case.capability, case.target, case.parameters) ), ) _assert_successful_preview(result, name=case.mode_name, operation_count=1) _assert_parameter_subset(result, case.capability, case.target, case.parameters) assert not list(user_root.rglob("*.yaml")) BASE_MODE_NAME = "参数基线模式" BASE_REQUESTS = ( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 23}), _scene_request("climate.set_fan_level", "vehicle", {"level": 2}), _scene_request("climate.set_cycle_mode", "vehicle", {"mode": "internal"}), _scene_request("roof.set_shade_position", "vehicle", {"position_pct": 30}), _scene_request( "ambient_light.set_illumination", "vehicle", {"color_hex": "#0000FF", "brightness_pct": 50}, ), _scene_request("window.set_position", "front_right", {"position_pct": 25}), _scene_request("seat.set_heating", "passenger", {"state": "on", "level": 2}), _scene_request( "outlet.set_position_percent", "driver_left", {"horizontal_pct": 25, "vertical_pct": 75}, ), ) def _build_parameter_base( service: SeatDemoService, *, conversation_id: str, ) -> dict[str, Any]: result = _preview_without_hardware( service, f"构建一个{BASE_MODE_NAME}", conversation_id=conversation_id, assistant_plan=_scene_plan(*BASE_REQUESTS), ) _assert_successful_preview( result, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS), ) return result @pytest.mark.parametrize( ("message", "expected_name"), ( ("麻烦帮我创建一个晨间一号模式", "晨间一号模式"), ("给我新建一套午后二号模式", "午后二号模式"), ("请帮忙新增夜间三号模式", "夜间三号模式"), ("生成一个周末四号模式", "周末四号模式"), ), ) def test_name_only_creation_verbs_counters_and_politeness_share_one_route( mode_service: tuple[SeatDemoService, Path], message: str, expected_name: str, ) -> None: service, user_root = mode_service result = _preview_without_hardware( service, message, conversation_id="conv-create-language-" + str(abs(hash(message)))[:12], assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 24}) ), ) draft = _assert_successful_preview(result, name=expected_name, operation_count=1) assert draft["generation_source"] == "codex_scene_composition" assert not list(user_root.rglob("*.yaml")) DRAFT_CHANGE_CASES = ( DraftChangeCase( "integer-temperature", "把参数基线模式的空调温度改为26度", "climate.set_temperature", "vehicle", {"temperature_c": 26}, ), DraftChangeCase( "percentage-shade", "请帮我把参数基线模式里的遮阳帘开到65%", "roof.set_shade_position", "vehicle", {"position_pct": 65}, ), DraftChangeCase( "color-pair", "将参数基线模式中的氛围灯改为红色且亮度40%", "ambient_light.set_illumination", "vehicle", {"color_hex": "#FF0000", "brightness_pct": 40}, ), DraftChangeCase( "fan-level", "参数基线模式，空调风量调到5档", "climate.set_fan_level", "vehicle", {"level": 5}, ), DraftChangeCase( "scoped-window", "把参数基线模式中的右前车窗开到60%", "window.set_position", "front_right", {"position_pct": 60}, ), DraftChangeCase( "seat-state-level", "把参数基线模式里的副驾座椅加热调到3档", "seat.set_heating", "passenger", {"state": "on", "level": 3}, ), DraftChangeCase( "cycle-enum", "修改参数基线模式，空调切换为外循环", "climate.set_cycle_mode", "vehicle", {"mode": "external"}, ), DraftChangeCase( "outlet-two-axis", "把参数基线模式里的主驾左风口调到水平60%垂直40%", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 60, "vertical_pct": 40}, ), ) @pytest.mark.parametrize("case", DRAFT_CHANGE_CASES, ids=lambda case: case.case_id) def test_pending_draft_can_replace_each_parameter_type_and_preserve_siblings( mode_service: tuple[SeatDemoService, Path], case: DraftChangeCase, ) -> None: service, user_root = mode_service conversation_id = f"conv-draft-change-{case.case_id}" built = _build_parameter_base(service, conversation_id=conversation_id) before = _action_map(built) revised = _preview_without_hardware( service, case.message, conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request(case.capability, case.target, case.parameters) ), draft_context=_draft_reference(built), ) source_draft = built["mode_profile_draft"] draft = _assert_successful_preview( revised, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS), ) assert draft["generation_source"] == "draft_revision" assert draft["draft_revision"] == source_draft["draft_revision"] + 1 assert draft["revises_source_request_id"] == source_draft["source_request_id"] assert draft["requested_operation_count"] == 1 assert draft["changed_operation_count"] == 1 _assert_parameter_subset( revised, case.capability, case.target, case.parameters, ) after = _action_map(revised) changed_key = (case.capability, case.target) assert set(after) == set(before) assert { key: parameters for key, parameters in after.items() if key != changed_key } == { key: parameters for key, parameters in before.items() if key != changed_key } assert not list(user_root.rglob("*.yaml")) def test_pending_draft_can_append_one_new_capability_without_replacing_siblings( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service conversation_id = "conv-draft-append-capability" built = _build_parameter_base(service, conversation_id=conversation_id) before = _action_map(built) revised = _preview_without_hardware( service, "把参数基线模式里的氛围灯关闭", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("ambient_light.set_power", "vehicle", {"state": "off"}) ), draft_context=_draft_reference(built), ) draft = _assert_successful_preview( revised, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS) + 1, ) assert draft["requested_operation_count"] == 1 assert draft["changed_operation_count"] == 1 after = _action_map(revised) assert after[("ambient_light.set_power", "vehicle")] == {"state": "off"} assert {key: after[key] for key in before} == before assert not list(user_root.rglob("*.yaml")) def test_repeating_the_existing_value_is_a_safe_noop_revision( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service conversation_id = "conv-draft-noop-revision" built = _build_parameter_base(service, conversation_id=conversation_id) revised = _preview_without_hardware( service, "把参数基线模式的空调温度调到23度", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 23}) ), draft_context=_draft_reference(built), ) draft = _assert_successful_preview( revised, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS), ) assert draft["draft_revision"] == 2 assert draft["requested_operation_count"] == 1 assert draft["changed_operation_count"] == 0 assert _action_map(revised) == _action_map(built) assert not list(user_root.rglob("*.yaml")) @pytest.mark.parametrize( "message", ( "把参数基线模式的空调温度调整到26度", "请帮我 把 参数基线模式 中的 空调温度设置为26度", "修改参数基线模式，空调温度调至26℃", "参数基线模式，空调温度调成26度。", "麻烦将参数基线模式里的空调温度变更为26度", ), ) def test_draft_update_word_order_synonyms_and_politeness_keep_one_meaning( mode_service: tuple[SeatDemoService, Path], message: str, ) -> None: service, user_root = mode_service conversation_id = "conv-language-" + str(abs(hash(message)))[:12] built = _build_parameter_base(service, conversation_id=conversation_id) revised = _preview_without_hardware( service, message, conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request( "climate.set_temperature", "vehicle", {"temperature_c": 26}, ) ), draft_context=_draft_reference(built), ) _assert_successful_preview( revised, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS), ) assert revised["mode_profile_draft"]["requested_operation_count"] == 1 assert revised["mode_profile_draft"]["changed_operation_count"] == 1 _assert_parameter_subset( revised, "climate.set_temperature", "vehicle", {"temperature_c": 26}, ) assert not list(user_root.rglob("*.yaml")) def test_multi_action_revision_is_atomic_and_preserves_every_unmentioned_sibling( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service conversation_id = "conv-multi-action-revision" built = _build_parameter_base(service, conversation_id=conversation_id) before = _action_map(built) replacements = ( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 27}), _scene_request("climate.set_fan_level", "vehicle", {"level": 6}), _scene_request("roof.set_shade_position", "vehicle", {"position_pct": 10}), ) revised = _preview_without_hardware( service, "把参数基线模式的空调温度调到27度、风量调到6档，遮阳帘开到10%", conversation_id=conversation_id, assistant_plan=_scene_plan(*replacements), draft_context=_draft_reference(built), ) _assert_successful_preview( revised, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS), ) assert revised["mode_profile_draft"]["requested_operation_count"] == 3 assert revised["mode_profile_draft"]["changed_operation_count"] == 3 expected_changes = { ("climate.set_temperature", "vehicle"): {"temperature_c": 27}, ("climate.set_fan_level", "vehicle"): {"level": 6}, ("roof.set_shade_position", "vehicle"): {"position_pct": 10}, } after = _action_map(revised) for key, parameters in expected_changes.items(): for name, value in parameters.items(): assert after[key][name] == value assert { key: parameters for key, parameters in after.items() if key not in expected_changes } == { key: parameters for key, parameters in before.items() if key not in expected_changes } assert not list(user_root.rglob("*.yaml")) def test_continuous_revisions_form_a_lineage_and_keep_prior_changes( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service conversation_id = "conv-continuous-draft-lineage" first = _build_parameter_base(service, conversation_id=conversation_id) second = _preview_without_hardware( service, "把参数基线模式的空调温度调到26度", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 26}) ), draft_context=_draft_reference(first), ) third = _preview_without_hardware( service, "把参数基线模式的遮阳帘开到70%", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("roof.set_shade_position", "vehicle", {"position_pct": 70}) ), draft_context=_draft_reference(second), ) first_draft = first["mode_profile_draft"] second_draft = second["mode_profile_draft"] third_draft = _assert_successful_preview( third, name=BASE_MODE_NAME, operation_count=len(BASE_REQUESTS), ) assert [ first_draft["draft_revision"], second_draft["draft_revision"], third_draft["draft_revision"], ] == [1, 2, 3] assert second_draft["revises_source_request_id"] == first_draft["source_request_id"] assert third_draft["revises_source_request_id"] == second_draft["source_request_id"] _assert_parameter_subset( third, "climate.set_temperature", "vehicle", {"temperature_c": 26}, ) _assert_parameter_subset( third, "roof.set_shade_position", "vehicle", {"position_pct": 70}, ) assert not list(user_root.rglob("*.yaml")) def test_unscoped_revision_rejects_a_multi_target_capability_atomically( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service conversation_id = "conv-multizone-ambiguous-revision" built = _preview_without_hardware( service, "构建一个双区矩阵模式", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "driver", {"temperature_c": 23}), _scene_request("climate.set_temperature", "passenger", {"temperature_c": 24}), _scene_request("climate.set_fan_level", "vehicle", {"level": 2}), ), ) _assert_successful_preview(built, name="双区矩阵模式", operation_count=3) rejected = _preview_without_hardware( service, "把双区矩阵模式的空调温度调到25度", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 25}) ), draft_context=_draft_reference(built), ) _assert_atomic_failure(rejected, user_root) assert "多个目标" in rejected["mode_creation_failure"]["reason"] def test_explicit_target_revision_changes_only_that_zone( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service conversation_id = "conv-multizone-explicit-revision" built = _preview_without_hardware( service, "构建一个双区独立模式", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "driver", {"temperature_c": 23}), _scene_request("climate.set_temperature", "passenger", {"temperature_c": 24}), ), ) revised = _preview_without_hardware( service, "把双区独立模式的副驾空调温度调到26度", conversation_id=conversation_id, assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "passenger", {"temperature_c": 26}) ), draft_context=_draft_reference(built), ) _assert_successful_preview(revised, name="双区独立模式", operation_count=2) _assert_parameter_subset( revised, "climate.set_temperature", "driver", {"temperature_c": 23}, ) _assert_parameter_subset( revised, "climate.set_temperature", "passenger", {"temperature_c": 26}, ) assert not list(user_root.rglob("*.yaml")) @pytest.mark.parametrize( "message", ( "把参数基线模式的空调温度调到26度合适吗？", "参数基线模式里的遮阳帘为什么要关到30%？", "调整参数基线模式的氛围灯设置有哪些？", "把参数基线模式的副驾座椅加热调到4档会怎样？", "参数基线模式的右前车窗开到60%是否安全？", ), ) def test_question_shaped_mode_turns_never_modify_a_draft_or_touch_hardware( mode_service: tuple[SeatDemoService, Path], message: str, ) -> None: service, user_root = mode_service conversation_id = "conv-question-" + str(abs(hash(message)))[:12] built = _build_parameter_base(service, conversation_id=conversation_id) reference = _draft_reference(built) authorization = _authorize_vehicle_request( message, mode_profile_draft_context=reference, ) assert authorization.instruction is None assert authorization.reason == "mode_profile_query" result = _preview_without_hardware( service, message, conversation_id=conversation_id, draft_context=reference, ) assert result["status"] == "needs_confirmation" assert result["orchestration_route"] == "mode_profile_query" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result.get("mode_profile_draft") is None assert _intent_actions(result) == [] assert result["function_configs"] == {} assert result["command_plans"] == {} assert not list(user_root.rglob("*.yaml")) @pytest.mark.parametrize( "message", ( "能否创建一个观影模式？", "构建一个观影模式，空调温度是25度还是26度？", ), ) def test_create_shaped_questions_never_reach_hardware_or_scene_composer( mode_service: tuple[SeatDemoService, Path], message: str, ) -> None: service, user_root = mode_service with patch.object( service.hardware_execution, "reserve", side_effect=AssertionError("mode-create question reached a hardware lease"), ): result = service.preview_native_can( message, provider="local", conversation_id="conv-create-question-" + str(abs(hash(message)))[:12], execution_mode="hardware", vehicle_id="denza_n7", ) assert result["status"] == "needs_confirmation" assert result["orchestration_route"] == "mode_profile_query" assert result["execution_mode"] == "dry_run" assert result["hardware_execution_allowed"] is False assert result.get("mode_profile_draft") is None assert _intent_actions(result) == [] assert not list(user_root.rglob("*.yaml")) INVALID_PARAMETER_CASES = ( SceneCase("temperature-low", "非法低温模式", "climate.set_temperature", "vehicle", {"temperature_c": 17}), SceneCase("temperature-high", "非法高温模式", "climate.set_temperature", "vehicle", {"temperature_c": 33}), SceneCase("fan-low", "非法低风模式", "climate.set_fan_level", "vehicle", {"level": 0}), SceneCase("fan-high", "非法高风模式", "climate.set_fan_level", "vehicle", {"level": 8}), SceneCase("shade-low", "非法遮阳下界模式", "roof.set_shade_position", "vehicle", {"position_pct": -1}), SceneCase("shade-high", "非法遮阳上界模式", "roof.set_shade_position", "vehicle", {"position_pct": 101}), SceneCase("window-target", "非法全窗比例模式", "window.set_position", "all", {"position_pct": 50}), SceneCase("window-high", "非法车窗比例模式", "window.set_position", "front_left", {"position_pct": 101}), SceneCase("color-format", "非法颜色模式", "ambient_light.set_illumination", "vehicle", {"color_hex": "red", "brightness_pct": 50}), SceneCase("color-missing-brightness", "缺少亮度模式", "ambient_light.set_illumination", "vehicle", {"color_hex": "#FF0000"}), SceneCase("brightness-high", "非法亮度模式", "ambient_light.set_brightness_level", "vehicle", {"brightness_level": 6}), SceneCase("outlet-missing-axis", "缺少风口轴模式", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 50}), SceneCase("outlet-high", "非法风口比例模式", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 101, "vertical_pct": 50}), SceneCase("heating-native-gap-4", "暖座四档模式", "seat.set_heating", "passenger", {"state": "on", "level": 4}), SceneCase("heating-native-gap-5", "暖座五档模式", "seat.set_heating", "passenger", {"state": "on", "level": 5}), SceneCase("heating-level", "非法暖座档位模式", "seat.set_heating", "passenger", {"state": "on", "level": 6}), SceneCase("heating-target", "非法暖座目标模式", "seat.set_heating", "driver", {"state": "on", "level": 2}), SceneCase("cycle-enum", "非法循环模式", "climate.set_cycle_mode", "vehicle", {"mode": "mixed"}), ) @pytest.mark.parametrize("case", INVALID_PARAMETER_CASES, ids=lambda case: case.case_id) def test_one_invalid_parameter_discards_valid_siblings_and_never_returns_a_card( mode_service: tuple[SeatDemoService, Path], case: SceneCase, ) -> None: service, user_root = mode_service result = _preview_without_hardware( service, f"构建一个{case.mode_name}", conversation_id=f"conv-invalid-{case.case_id}", assistant_plan=_scene_plan( _scene_request( "ambient_light.set_power", "vehicle", {"state": "off"}, ), _scene_request(case.capability, case.target, case.parameters), ), ) _assert_atomic_failure(result, user_root) INVALID_DRAFT_CHANGES = ( ( "temperature-low", "把参数基线模式的空调温度调到17度", _scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 17}) ), ), ( "temperature-high", "把参数基线模式的空调温度调到33度", _scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 33}) ), ), ( "fan-high", "把参数基线模式的空调风量调到8档", _scene_plan( _scene_request("climate.set_fan_level", "vehicle", {"level": 8}) ), ), ( "shade-high", "把参数基线模式的遮阳帘开到101%", _scene_plan( _scene_request("roof.set_shade_position", "vehicle", {"position_pct": 101}) ), ), ( "window-high", "把参数基线模式的右前车窗开到101%", _scene_plan( _scene_request("window.set_position", "front_right", {"position_pct": 101}) ), ), ( "heating-native-gap", "把参数基线模式的副驾座椅加热调到4档", _scene_plan( _scene_request( "seat.set_heating", "passenger", {"state": "on", "level": 4}, ) ), ), ( "outlet-missing-axis", "把参数基线模式的主驾左风口水平调到60%", _scene_plan( _scene_request( "outlet.set_position_percent", "driver_left", {"horizontal_pct": 60}, ) ), ), ( "multiple-values", "把参数基线模式的空调温度调到24度和26度", None, ), ( "valid-plus-unsupported", "把参数基线模式的空调温度调到26度，同时启动曲速引擎", None, ), ) @pytest.mark.parametrize( ("case_id", "message", "assistant_plan"), INVALID_DRAFT_CHANGES, ids=[item[0] for item in INVALID_DRAFT_CHANGES], ) def test_invalid_draft_change_is_atomic_and_leaves_the_source_draft_usable( mode_service: tuple[SeatDemoService, Path], case_id: str, message: str, assistant_plan: dict[str, object] | None, ) -> None: service, user_root = mode_service conversation_id = f"conv-invalid-draft-{case_id}" built = _build_parameter_base(service, conversation_id=conversation_id) before = _action_map(built) source_draft = built["mode_profile_draft"] rejected = _preview_without_hardware( service, message, conversation_id=conversation_id, assistant_plan=assistant_plan, draft_context=_draft_reference(built), ) _assert_atomic_failure(rejected, user_root) assert any( isinstance(issue, dict) and issue.get("code") == "MODE_PROFILE_DRAFT_UPDATE_REJECTED" for issue in rejected.get("validation", []) ) with patch.object( service.hardware_execution, "reserve", side_effect=AssertionError("source draft trial reached a hardware lease"), ): trial = service.trial_mode_profile_draft( str(source_draft["name"]), str(source_draft["instruction"]), str(source_draft["source_request_id"]), str(source_draft["vehicle_id"]), ) assert trial["status"] == "succeeded" assert trial["execution_mode"] == "dry_run" assert trial["hardware_execution_allowed"] is False assert _action_map(trial) == before assert not list(user_root.rglob("*.yaml")) FORBIDDEN_REUSABLE_ACTIONS = ( ( "rear-door-lock", _scene_request("door.set_rear_lock", "rear", {"state": "lock"}), ), ( "compressor-shortcut", _scene_request("climate.set_ac", "vehicle", {"state": "on"}), ), ( "driver-timed-axis", _scene_request( "seat.set_driver_axis_native", "driver", {"axis": "horizontal", "direction": "forward"}, duration_ms=500, ), ), ( "dynamic-seat", _scene_request("seat.start_dynamic_mode", "passenger", {"mode_id": 1}), ), ( "save-preset-side-effect", _scene_request( "seat.save_posture_preset", "passenger", {"preset_name": "不应保存"}, operation="save", ), ), ( "apply-preset-indirection", _scene_request( "seat.apply_posture_preset", "passenger", {"preset_name": "不存在"}, operation="apply", ), ), ( "calibrated-outlet", _scene_request( "outlet.set_calibrated_position", "driver_left", {"horizontal_value": -20, "vertical_value": 30}, ), ), ( "transient-window-motion", _scene_request("window.set_motion", "front_left", {"motion": "stop"}), ), ( "transient-shade-motion", _scene_request("roof.set_shade_motion", "vehicle", {"motion": "stop"}), ), ) @pytest.mark.parametrize( ("case_id", "forbidden_request"), FORBIDDEN_REUSABLE_ACTIONS, ids=[item[0] for item in FORBIDDEN_REUSABLE_ACTIONS], ) def test_registered_but_non_reusable_or_transient_actions_are_rejected_atomically( mode_service: tuple[SeatDemoService, Path], case_id: str, forbidden_request: dict[str, object], ) -> None: service, user_root = mode_service catalog_ids = { item["id"] for item in AssistantCapabilityCatalog(service).snapshot()["capabilities"] } assert forbidden_request["capability"] in catalog_ids result = _preview_without_hardware( service, f"构建一个禁止动作{case_id}模式", conversation_id=f"conv-forbidden-{case_id}", assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 24}), forbidden_request, ), ) _assert_atomic_failure(result, user_root) def test_duplicate_scene_key_is_rejected_instead_of_last_write_wins( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service result = _preview_without_hardware( service, "构建一个冲突温度模式", conversation_id="conv-duplicate-scene-key", assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 24}), _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 26}), ), ) _assert_atomic_failure(result, user_root) MAXIMUM_SCENE_REQUESTS = ( _scene_request("climate.set_power", "vehicle", {"state": "on"}), _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 24}), _scene_request("climate.set_fan_level", "vehicle", {"level": 2}), _scene_request("climate.set_cycle_mode", "vehicle", {"mode": "internal"}), _scene_request("climate.set_control_mode", "vehicle", {"mode": "auto"}), _scene_request("climate.set_front_defrost", "vehicle", {"state": "off"}), _scene_request("roof.set_shade_position", "vehicle", {"position_pct": 0}), _scene_request("window.set_position", "front_left", {"position_pct": 0}), _scene_request("window.set_position", "front_right", {"position_pct": 0}), _scene_request("ambient_light.set_power", "vehicle", {"state": "off"}), _scene_request("interior_light.set_indoor_power", "vehicle", {"state": "off"}), _scene_request("interior_light.set_reading_power", "all", {"state": "off"}), ) def test_scene_request_limit_accepts_exactly_twelve_verified_actions( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service assert len(MAXIMUM_SCENE_REQUESTS) == MAX_PLAN_REQUESTS == 12 result = _preview_without_hardware( service, "构建一个十二动作模式", conversation_id="conv-twelve-scene-actions", assistant_plan=_scene_plan(*MAXIMUM_SCENE_REQUESTS), ) _assert_successful_preview(result, name="十二动作模式", operation_count=12) assert not list(user_root.rglob("*.yaml")) def test_scene_request_limit_rejects_thirteen_before_any_hardware_or_persistence( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service oversized = ( *MAXIMUM_SCENE_REQUESTS, _scene_request("outlet.set_switch_state", "driver_left", {"state": "close"}), ) assert len(oversized) == MAX_PLAN_REQUESTS + 1 with pytest.raises(ValueError, match="too many requests"): _preview_without_hardware( service, "构建一个十三动作模式", conversation_id="conv-thirteen-scene-actions", assistant_plan=_scene_plan(*oversized), ) assert not list(user_root.rglob("*.yaml")) def test_known_mode_name_reuses_repository_profile_even_if_a_plan_is_supplied( mode_service: tuple[SeatDemoService, Path], ) -> None: service, user_root = mode_service nap = service.mode_profiles.get_profile("nap", "denza_n7") result = _preview_without_hardware( service, "构建一个小憩模式", conversation_id="conv-known-mode-exact-reuse-matrix", assistant_plan=_scene_plan( _scene_request("climate.set_temperature", "vehicle", {"temperature_c": 31}) ), ) draft = _assert_successful_preview( result, name=nap.name, operation_count=len(nap.operations), ) assert draft["generation_source"] == "registered_mode_profile" assert draft["existing_mode_id"] == "nap" expected = { (operation.capability, operation.target_scope): operation.parameters for operation in nap.operations } actual = _action_map(result) for (capability, target), parameters in expected.items(): rendered_target = "passenger" if target == "front_right" and capability.startswith("seat.") else target for name, value in parameters.items(): assert actual[(capability, rendered_target)][name] == value assert not list(user_root.rglob("*.yaml"))