Miayui
bfff1b515c
Expose safe NFC-bound infrared actions to semantic scene composition, require flashing-light actions for romantic and Mid-Autumn modes, and preserve mixed CAN/IR mode flows. Fix NFC device reassignment and retry extension/toy MQTT publication when an active card's binding catalog changes.
260 lines
9.0 KiB
Python
260 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.assistant_capabilities import (
|
|
AssistantCapabilityCatalog,
|
|
)
|
|
from cognitive_agent.application.codex_assistant import (
|
|
CodexAssistantService,
|
|
_catalog_exact_scene_template_plan,
|
|
)
|
|
from cognitive_agent.application.mode_profile import ModeProfileRegistry
|
|
from cognitive_agent.application.scene_composer import normalize_scene_output
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
from cognitive_agent.application.self_extension import (
|
|
ExtensionIntegrationError,
|
|
SelfExtensionService,
|
|
SimulationGateway,
|
|
)
|
|
from cognitive_agent.server import SeatDemoHandler, VehicleDisplaySync
|
|
|
|
|
|
def _learned_light() -> dict[str, object]:
|
|
return {
|
|
"id": "ir-slot:romance-light:7",
|
|
"status": "learned",
|
|
"enabled": True,
|
|
"device_id": "romance-light",
|
|
"device_name": "氛围灯带",
|
|
"object_names": ["氛围灯带", "灯带"],
|
|
"device_type": "light",
|
|
"card_uid": "2CC8B103",
|
|
"action": "设置为红色",
|
|
"action_names": ["设置为红色", "红色"],
|
|
"risk": "low",
|
|
"slot": 7,
|
|
}
|
|
|
|
|
|
def test_scene_composer_mixed_can_ir_preview_save_and_revalidate(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service.artifact_root = tmp_path / "runs"
|
|
service.mode_profiles = ModeProfileRegistry(
|
|
REPOSITORY_ROOT / "config" / "modes",
|
|
user_root=tmp_path / "modes",
|
|
)
|
|
extension = SelfExtensionService(
|
|
SimulationGateway(),
|
|
catalog_db_path=tmp_path / "infrared.sqlite3",
|
|
)
|
|
extension.sync_peripheral_catalog([_learned_light()])
|
|
catalog = AssistantCapabilityCatalog(
|
|
service,
|
|
peripheral_provider=extension.list_peripheral_routing_capabilities,
|
|
)
|
|
context = catalog.scene_composition_context(catalog.snapshot())
|
|
capability_ref = context["peripheral_actions"][0]["ref"]
|
|
encoded_context = json.dumps(context, ensure_ascii=False)
|
|
assert "2CC8B103" not in encoded_context
|
|
assert '"slot":7' not in encoded_context
|
|
|
|
plan = normalize_scene_output(
|
|
{
|
|
"requests": [
|
|
{
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": "vehicle",
|
|
"parameters": {"temperature_c": 24},
|
|
},
|
|
{"capability_ref": capability_ref},
|
|
],
|
|
"ambiguities": [],
|
|
"confidence": 0.94,
|
|
},
|
|
context,
|
|
)
|
|
handler = object.__new__(SeatDemoHandler)
|
|
handler.service = service
|
|
handler.extension = extension
|
|
handler.nfc = SimpleNamespace(snapshot=lambda: {"card": None})
|
|
handler.vehicle_display_sync = VehicleDisplaySync()
|
|
message = "创建一个星夜外设模式"
|
|
arguments: dict[str, object] = {
|
|
"utterance": message,
|
|
"provider": "local",
|
|
"conversation_id": "conv-scene-peripheral-flow",
|
|
# The scene path must override this and never emit while creating.
|
|
"execution_mode": "hardware",
|
|
"vehicle_id": "denza_n7",
|
|
"simulation_fault": "normal",
|
|
"hardware_access_bypass": True,
|
|
"reasoning_effort": "low",
|
|
"control_transport": "wired_can",
|
|
"assistant_plan": plan,
|
|
}
|
|
|
|
with patch.object(
|
|
extension,
|
|
"execute_gateway",
|
|
side_effect=AssertionError("mode creation must not emit infrared"),
|
|
):
|
|
preview = handler._preview_mode_with_scene_peripherals(
|
|
"/api/requests/preview",
|
|
arguments,
|
|
plan,
|
|
on_commands_validated=None,
|
|
)
|
|
|
|
assert preview["status"] == "succeeded"
|
|
assert preview["execution_mode"] == "dry_run"
|
|
assert preview["hardware_execution_allowed"] is False
|
|
assert preview["mode_profile_draft"]["domains"] == [
|
|
"climate",
|
|
"self_extension",
|
|
]
|
|
assert preview["mode_profile_draft"]["operation_count"] == 2
|
|
assert set(preview["function_configs"]) == {"native_climate", "self_extension"}
|
|
infrared_command = preview["command_plans"]["self_extension"]["spec"][
|
|
"commands"
|
|
][0]
|
|
assert infrared_command["message"] == "infrared.emit"
|
|
assert infrared_command["signals"] == {"slot": 7}
|
|
assert preview["infrared_actions"][0]["parse_source"] == (
|
|
"scene_composer_capability_ref"
|
|
)
|
|
assert "2CC8B103" not in json.dumps(
|
|
CodexAssistantService._safe_vehicle_result(preview),
|
|
ensure_ascii=False,
|
|
)
|
|
assert "2CC8B103" not in json.dumps(
|
|
CodexAssistantService._validated_control_payload(preview),
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
draft = preview["mode_profile_draft"]
|
|
saved = service.create_mode_profile_card(
|
|
draft["name"],
|
|
draft["instruction"],
|
|
preview["request_id"],
|
|
"denza_n7",
|
|
)
|
|
profile = service.get_mode_profile_card(saved["mode_profile"]["id"], "denza_n7")
|
|
assert [operation.capability for operation in profile.operations] == [
|
|
"climate.set_temperature",
|
|
"infrared.emit",
|
|
]
|
|
entries = handler._saved_mode_infrared_entries(profile)
|
|
assert [entry["capability"]["slot"] for entry in entries] == [7]
|
|
native_execution = service.execute_mode_profile_card(
|
|
profile.id,
|
|
conversation_id="conv-saved-scene-peripheral",
|
|
execution_mode="dry_run",
|
|
vehicle_id="denza_n7",
|
|
)
|
|
infrared_execution = handler._execute_local_peripheral_actions(
|
|
profile.name,
|
|
entries,
|
|
provider="saved_mode_command_flow",
|
|
conversation_id="conv-saved-scene-peripheral",
|
|
execution_mode="dry_run",
|
|
)
|
|
executed = handler._merge_can_and_peripheral_results(
|
|
profile.name,
|
|
native_execution,
|
|
infrared_execution,
|
|
entries,
|
|
)
|
|
assert executed["status"] == "succeeded"
|
|
assert [action["slot"] for action in executed["infrared_actions"]] == [7]
|
|
invoked = handler._execute_request_with_peripherals(
|
|
"/api/requests/preview",
|
|
{
|
|
"utterance": f"进入{profile.name}",
|
|
"provider": "local",
|
|
"conversation_id": "conv-invoke-saved-peripheral-mode",
|
|
"execution_mode": "hardware",
|
|
"vehicle_id": "denza_n7",
|
|
"simulation_fault": "normal",
|
|
"hardware_access_bypass": True,
|
|
"control_transport": "wired_can",
|
|
},
|
|
)
|
|
assert invoked is not None
|
|
assert invoked["status"] == "succeeded"
|
|
assert invoked["execution_mode"] == "dry_run"
|
|
assert [action["slot"] for action in invoked["infrared_actions"]] == [7]
|
|
|
|
template_plan = _catalog_exact_scene_template_plan(
|
|
catalog.snapshot(),
|
|
"浪漫模式",
|
|
)
|
|
assert template_plan is not None
|
|
template_external = normalize_scene_output(
|
|
{
|
|
"requests": [{"capability_ref": capability_ref}],
|
|
"ambiguities": [],
|
|
"confidence": 0.95,
|
|
},
|
|
context,
|
|
)
|
|
augmented_template = {
|
|
**template_plan,
|
|
"requests": [
|
|
*template_plan["requests"],
|
|
*template_external["requests"],
|
|
],
|
|
}
|
|
template_preview = handler._preview_mode_with_scene_peripherals(
|
|
"/api/requests/preview",
|
|
{
|
|
**arguments,
|
|
"utterance": "创建一个浪漫模式",
|
|
"conversation_id": "conv-romantic-template-peripheral",
|
|
"assistant_plan": augmented_template,
|
|
},
|
|
augmented_template,
|
|
on_commands_validated=None,
|
|
)
|
|
assert template_preview["status"] == "succeeded"
|
|
assert template_preview["mode_profile_draft"]["source_template_id"] == "romantic"
|
|
assert template_preview["mode_profile_draft"]["operation_count"] == (
|
|
len(template_preview["intent_ir"]["spec"]["actions"])
|
|
)
|
|
assert template_preview["mode_profile_draft"]["domains"][-1] == "self_extension"
|
|
|
|
explicit = handler._execute_request_with_peripherals(
|
|
"/api/requests/preview",
|
|
{
|
|
"utterance": (
|
|
"创建一个外设办公模式，空调温度设置为25度，"
|
|
"氛围灯带设置为红色"
|
|
),
|
|
"provider": "local",
|
|
"conversation_id": "conv-explicit-peripheral-mode",
|
|
"execution_mode": "hardware",
|
|
"vehicle_id": "denza_n7",
|
|
"simulation_fault": "normal",
|
|
"hardware_access_bypass": True,
|
|
"control_transport": "wired_can",
|
|
},
|
|
)
|
|
assert explicit is not None
|
|
assert explicit["status"] == "succeeded"
|
|
assert explicit["execution_mode"] == "dry_run"
|
|
assert explicit["mode_profile_draft"]["operation_count"] == 2
|
|
assert explicit["infrared_actions"][0]["slot"] == 7
|
|
|
|
# Removing the durable NFC whitelist invalidates both cached refs and saved
|
|
# mode execution before any transport can be reached.
|
|
extension.sync_peripheral_catalog([])
|
|
with pytest.raises(ExtensionIntegrationError, match="未授权"):
|
|
handler._saved_mode_infrared_entries(profile)
|