Cognitive_Agent/tests/integration/test_scene_peripheral_mode_flow.py

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,
    scene_context_for_mode_creation,
    scene_required_native_fallbacks,
    scene_required_peripheral_refs,
    scene_semantic_guidance,
)
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 _learned_light_power_on() -> dict[str, object]:
    return {
        "id": "ir-slot:romance-light:1",
        "status": "learned",
        "enabled": True,
        "device_id": "romance-light",
        "device_name": "氛围灯带",
        "object_names": ["氛围灯带", "灯带"],
        "device_type": "light",
        "card_uid": "2CC8B103",
        "action": "打开",
        "action_names": ["打开", "开启", "点亮"],
        "risk": "low",
        "slot": 1,
    }


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_power_on(), _learned_light()]
    )
    catalog = AssistantCapabilityCatalog(
        service,
        peripheral_provider=extension.list_peripheral_routing_capabilities,
    )
    context = catalog.scene_composition_context(catalog.snapshot())
    capability_ref = next(
        item["ref"]
        for item in context["peripheral_actions"]
        if "红色" in str(item.get("action_label") or "")
    )
    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"] == 3
    assert set(preview["function_configs"]) == {"native_climate", "self_extension"}
    infrared_commands = preview["command_plans"]["self_extension"]["spec"][
        "commands"
    ]
    assert [command["message"] for command in infrared_commands] == [
        "infrared.emit",
        "infrared.emit",
    ]
    assert [command["signals"] for command in infrared_commands] == [
        {"slot": 1},
        {"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",
        "infrared.emit",
    ]
    entries = handler._saved_mode_infrared_entries(profile)
    assert [entry["capability"]["slot"] for entry in entries] == [1, 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"]] == [1, 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"]] == [1, 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_context = scene_context_for_mode_creation(
        context,
        "创建一个浪漫模式",
    )
    template_semantics = scene_semantic_guidance("浪漫模式")
    template_native_fallbacks = scene_required_native_fallbacks(
        "浪漫模式",
        template_context,
        include_default_bundle=True,
    )
    template_semantics["required_native_domains"] = [
        str(item["domain"])
        for item in template_native_fallbacks
    ]
    template_semantics["native_fallback_requests"] = template_native_fallbacks
    template_semantics["peripheral_policy"] = {
        "required_refs": scene_required_peripheral_refs(
            "浪漫模式",
            template_context,
            include_default_bundle=True,
        ),
    }
    template_context["scene_semantics"] = template_semantics
    augmented_template = CodexAssistantService._with_required_scene_peripherals(
        augmented_template,
        template_context,
    )
    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"]["generation_source"] == (
        "codex_scene_composition"
    )
    template_actions = template_preview["intent_ir"]["spec"]["actions"]
    assert template_preview["mode_profile_draft"]["operation_count"] == (
        len(template_actions)
    )
    assert {
        "climate",
        "ambient_light",
        "roof",
        "self_extension",
    }.issubset(template_preview["mode_profile_draft"]["domains"])
    template_capabilities = {
        action["capability"]
        for action in template_actions
    }
    assert any(
        capability.startswith("climate.")
        for capability in template_capabilities
    )
    assert "ambient_light.set_power" in template_capabilities
    assert "roof.set_shade_position" in template_capabilities
    assert "infrared.emit" in template_capabilities

    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"] == 3
    assert [action["slot"] for action in explicit["infrared_actions"]] == [1, 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)