Files
Cognitive_Agent/tests/integration/test_mode_draft_parameter_matrix.py
T

1279 lines
44 KiB
Python

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"}),
_scene_request("window.set_position", "rear_left", {"position_pct": 0}),
_scene_request("window.set_position", "rear_right", {"position_pct": 0}),
_scene_request("ambient_light.set_mode", "vehicle", {"mode": "romantic"}),
_scene_request("ambient_light.set_brightness_level", "vehicle", {"brightness_level": 2}),
_scene_request("ambient_light.set_color_index", "vehicle", {"color_index": 6}),
_scene_request("seat.set_heating", "passenger", {"state": "on", "level": 1}),
_scene_request("seat.set_backrest_recline", "passenger", {"position_pct": 45}),
_scene_request("seat.set_legrest_height", "passenger", {"position_pct": 35}),
)
def test_scene_request_limit_accepts_exactly_twenty_verified_actions(
mode_service: tuple[SeatDemoService, Path],
) -> None:
service, user_root = mode_service
assert len(MAXIMUM_SCENE_REQUESTS) == MAX_PLAN_REQUESTS == 20
result = _preview_without_hardware(
service,
"构建一个二十动作模式",
conversation_id="conv-twenty-scene-actions",
assistant_plan=_scene_plan(*MAXIMUM_SCENE_REQUESTS),
)
_assert_successful_preview(result, name="二十动作模式", operation_count=20)
assert not list(user_root.rglob("*.yaml"))
def test_scene_request_limit_rejects_twenty_one_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-twenty-one-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")
selected_operations = [
operation for operation in nap.operations
if operation.domain != "seat" or operation.target_scope in {"passenger", "front_right"}
]
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(selected_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 selected_operations
}
actual = _action_map(result)
assert {target for capability, target in actual if capability.startswith("seat.")} == {"passenger"}
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"))