Cognitive_Agent/tests/unit/test_mode_scene_ingress.py

from __future__ import annotations

import json
import time
from copy import deepcopy
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from cognitive_agent.application.assistant_capabilities import (
    AssistantCapabilityCatalog,
)
from cognitive_agent.application.mode_profile import ModeProfileRegistry
from cognitive_agent.application.mode_scene_ingress import (
    ModeSceneIngressConsumer,
    ModeSceneIngressError,
    parse_mode_scene_ingress_payload,
)
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService


def scene_card(*, temperature: int = 24) -> dict[str, object]:
    return {
        "id": "n7-reading",
        "label": "N7 阅读模式",
        "name": "N7 阅读模式",
        "background_time": "evening",
        "description": "由车端同步的阅读场景。",
        "icon": "book-open",
        "aliases": ["N7 阅读模式", "车端阅读"],
        "origin": "user",
        "deletable": True,
        "behavior": "static",
        "domains": ["climate"],
        "operation_count": 1,
        "scope": {"common": False, "vehicle_ids": ["denza_n7"]},
        "profile_revision": f"remote-{temperature}",
        "actions": [
            {
                "type": "climate_temperature",
                "at_ms": 250,
                "domain": "climate",
                "capability": "climate.set_temperature",
                "target_scope": "passenger",
                "operation": "set",
                "parameters": {"temperature_c": temperature},
                "zone": "passenger",
                "celsius": temperature,
            }
        ],
    }


def cinema_scene_card() -> dict[str, object]:
    """Representative N7 card containing local and head-unit-only actions."""

    actions: list[dict[str, object]] = [
        *[
            {
                "type": "window_position",
                "at_ms": 0,
                "domain": "window",
                "capability": "window.set_position",
                "target_scope": target,
                "operation": "set",
                "parameters": {"position_pct": 0},
            }
            for target in ("front_left", "front_right", "rear_left", "rear_right")
        ],
        {
            "type": "roof_shade_position",
            "at_ms": 0,
            "domain": "roof",
            "capability": "roof.set_shade_position",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"position_pct": 0},
        },
        {
            "type": "seat_preset",
            "at_ms": 0,
            "domain": "local_vehicle",
            "capability": "local_scene.seat_preset",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"targets_pulse": [246, 112, 405, 342]},
        },
        {
            "type": "seat_ventilation",
            "at_ms": 0,
            "domain": "seat",
            "capability": "seat.set_ventilation",
            "target_scope": "passenger",
            "operation": "set",
            "parameters": {"state": "off"},
        },
        {
            "type": "climate_power",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_power",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"state": "on"},
        },
        {
            "type": "climate_fan",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_fan_level",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"level": 2},
        },
        {
            "type": "climate_outlet_mode",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_outlet_mode",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"mode": "face"},
        },
        {
            "type": "climate_cycle_mode",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_cycle_mode",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"mode": "internal"},
        },
        {
            "type": "climate_control_mode",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_control_mode",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"mode": "manual"},
        },
        {
            "type": "climate_temperature",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_temperature",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"temperature_c": 25},
        },
        {
            "type": "interior_light_reading_power",
            "at_ms": 0,
            "domain": "interior_light",
            "capability": "interior_light.set_reading_power",
            "target_scope": "all",
            "operation": "set",
            "parameters": {"state": "off"},
        },
        {
            "type": "ambient_light_power",
            "at_ms": 0,
            "domain": "ambient_light",
            "capability": "ambient_light.set_power",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"state": "on"},
        },
        {
            "type": "ambient_light_color",
            "at_ms": 0,
            "domain": "ambient_light",
            "capability": "ambient_light.set_color_index",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"color_index": 36},
        },
        {
            "type": "ambient_light_brightness",
            "at_ms": 0,
            "domain": "ambient_light",
            "capability": "ambient_light.set_brightness_level",
            "target_scope": "vehicle",
            "operation": "set",
            "parameters": {"brightness_level": 4},
        },
    ]
    return {
        "id": "cinema",
        "label": "Cinema mode",
        "name": "Cinema mode",
        "background_time": "night",
        "description": "Representative N7 mixed-domain scene.",
        "icon": "bookmark-plus",
        "aliases": ["Cinema mode"],
        "origin": "headunit",
        "deletable": True,
        "behavior": "static",
        "domains": list(dict.fromkeys(str(item["domain"]) for item in actions)),
        "operation_count": len(actions),
        "scope": {"common": False, "vehicle_ids": ["denza_n7"]},
        "actions": actions,
        "profile_revision": "cinema-test-revision",
    }


def wait_until(predicate, timeout: float = 1.5) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if predicate():
            return True
        time.sleep(0.01)
    return bool(predicate())


def configured_service(tmp_path: Path) -> SeatDemoService:
    service = SeatDemoService(REPOSITORY_ROOT)
    service.mode_profiles = ModeProfileRegistry(
        tmp_path / "presets",
        user_root=tmp_path / "profiles",
    )
    service.artifact_root = tmp_path / "runs"
    return service


def test_parser_uses_exact_single_card_contract_and_rejects_duplicate_keys() -> None:
    scene = scene_card()
    parsed = parse_mode_scene_ingress_payload(
        json.dumps(scene, ensure_ascii=False).encode("utf-8")
    )
    assert parsed == scene

    duplicate = (
        '{"id":"one","id":"two","label":"x","name":"x",'
        '"background_time":"noon","description":"",'
        '"icon":"x","aliases":[],"origin":"user",'
        '"deletable":true,"behavior":"static","domains":["x"],'
        '"operation_count":1,"scope":{},"profile_revision":"x",'
        '"actions":[]}'
    )
    with pytest.raises(ModeSceneIngressError, match="duplicate key"):
        parse_mode_scene_ingress_payload(duplicate)


def test_three_identical_frames_are_imported_once() -> None:
    imports: list[dict[str, object]] = []
    consumer = ModeSceneIngressConsumer(
        lambda scene: imports.append(dict(scene)) or {"status": "imported"},
        deduplication_window_s=1.0,
    )
    payload = json.dumps(scene_card(), ensure_ascii=False).encode("utf-8")
    consumer.start()
    try:
        for _ in range(3):
            consumer.submit(payload)
        assert wait_until(lambda: consumer.snapshot()["processed_count"] == 3)
        assert imports == [scene_card()]
        status = consumer.snapshot()
        assert status["imported_count"] == 1
        assert status["duplicate_count"] == 2
        assert status["rejected_count"] == 0
    finally:
        consumer.close()


def test_failed_import_does_not_suppress_identical_retry() -> None:
    importer = MagicMock(
        side_effect=[
            RuntimeError("temporary persistence failure"),
            {"status": "imported"},
        ]
    )
    consumer = ModeSceneIngressConsumer(
        importer,
        deduplication_window_s=1.0,
    )
    payload = json.dumps(scene_card(), ensure_ascii=False).encode("utf-8")
    consumer.start()
    try:
        consumer.submit(payload)
        assert wait_until(lambda: consumer.snapshot()["processed_count"] == 1)
        first_status = consumer.snapshot()
        assert first_status["rejected_count"] == 1
        assert first_status["imported_count"] == 0

        consumer.submit(payload)
        assert wait_until(lambda: consumer.snapshot()["processed_count"] == 2)
        retry_status = consumer.snapshot()
        assert retry_status["rejected_count"] == 1
        assert retry_status["imported_count"] == 1
        assert retry_status["duplicate_count"] == 0
        assert importer.call_count == 2

        consumer.submit(payload)
        assert wait_until(lambda: consumer.snapshot()["processed_count"] == 3)
        final_status = consumer.snapshot()
        assert final_status["duplicate_count"] == 1
        assert importer.call_count == 2
    finally:
        consumer.close()


def test_import_upserts_yaml_refreshes_catalog_and_does_not_echo(
    tmp_path: Path,
) -> None:
    service = configured_service(tmp_path)
    service.configure_mode_scene_catalog(tmp_path / "scenes.json")
    outbound = MagicMock()
    service._vehicle_mqtt_transport.publish_mode_scene = outbound

    first = service.import_n7_mode_scene(scene_card(temperature=24))
    updated = service.import_n7_mode_scene(scene_card(temperature=26))

    assert first["scene_id"] == "n7-reading"
    assert updated["scene_id"] == "n7-reading"
    profiles = service.mode_profiles.public_modes("denza_n7")
    assert len(profiles) == 1
    profile = profiles[0]
    assert profile["origin"] == "n7_mqtt"
    assert profile["source_topic"] == "from_n7/scenes"
    assert profile["upstream_origin"] == "user"
    assert profile["operations"][0]["parameters"] == {"temperature_c": 26}

    catalog = service.mode_scene_catalog("denza_n7")
    assert catalog["scene_count"] == 1
    card = catalog["scenes"][0]
    assert card["origin"] == "n7_mqtt"
    assert card["source_topic"] == "from_n7/scenes"
    assert card["actions"] == scene_card(temperature=26)["actions"]
    assert card["actions"][0]["at_ms"] == 250
    assert outbound.call_count == 0

    yaml_path = Path(service.mode_profiles.get_profile("n7-reading", "denza_n7").source)
    yaml_text = yaml_path.read_text(encoding="utf-8")
    assert "sync_origin: from_n7/scenes" in yaml_text
    assert "upstream_origin: user" in yaml_text


def test_unknown_n7_action_is_forwarded_once_only_during_hardware_execution(
    tmp_path: Path,
) -> None:
    service = configured_service(tmp_path)
    unknown = scene_card()
    unknown.update(
        domains=["media"],
        operation_count=1,
        actions=[
            {
                "type": "media_play_favorite",
                "at_ms": 0,
                "domain": "media",
                "capability": "media.play_favorite",
                "target_scope": "vehicle",
                "operation": "play",
                "parameters": {"playlist": "focus"},
            }
        ],
    )
    service.import_n7_mode_scene(unknown)
    publisher = MagicMock(
        return_value={
            "topic": "transto_n7/ctl",
            "qos": 1,
            "retain": False,
            "delivery_status": "published",
            "message_id": 7,
        }
    )
    service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher

    dry_run = service.execute_mode_profile_card(
        "n7-reading",
        execution_mode="dry_run",
        control_transport="wireless_mqtt",
    )
    assert dry_run["status"] == "capability_unavailable"
    publisher.assert_not_called()

    result = service.execute_mode_profile_card(
        "n7-reading",
        execution_mode="hardware",
        control_transport="wireless_mqtt",
    )

    assert result["status"] == "capability_unavailable"
    assert result["n7_control_transfer"]["state"] == "published"
    publisher.assert_called_once()
    payload = publisher.call_args.args[0]
    assert payload == {
        "scene_id": "n7-reading",
        "scene_name": "N7 阅读模式",
        "request_id": result["request_id"],
        "operation_count": 1,
        "actions": deepcopy(unknown["actions"]),
    }
    assert result["deferred_requests"][0]["transport_submitted"] is False


def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
    tmp_path: Path,
) -> None:
    service = configured_service(tmp_path)
    scene = cinema_scene_card()
    service.import_n7_mode_scene(scene)

    result = service.execute_mode_profile_card(
        "cinema",
        execution_mode="dry_run",
        control_transport="wireless_mqtt",
    )

    assert result["status"] == "partially_succeeded"
    assert not [
        issue
        for issue in result["validation"]
        if issue.get("severity") == "error"
    ]
    compiled_operations = [
        operation
        for config in result["function_configs"].values()
        for operation in config["spec"]["operations"]
    ]
    assert len(compiled_operations) == 16
    assert all(
        operation.get("parse_source") == "saved_mode_command_flow"
        for operation in compiled_operations
    )
    assert {
        operation["target"]["zone"]
        for operation in compiled_operations
        if operation["capability"] == "window.set_position"
    } == {"front_left", "front_right", "rear_left", "rear_right"}
    transferred = scene["actions"][5]
    assert result["n7_transfer_candidates"] == [transferred]
    assert result["deferred_requests"] == [
        {
            "domain": "local_vehicle",
            "capability": "local_scene.seat_preset",
            "target_scope": "vehicle",
            "desired_change": "set",
            "parameters": {"targets_pulse": [246, 112, 405, 342]},
            "reason": "当前 PC 未注册该能力，将由 N7 端继续处理。",
            "execution_backend": "n7_vehicle_transfer",
            "rejection_code": "PC_CAPABILITY_UNAVAILABLE",
            "transport_submitted": False,
        }
    ]
    assert len(compiled_operations) + len(result["n7_transfer_candidates"]) == len(
        scene["actions"]
    )

    text_result = service.process(
        "进入Cinema mode模式",
        provider="local",
        conversation_id="conv-n7-cinema-text",
        execution_mode="dry_run",
        vehicle_id="denza_n7",
        control_transport="wireless_mqtt",
    )
    assert text_result["status"] == "partially_succeeded"
    assert text_result["mode_profile"] is None
    assert [item["id"] for item in text_result["matched_modes"]] == ["cinema"]
    assert len(text_result["n7_transfer_candidates"]) == 1
    assert service.mode_profiles.match_text(
        "氛围灯设置为Cinema mode模式",
        vehicle_id="denza_n7",
    ) == []

    catalog_mode = next(
        mode
        for mode in AssistantCapabilityCatalog(service).snapshot("denza_n7")["modes"]
        if mode["id"] == "cinema"
    )
    assert catalog_mode["assistant_callable"] is True
    assert catalog_mode["n7_transfer_operation_ids"] == [
        "local_scene.seat_preset"
    ]


@pytest.mark.parametrize(
    ("domain", "target_scope", "parameters"),
    [
        ("media", "passenger", {"temperature_c": 24}),
        ("climate", "unsupported_zone", {"temperature_c": 24}),
        ("climate", "passenger", {"unsupported_parameter": 24}),
    ],
)
def test_registered_but_invalid_action_is_not_forwarded_to_n7(
    tmp_path: Path,
    domain: str,
    target_scope: str,
    parameters: dict[str, object],
) -> None:
    service = configured_service(tmp_path)
    invalid = scene_card()
    invalid["actions"] = [
        {
            "type": "climate_temperature",
            "at_ms": 0,
            "domain": domain,
            "capability": "climate.set_temperature",
            "target_scope": target_scope,
            "operation": "set",
            "parameters": parameters,
        }
    ]
    service.import_n7_mode_scene(invalid)
    profile = service.get_mode_profile_card("n7-reading", "denza_n7")

    local_requests, transfer_actions = service._mode_requests_and_n7_transfers(profile)

    assert len(local_requests) == 1
    assert transfer_actions == []


def test_frontend_marks_n7_synchronized_cards() -> None:
    script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text(
        encoding="utf-8"
    )
    styles = (
        REPOSITORY_ROOT / "apps" / "web-console" / "styles.css"
    ).read_text(encoding="utf-8")
    assert 'mode?.origin === "n7_mqtt"' in script
    assert 'sourceBadge.textContent = "N7 同步"' in script
    assert ".mode-gallery-source-badge" in styles


def test_mode_profile_sync_status_exposes_catalog_and_ingress_counters(
    tmp_path: Path,
) -> None:
    service = configured_service(tmp_path)
    service.configure_mode_scene_catalog(tmp_path / "scenes.json")

    status = service.mode_profile_sync_status()

    assert len(status["catalog_revision"]) == 16
    assert status["scene_count"] == 0
    assert status["catalog_state"] == "ready"
    assert status["mode_scene_ingress"]["imported_count"] == 0


def test_frontend_polls_lightweight_status_and_refreshes_on_n7_import() -> None:
    script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text(
        encoding="utf-8"
    )

    assert 'apiRequest("/api/mode-profiles/status"' in script
    assert "MODE_PROFILE_SYNC_POLL_INTERVAL_MS = 1000" in script
    assert 'document.visibilityState === "hidden"' in script
    assert "if (catalogChanged || (imported && !sceneAlreadyVisible))" in script
    assert "await loadRuntime()" in script
    assert "revealModeProfile(sceneId)" in script
    assert "已从 N7 同步模式卡片" in script
    assert "N7 模式卡片导入失败" in script