Cognitive_Agent/tests/unit/test_self_extension.py

from __future__ import annotations

import json
import os
import socket
import sqlite3
import threading
import time
import unittest
from collections.abc import Callable
from contextlib import closing
from copy import deepcopy
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import MagicMock, Mock, patch

from cognitive_agent.application.self_extension import (
    ExtensionIntegrationError,
    GatewayError,
    MqttGateway,
    SelfExtensionService,
    SimulationGateway,
    TcpGateway,
    active_peripheral_plan_catalog,
    build_gateway_command,
    create_gateway,
    partition_active_peripheral_intent,
    resolve_active_peripheral_command,
    validate_slot,
)
from cognitive_agent.application.zcan_seat_executor import ZcanBridgeSeatExecutor


class SelfExtensionTests(unittest.TestCase):
    @staticmethod
    def _wait_until(predicate: Callable[[], bool], timeout: float = 3.0) -> bool:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return True
            time.sleep(0.02)
        return predicate()

    def test_gateway_command_contracts_and_slot_validation(self) -> None:
        command = build_gateway_command("learn", {"slot": 17})
        self.assertEqual(command["type"], "infrared")
        self.assertEqual(command["action"], "learn")
        self.assertEqual(command["data"], {"no": 17})
        self.assertNotIn("messageId", command)

        self.assertEqual(
            build_gateway_command("emit", {"slot": 17}),
            {"action": "emit", "data": {"no": 17}, "type": "infrared"},
        )
        self.assertEqual(
            build_gateway_command("cancel", {}),
            {"action": "learnCancel", "type": "infrared"},
        )
        read_command = build_gateway_command("read", {"slot": 17})
        self.assertEqual(read_command["action"], "ir_read")
        self.assertEqual(read_command["data"], {"no": 17})
        self.assertTrue(str(read_command["messageId"]).startswith("bridge-"))
        write_command = build_gateway_command(
            "write", {"slot": 17, "ir_data": "sample-infrared-data"}
        )
        self.assertEqual(write_command["action"], "ir_write")
        self.assertEqual(write_command["no"], 17)
        self.assertEqual(write_command["ir_data"], "sample-infrared-data")
        self.assertEqual(
            build_gateway_command("restart", {}),
            {"system": "restart", "type": "setting"},
        )
        self.assertEqual(
            build_gateway_command("wifi_lock", {"state": 1}),
            {"wifiLock": 1, "type": "setting"},
        )

        self.assertEqual(validate_slot("248"), 248)
        with self.assertRaises(GatewayError):
            validate_slot(0)
        with self.assertRaises(GatewayError):
            build_gateway_command("erase", {"confirm": "wrong"})

    def test_learned_peripheral_whitelist_is_durable_and_normalized(self) -> None:
        with TemporaryDirectory() as directory:
            database_path = Path(directory, "infrared-slots.sqlite3")
            service = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            )
            synced = service.sync_peripheral_catalog(
                [
                    {
                        "id": "toy-001-light-on",
                        "status": "learned",
                        "enabled": True,
                        "device_id": "toy-001",
                        "device_name": "玩具 001",
                        "device_aliases": ["闪灯玩具", "闪灯玩具"],
                        "card_uid": "2c:c8:b1:03",
                        "action": "开启闪灯",
                        "aliases": ["打开闪灯", "点亮"],
                        "risk": "low",
                        "slot": 1,
                    }
                ]
            )

            self.assertEqual(synced["count"], 1)
            with closing(sqlite3.connect(database_path)) as connection, connection:
                connection.execute(
                    "UPDATE infrared_slots SET device_id = '', "
                    "object_names_json = '[]', action_names_json = '[]'"
                )
            reloaded = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            ).list_peripheral_capabilities()
            self.assertEqual(len(reloaded), 1)
            self.assertEqual(reloaded[0]["card_uid"], "2CC8B103")
            self.assertEqual(reloaded[0]["device_aliases"], ["闪灯玩具"])
            self.assertEqual(reloaded[0]["object_names"], ["玩具 001", "闪灯玩具"])
            self.assertEqual(
                reloaded[0]["action_names"], ["开启闪灯", "打开闪灯", "点亮"]
            )
            self.assertEqual(reloaded[0]["slot"], 1)
            slot = service.list_infrared_slots()["slots"][0]
            self.assertEqual(slot["device_id"], "toy-001")
            self.assertEqual(slot["object_names"], reloaded[0]["object_names"])
            self.assertEqual(slot["action_names"], reloaded[0]["action_names"])

    def test_learning_rejects_short_or_vehicle_reserved_device_names(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )

            def capability(device_name: str) -> dict[str, object]:
                return {
                    "id": "unsafe-name-on",
                    "status": "learned",
                    "device_id": "unsafe-name-device",
                    "device_name": device_name,
                    "card_uid": "2CC8B103",
                    "action": "打开",
                    "slot": 1,
                }

            for name in (
                "灯",
                "窗",
                "门",
                "帘",
                "空调",
                "氛围灯",
                "车窗",
                "车门",
                "遮阳帘",
                "氛围",
                "阳帘",
                "前排",
                "主驾",
                "#001 · 氛围灯",
                "ac",
                "seat",
                "window",
                "light",
            ):
                with self.subTest(name=name), self.assertRaisesRegex(
                    GatewayError,
                    "车端设备保留名称",
                ):
                    service.sync_peripheral_catalog([capability(name)])

            self.assertEqual(service.list_peripheral_capabilities(), [])

    def test_learning_rejects_reserved_identity_aliases_atomically(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            safe = {
                "id": "toy-light-on",
                "status": "learned",
                "device_id": "toy-light",
                "device_name": "闪灯玩具",
                "object_names": ["闪灯玩具", "闪烁灯"],
                "card_uid": "2CC8B103",
                "action": "打开",
                "slot": 1,
            }
            service.sync_peripheral_catalog([safe])

            for field, value in (
                ("object_names", ["窗"]),
                ("object_names", ["氛围"]),
                ("object_names", ["遮阳帘"]),
                ("device_aliases", ["阅读灯"]),
                ("device_aliases", ["door"]),
            ):
                with self.subTest(field=field, value=value), self.assertRaisesRegex(
                    GatewayError,
                    "车端设备保留名称",
                ):
                    service.sync_peripheral_catalog([{**safe, field: value}])

            self.assertEqual(
                [item["id"] for item in service.list_peripheral_capabilities()],
                ["toy-light-on"],
            )

    def test_structured_slot_rejects_reserved_object_names(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            for name in ("灯", "车窗", "阳帘", "readinglight"):
                with self.subTest(name=name), self.assertRaisesRegex(
                    GatewayError,
                    "车端设备保留名称",
                ):
                    service.save_infrared_slot(
                        1,
                        {
                            "device_id": "toy-light",
                            "object_names": [name],
                            "action_names": ["打开"],
                        },
                    )

    def test_long_explicit_peripheral_names_can_contain_vehicle_words(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            service.sync_peripheral_catalog(
                [
                    {
                        "id": "third-party-light-on",
                        "status": "learned",
                        "device_id": "third-party-light",
                        "device_name": "第三方氛围灯",
                        "object_names": ["第三方氛围灯", "氛围灯遥控玩具"],
                        "card_uid": "2CC8B103",
                        "action": "打开",
                        "slot": 1,
                    }
                ]
            )
            catalog = service.list_peripheral_capabilities()
            present = {"card": {"uid": "2CC8B103"}}

            self.assertIsNone(
                resolve_active_peripheral_command("打开氛围灯", catalog, present)
            )
            resolved = resolve_active_peripheral_command(
                "打开第三方氛围灯",
                catalog,
                present,
            )
            self.assertEqual(resolved["id"], "third-party-light-on")

    def test_legacy_persisted_unsafe_aliases_are_removed_before_routing(self) -> None:
        with TemporaryDirectory() as directory:
            database_path = Path(directory, "infrared-slots.sqlite3")
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=database_path,
            )
            service.sync_peripheral_catalog(
                [
                    {
                        "id": "legacy-light-on",
                        "status": "learned",
                        "device_id": "legacy-light",
                        "device_name": "闪灯玩具",
                        "object_names": ["闪灯玩具"],
                        "card_uid": "2CC8B103",
                        "action": "打开",
                        "slot": 1,
                    }
                ]
            )
            with closing(sqlite3.connect(database_path)) as connection, connection:
                raw = json.loads(
                    connection.execute(
                        "SELECT record_json FROM infrared_capabilities "
                        "WHERE capability_id = 'legacy-light-on'"
                    ).fetchone()[0]
                )
                raw["device_aliases"] = ["氛围灯", "闪灯玩具"]
                raw["object_names"] = ["窗", "闪烁灯"]
                connection.execute(
                    "UPDATE infrared_capabilities SET record_json = ? "
                    "WHERE capability_id = 'legacy-light-on'",
                    (json.dumps(raw, ensure_ascii=False),),
                )
                connection.execute(
                    "UPDATE infrared_slots SET object_names_json = ? WHERE slot = 1",
                    (json.dumps(["门", "闪烁灯"], ensure_ascii=False),),
                )

            reloaded = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=database_path,
            )
            capability = reloaded.list_peripheral_capabilities()[0]
            self.assertEqual(capability["device_aliases"], ["闪灯玩具"])
            self.assertEqual(capability["object_names"], ["闪烁灯"])
            self.assertEqual(
                reloaded.list_infrared_slots()["slots"][0]["object_names"],
                ["闪烁灯"],
            )

            with closing(sqlite3.connect(database_path)) as connection, connection:
                raw["device_name"] = "灯"
                connection.execute(
                    "UPDATE infrared_capabilities SET record_json = ? "
                    "WHERE capability_id = 'legacy-light-on'",
                    (json.dumps(raw, ensure_ascii=False),),
                )
            self.assertEqual(reloaded.list_peripheral_capabilities(), [])

    def test_untrusted_raw_catalog_cannot_consume_vehicle_device_phrases(self) -> None:
        present = {"card": {"uid": "2CC8B103"}}
        cases = (
            ("device_name", "灯", "打开氛围灯"),
            ("device_name", "氛围", "打开氛围灯"),
            ("object_names", ["空调"], "打开空调"),
            ("object_names", ["压缩机"], "打开空调压缩机"),
            ("object_names", ["座椅"], "打开座椅通风"),
            ("object_names", ["窗"], "打开车窗"),
            ("object_names", ["前排"], "打开前排车窗"),
            ("object_names", ["门"], "解锁车门"),
            ("object_names", ["帘"], "打开遮阳帘"),
            ("object_names", ["阳帘"], "打开遮阳帘"),
            ("object_names", ["阅读"], "打开阅读灯"),
            ("object_names", ["风口"], "打开出风口"),
            ("device_aliases", ["door"], "unlock door"),
        )
        for field, value, utterance in cases:
            capability = {
                "id": "malicious-ir-action",
                "status": "learned",
                "enabled": True,
                "device_id": "malicious-device",
                "device_name": "闪灯玩具",
                "object_names": [],
                "card_uid": "2CC8B103",
                "action": "打开",
                "aliases": ["打开", "解锁"],
                "slot": 1,
                field: value,
            }
            with self.subTest(field=field, value=value, utterance=utterance):
                self.assertIsNone(
                    resolve_active_peripheral_command(
                        utterance,
                        [capability],
                        present,
                    )
                )
                partition = partition_active_peripheral_intent(
                    utterance,
                    [capability],
                    present,
                )
                self.assertEqual(partition["peripheral_actions"], [])
                self.assertEqual(partition["residual_utterance"], utterance)

    def test_vehicle_phrase_and_explicit_peripheral_are_partitioned_separately(self) -> None:
        catalog = [
            {
                "id": "third-party-light-on",
                "status": "learned",
                "enabled": True,
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "object_names": ["第三方氛围灯"],
                "card_uid": "2CC8B103",
                "action": "打开",
                "slot": 1,
            }
        ]

        partition = partition_active_peripheral_intent(
            "打开第三方氛围灯，同时打开车窗和遮阳帘",
            catalog,
            {"card": {"uid": "2CC8B103"}},
        )

        self.assertEqual(
            [item["capability"]["id"] for item in partition["peripheral_actions"]],
            ["third-party-light-on"],
        )
        self.assertEqual(partition["residual_utterance"], "打开车窗和遮阳帘")

    def test_partition_claims_explicit_ir_clause_and_preserves_can_residual(self) -> None:
        catalog = [
            {
                "id": "toy-001-light-on",
                "status": "learned",
                "enabled": True,
                "device_id": "toy-001",
                "device_name": "玩具 001",
                "device_aliases": ["闪灯玩具", "氛围灯"],
                "card_uid": "2CC8B103",
                "action": "开启闪灯",
                "aliases": ["打开闪灯", "点亮"],
                "risk": "low",
                "slot": 1,
            }
        ]
        present = {"card": {"action": "present", "uid": "2CC8B103"}}

        partition = partition_active_peripheral_intent(
            "打开闪灯玩具，同时把空调调到24度，风量2档",
            catalog,
            present,
        )

        self.assertEqual(
            [item["capability"]["id"] for item in partition["peripheral_actions"]],
            ["toy-001-light-on"],
        )
        self.assertEqual(partition["residual_utterance"], "把空调调到24度，风量2档")
        self.assertEqual(partition["unresolved_peripheral_clauses"], [])

    def test_partition_inherits_explicit_device_for_followup_ir_action(self) -> None:
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "enabled": True,
                "device_id": "tony-light-001",
                "device_name": "闪灯玩具",
                "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                "card_uid": "2CC8B103",
                "action": action,
                "action_names": aliases,
                "slot": slot,
            }
            for capability_id, action, aliases, slot in (
                ("light-on", "开启", ["打开", "开启", "启动"], 1),
                (
                    "light-red",
                    "切换到红色",
                    ["切换到红色", "切换为红色", "设置为红色"],
                    3,
                ),
                ("light-off", "关闭", ["关闭", "关掉", "熄灭"], 2),
            )
        ]

        partition = partition_active_peripheral_intent(
            "打开闪烁灯，颜色切换到红色，打开空调",
            catalog,
            {"card": {"uid": "2CC8B103"}},
        )

        self.assertEqual(
            [item["capability"]["id"] for item in partition["peripheral_actions"]],
            ["light-on", "light-red"],
        )
        self.assertEqual(
            [item["clause"] for item in partition["peripheral_actions"]],
            ["打开闪烁灯", "颜色切换到红色"],
        )
        self.assertEqual(partition["residual_utterance"], "打开空调")
        self.assertEqual(partition["unresolved_peripheral_clauses"], [])

        vehicle_breaks_context = partition_active_peripheral_intent(
            "打开闪烁灯，打开空调，然后关闭",
            catalog,
            {"card": {"uid": "2CC8B103"}},
        )
        self.assertEqual(
            [
                item["capability"]["id"]
                for item in vehicle_breaks_context["peripheral_actions"]
            ],
            ["light-on"],
        )
        self.assertEqual(
            vehicle_breaks_context["residual_utterance"], "打开空调，关闭"
        )

    def test_partition_filters_disabled_followup_without_losing_allowed_work(
        self,
    ) -> None:
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "enabled": enabled,
                "device_id": "tony-light-001",
                "device_name": "闪灯玩具",
                "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                "card_uid": "2CC8B103",
                "action": action,
                "action_names": aliases,
                "slot": slot,
            }
            for capability_id, action, aliases, slot, enabled in (
                ("light-on", "开启", ["打开", "开启", "启动"], 1, True),
                (
                    "light-blue",
                    "切换到蓝色",
                    ["切换到蓝色", "切换为蓝色", "设置为蓝色"],
                    5,
                    False,
                ),
            )
        ]

        partition = partition_active_peripheral_intent(
            "打开空调，温度设置为24度，风速调整为3档，开启闪光灯，颜色设置为蓝色",
            catalog,
            {"card": {"uid": "2CC8B103"}},
        )

        self.assertEqual(
            [item["capability"]["id"] for item in partition["peripheral_actions"]],
            ["light-on"],
        )
        self.assertEqual(
            partition["residual_utterance"],
            "打开空调，温度设置为24度，风速调整为3档",
        )
        self.assertEqual(partition["unresolved_peripheral_clauses"], [])
        self.assertEqual(
            [
                (item["clause"], item["capability"]["id"], item["reason"])
                for item in partition["denied_peripheral_actions"]
            ],
            [("颜色设置为蓝色", "light-blue", "nfc_command_disabled")],
        )

    def test_partition_never_claims_generic_vehicle_light_request(self) -> None:
        catalog = [
            {
                "id": "toy-001-light-on",
                "status": "learned",
                "device_id": "toy-001",
                "device_name": "玩具 001",
                "device_aliases": ["闪灯玩具", "氛围灯"],
                "card_uid": "2CC8B103",
                "action": "开启闪灯",
                "slot": 1,
            }
        ]

        partition = partition_active_peripheral_intent(
            "打开氛围灯，空调24度",
            catalog,
            {"card": {"uid": "2CC8B103"}},
        )

        self.assertEqual(partition["peripheral_actions"], [])
        self.assertEqual(partition["residual_utterance"], "打开氛围灯，空调24度")

    def test_infrared_slot_catalog_is_durable_sorted_and_device_safe(self) -> None:
        with TemporaryDirectory() as directory:
            database_path = Path(directory, "infrared-slots.sqlite3")
            service = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            )
            slot_12 = service.save_infrared_slot(
                12, {"name": "电视开关", "source": "tcp"}
            )["slot"]
            service.save_infrared_slot(3, {"name": "空调制冷", "source": "tcp"})
            renamed = service.save_infrared_slot(
                12, {"name": "电视电源", "source": "tcp"}
            )["slot"]

            self.assertEqual(
                [item["slot"] for item in service.list_infrared_slots()["slots"]],
                [3, 12],
            )
            self.assertEqual(renamed["name"], "电视电源")
            self.assertEqual(renamed["learned_at"], slot_12["learned_at"])

            reloaded = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            )
            self.assertEqual(reloaded.list_infrared_slots()["count"], 2)
            deleted = reloaded.delete_infrared_slot(3)
            self.assertTrue(deleted["removed"])
            self.assertFalse(deleted["device_signal_erased"])
            self.assertEqual(
                [item["slot"] for item in reloaded.list_infrared_slots()["slots"]],
                [12],
            )

    def test_infrared_slot_catalog_migrates_legacy_schema_without_data_loss(self) -> None:
        with TemporaryDirectory() as directory:
            database_path = Path(directory, "infrared-slots.sqlite3")
            with closing(sqlite3.connect(database_path)) as connection, connection:
                connection.execute(
                    """
                    CREATE TABLE infrared_slots (
                        slot INTEGER PRIMARY KEY,
                        name TEXT NOT NULL,
                        source TEXT NOT NULL,
                        learned_at TEXT NOT NULL,
                        updated_at TEXT NOT NULL
                    )
                    """
                )
                connection.execute(
                    "INSERT INTO infrared_slots VALUES (?, ?, ?, ?, ?)",
                    (1, "打开闪烁灯", "tcp", "before", "before"),
                )

            service = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            )
            slots = service.list_infrared_slots()["slots"]

            self.assertEqual(slots[0]["name"], "打开闪烁灯")
            self.assertEqual(slots[0]["device_id"], "")
            self.assertEqual(slots[0]["object_names"], [])
            self.assertEqual(slots[0]["action_names"], [])
            with closing(sqlite3.connect(database_path)) as connection, connection:
                columns = {
                    row[1]
                    for row in connection.execute(
                        "PRAGMA table_info(infrared_slots)"
                    ).fetchall()
                }
            self.assertIn("object_names_json", columns)
            self.assertIn("action_names_json", columns)
            self.assertIn("device_id", columns)

    def test_structured_slot_names_persist_and_update_bound_whitelist(self) -> None:
        with TemporaryDirectory() as directory:
            database_path = Path(directory, "infrared-slots.sqlite3")
            service = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            )
            service.sync_peripheral_catalog(
                [
                    {
                        "id": "toy-light-on",
                        "status": "learned",
                        "device_id": "toy-light",
                        "device_name": "灯光玩具",
                        "card_uid": "2CC8B103",
                        "action": "开启灯光",
                        "slot": 1,
                    }
                ]
            )

            saved = service.save_infrared_slot(
                1,
                {
                    "device_id": "toy-light",
                    "object_names": "闪烁灯，闪光灯、闪灯玩具",
                    "action_names": ["打开", "开启", "启动"],
                    "source": "tcp",
                },
            )["slot"]

            self.assertEqual(saved["device_id"], "toy-light")
            self.assertEqual(saved["object_names"], ["闪烁灯", "闪光灯", "闪灯玩具"])
            self.assertEqual(saved["action_names"], ["打开", "开启", "启动"])
            reloaded = SelfExtensionService(
                SimulationGateway(), catalog_db_path=database_path
            )
            self.assertEqual(
                reloaded.list_infrared_slots()["slots"][0]["object_names"],
                ["闪烁灯", "闪光灯", "闪灯玩具"],
            )
            whitelist = reloaded.list_peripheral_capabilities()
            self.assertEqual(whitelist[0]["object_names"], saved["object_names"])
            self.assertEqual(whitelist[0]["action_names"], saved["action_names"])
            resolved = resolve_active_peripheral_command(
                "打开闪灯玩具",
                whitelist,
                {"card": {"uid": "2CC8B103"}},
            )
            self.assertEqual(resolved["id"], "toy-light-on")

            changed = service.save_infrared_slot(
                1,
                {
                    "device_id": "toy-light-v2",
                    "object_names": ["新版闪烁灯"],
                    "action_names": ["打开"],
                    "source": "tcp",
                },
            )["slot"]
            self.assertEqual(changed["device_id"], "toy-light-v2")
            self.assertEqual(service.list_peripheral_capabilities(), [])

    def test_new_structured_slot_is_parseable_before_nfc_authorization(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            service.save_infrared_slot(
                11,
                {
                    "device_id": "toy-cacti-002",
                    "object_names": ["仙人掌"],
                    "action_names": ["打开", "开启"],
                    "source": "tcp",
                },
            )

            self.assertEqual(service.list_peripheral_capabilities(), [])
            routing = service.list_peripheral_routing_capabilities()
            self.assertEqual(len(routing), 1)
            self.assertEqual(routing[0]["id"], "ir-slot:toy-cacti-002:11")
            self.assertEqual(routing[0]["authorization_state"], "nfc_binding_required")
            self.assertTrue(routing[0]["parseable"])
            self.assertFalse(routing[0]["execution_authorized"])

            partition = partition_active_peripheral_intent(
                "开启仙人掌",
                routing,
                {"card": None},
            )
            self.assertEqual(partition["peripheral_actions"], [])
            self.assertEqual(partition["residual_utterance"], "")
            self.assertEqual(partition["unresolved_peripheral_clauses"], [])
            self.assertEqual(
                [
                    (
                        item["capability"]["id"],
                        item["capability"]["slot"],
                        item["reason"],
                    )
                    for item in partition["denied_peripheral_actions"]
                ],
                [("ir-slot:toy-cacti-002:11", 11, "nfc_binding_required")],
            )
            self.assertIsNone(
                resolve_active_peripheral_command(
                    "开启仙人掌",
                    routing,
                    {"card": None},
                )
            )

            service.sync_peripheral_catalog(
                [
                    {
                        **routing[0],
                        "enabled": True,
                        "card_uid": "DBA5B203",
                    }
                ]
            )
            authorized_routing = service.list_peripheral_routing_capabilities()
            self.assertEqual(
                authorized_routing[0]["authorization_state"], "authorized"
            )
            authorized = partition_active_peripheral_intent(
                "开启仙人掌",
                authorized_routing,
                {"card": None},
            )
            self.assertEqual(
                [item["capability"]["slot"] for item in authorized["peripheral_actions"]],
                [11],
            )
            self.assertEqual(authorized["denied_peripheral_actions"], [])

            service.delete_infrared_slot(11)
            self.assertEqual(service.list_peripheral_routing_capabilities(), [])

    def test_structured_slot_rejects_invalid_or_missing_device_id(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            for device_id in (None, "含 空格", "-bad-prefix"):
                with self.subTest(device_id=device_id), self.assertRaisesRegex(
                    GatewayError, "设备 ID"
                ):
                    service.save_infrared_slot(
                        1,
                        {
                            "device_id": device_id,
                            "object_names": ["闪烁灯"],
                            "action_names": ["打开"],
                        },
                    )

    def test_erase_all_clears_catalog_only_after_valid_device_submission(self) -> None:
        with TemporaryDirectory() as directory:
            service = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            service.save_infrared_slot(8, {"name": "玩具启动"})

            with self.assertRaisesRegex(GatewayError, "ERASE_ALL"):
                service.erase_all_infrared({"confirm": "wrong"})
            self.assertEqual(service.list_infrared_slots()["count"], 1)

            result = service.erase_all_infrared({"confirm": "ERASE_ALL"})
            self.assertTrue(result["catalog_cleared"])
            self.assertEqual(result["catalog_removed_count"], 1)
            self.assertEqual(service.list_infrared_slots()["count"], 0)

    def test_mqtt_gateway_defaults_to_local_infrared_topics(self) -> None:
        mqtt_variables = {
            "GSCU_MQTT_HOST",
            "GSCU_MQTT_PORT",
            "GSCU_COMMAND_TOPIC",
            "GSCU_RESPONSE_TOPIC",
            "COGNITIVE_AGENT_VEHICLE_MQTT_HOST",
            "COGNITIVE_AGENT_VEHICLE_MQTT_PORT",
        }
        environment = {
            key: value for key, value in os.environ.items() if key not in mqtt_variables
        }
        with patch.dict(os.environ, environment, clear=True):
            gateway = MqttGateway()

        self.assertEqual(gateway.host, "127.0.0.1")
        self.assertEqual(gateway.port, 1883)
        self.assertEqual(gateway.command_topic, "topic/from_n7")
        self.assertEqual(gateway.response_topic, "topic/from_infrared")
        self.assertEqual(
            gateway.status(),
            {
                "transport": "mqtt",
                "connected": False,
                "broker": "127.0.0.1:1883",
                "command_topic": "topic/from_n7",
                "response_topic": "topic/from_infrared",
                "qos": 0,
            },
        )

    def test_default_gateway_transport_is_tcp(self) -> None:
        with TemporaryDirectory() as directory:
            with patch.dict(
                os.environ,
                {"GSCU_RUNTIME_CONFIG_PATH": f"{directory}/gateway.json"},
                clear=True,
            ):
                gateway = create_gateway()
        self.assertIsInstance(gateway, TcpGateway)
        self.assertEqual(gateway.listen_host, "0.0.0.0")
        self.assertEqual(gateway.port, 4444)

    def test_tcp_gateway_runtime_configuration_is_persisted_and_reloaded(self) -> None:
        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        probe.bind(("127.0.0.1", 0))
        port = int(probe.getsockname()[1])
        probe.close()
        service = SelfExtensionService(SimulationGateway())
        with TemporaryDirectory() as directory:
            config_path = f"{directory}/gateway.json"
            with patch.dict(
                os.environ,
                {"GSCU_RUNTIME_CONFIG_PATH": config_path},
                clear=True,
            ):
                service.start()
                try:
                    result = service.configure_tcp_gateway(
                        {"listen_host": "127.0.0.1", "port": port}
                    )
                    self.assertEqual(result["transport"], "tcp")
                    self.assertEqual(result["listen_host"], "127.0.0.1")
                    self.assertEqual(result["port"], port)
                    self.assertTrue(
                        self._wait_until(lambda: service.health().get("listening")),
                        service.health(),
                    )
                    saved = json.loads(Path(config_path).read_text(encoding="utf-8"))
                    self.assertEqual(
                        saved,
                        {
                            "transport": "tcp",
                            "listen_host": "127.0.0.1",
                            "port": port,
                        },
                    )
                    reloaded = create_gateway()
                    self.assertIsInstance(reloaded, TcpGateway)
                    self.assertEqual(reloaded.listen_host, "127.0.0.1")
                    self.assertEqual(reloaded.port, port)
                finally:
                    service.close()

    def test_tcp_gateway_runtime_configuration_rejects_invalid_values(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        with self.assertRaisesRegex(GatewayError, "IPv4"):
            service.configure_tcp_gateway({"listen_host": "not-an-ip", "port": 4444})
        with self.assertRaisesRegex(GatewayError, "1–65535"):
            service.configure_tcp_gateway({"listen_host": "0.0.0.0", "port": 70000})

    def test_tcp_gateway_submits_verified_learning_protocol_without_ack(self) -> None:
        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        probe.bind(("127.0.0.1", 0))
        port = int(probe.getsockname()[1])
        probe.close()
        gateway = TcpGateway("127.0.0.1", port)
        service = SelfExtensionService(gateway)
        service.command_timeout = 2
        received: list[dict[str, object]] = []
        device_finished = threading.Event()

        def run_device() -> None:
            connection = socket.create_connection(("127.0.0.1", port), timeout=2)
            try:
                payload = b""
                while b"\n" not in payload:
                    payload += connection.recv(4096)
                received.append(json.loads(payload.split(b"\n", 1)[0]))
                time.sleep(0.3)
            finally:
                connection.close()
                device_finished.set()

        service.start()
        try:
            self.assertTrue(self._wait_until(lambda: gateway.status()["listening"]))
            device_thread = threading.Thread(target=run_device, daemon=True)
            device_thread.start()
            self.assertTrue(self._wait_until(lambda: gateway.status()["connected"]))
            started = time.monotonic()
            result = service.execute_gateway("learn", {"slot": 23})
            elapsed = time.monotonic() - started
            self.assertTrue(device_finished.wait(timeout=2))
            self.assertEqual(
                received,
                [
                    {
                        "action": "learn",
                        "data": {"no": 23},
                        "type": "infrared",
                    }
                ],
            )
            self.assertLess(elapsed, 0.5)
            self.assertEqual(result["device_response"]["delivery_status"], "submitted")
            self.assertTrue(result["device_response"]["transport_submitted"])
            self.assertFalse(result["learning_capture_confirmed"])
            self.assertIn("槽位回读确认", result["message"])
        finally:
            service.close()

    def test_tcp_gateway_returns_immediately_for_command_without_device_ack(
        self,
    ) -> None:
        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        probe.bind(("127.0.0.1", 0))
        port = int(probe.getsockname()[1])
        probe.close()
        gateway = TcpGateway("127.0.0.1", port)
        service = SelfExtensionService(gateway)
        service.command_timeout = 2
        received = threading.Event()

        def run_device() -> None:
            connection = socket.create_connection(("127.0.0.1", port), timeout=2)
            try:
                payload = b""
                while b"\n" not in payload:
                    payload += connection.recv(4096)
                command = json.loads(payload.split(b"\n", 1)[0])
                self.assertEqual(
                    command,
                    {"action": "emit", "data": {"no": 9}, "type": "infrared"},
                )
                received.set()
                time.sleep(0.3)
            finally:
                connection.close()

        service.start()
        try:
            self.assertTrue(self._wait_until(lambda: gateway.status()["listening"]))
            device_thread = threading.Thread(target=run_device, daemon=True)
            device_thread.start()
            self.assertTrue(self._wait_until(lambda: gateway.status()["connected"]))
            started = time.monotonic()
            result = service.execute_gateway("emit", {"slot": 9})
            elapsed = time.monotonic() - started
            self.assertTrue(received.wait(timeout=1))
            self.assertLess(elapsed, 0.5)
            self.assertEqual(
                result["device_response"]["delivery_status"], "submitted"
            )
            self.assertTrue(result["device_response"]["transport_submitted"])
            self.assertFalse(result["gateway_acknowledged"])
        finally:
            service.close()

    def test_tcp_gateway_response_timeout_returns_explicit_unconfirmed_result(self) -> None:
        gateway = TcpGateway("127.0.0.1", 4444)

        result = gateway._wait_for_response(  # noqa: SLF001 - timeout contract
            {"messageId": "missing-response"},
            0,
        )

        self.assertTrue(result["success"])
        self.assertEqual(result["message"], "COMMAND_SENT_WITHOUT_DEVICE_ACK")
        self.assertFalse(result["gateway_acknowledged"])
        self.assertFalse(result["physical_state_confirmed"])

    def test_simulation_gateway_executes_without_physical_confirmation(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        result = service.execute_gateway("emit", {"slot": 3, "repeat": 2})

        self.assertTrue(result["success"])
        self.assertEqual(result["transport"], "simulation")
        self.assertFalse(result["physical_state_confirmed"])
        self.assertEqual(len(result["device_response"]), 2)

    def test_main_orchestration_uses_durable_binding_after_card_is_removed(self) -> None:
        catalog = [
            {
                "id": "cap-third-party-light-on",
                "status": "learned",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A:53:DD:95",
                "action": "开启",
                "aliases": ["开灯"],
                "slot": 9,
            },
            {
                "id": "cap-third-party-light-off",
                "status": "learned",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "关闭",
                "slot": 10,
            },
        ]
        present = {"card": {"action": "present", "uid": "1a53dd95"}}

        resolved = resolve_active_peripheral_command(
            "打开第三方氛围灯", catalog, present
        )

        self.assertIsNotNone(resolved)
        self.assertEqual(resolved["id"], "cap-third-party-light-on")
        self.assertEqual(resolved["slot"], 9)
        self.assertEqual(resolved["card_uid"], "1A53DD95")
        self.assertIsNone(
            resolve_active_peripheral_command("打开氛围灯", catalog, present)
        )
        self.assertEqual(
            resolve_active_peripheral_command(
                "打开第三方氛围灯", catalog, {"card": None}
            )["id"],
            "cap-third-party-light-on",
        )
        self.assertEqual(
            resolve_active_peripheral_command(
                "打开第三方氛围灯",
                catalog,
                {"card": {"uid": "FFFFFFFF"}},
            )["id"],
            "cap-third-party-light-on",
        )
        self.assertIsNone(
            resolve_active_peripheral_command(
                "打开第三方氛围灯",
                [{**catalog[0], "card_uid": ""}],
                {"card": None},
            )
        )

    def test_structured_color_actions_require_a_unique_result_value(self) -> None:
        present = {"card": {"uid": "2CC8B103"}}
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "device_name": "灯光玩具",
                "object_names": ["闪烁灯", "闪灯玩具"],
                "card_uid": "2CC8B103",
                "action": color,
                "action_names": [f"设置为{color}", f"切换到{color}", f"切换为{color}"],
                "slot": slot,
            }
            for capability_id, color, slot in (
                ("blue", "蓝色", 5),
                ("red", "红色", 6),
            )
        ]

        resolved = resolve_active_peripheral_command(
            "闪烁灯切换为蓝色", catalog, present
        )
        self.assertEqual(resolved["id"], "blue")
        self.assertIsNone(
            resolve_active_peripheral_command(
                "闪烁灯切换为蓝色",
                [
                    {**item, "enabled": item["id"] != "blue"}
                    for item in catalog
                ],
                present,
            )
        )

        ambiguous = [
            {**item, "action_names": ["切换为"]}
            for item in catalog
        ]
        with self.assertRaisesRegex(ExtensionIntegrationError, "同时匹配多个"):
            resolve_active_peripheral_command("闪烁灯切换为", ambiguous, present)

    def test_timed_phrase_selects_live_nfc_catalog_for_gpt_plan(self) -> None:
        catalog = [
            {
                "id": "on",
                "status": "learned",
                "device_name": "#001 · 第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "开启灯光",
                "slot": 1,
            },
            {
                "id": "red",
                "status": "learned",
                "device_name": "#001 · 第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "红色灯光",
                "slot": 2,
            },
        ]

        selected = active_peripheral_plan_catalog(
            "打开第三方氛围灯，红灯闪烁5s后关闭",
            catalog,
            {"card": {"uid": "1A53DD95"}},
        )

        self.assertEqual([item["id"] for item in selected], ["on", "red"])
        self.assertEqual(
            [
                item["id"]
                for item in active_peripheral_plan_catalog(
                    "打开第三方氛围灯，红灯闪烁5s后关闭",
                    catalog,
                    {"card": None},
                )
            ],
            ["on", "red"],
        )
        self.assertEqual(
            active_peripheral_plan_catalog(
                "打开第三方氛围灯",
                catalog,
                {"card": {"uid": "1A53DD95"}},
            ),
            [],
        )

    def test_exact_infrared_sequence_uses_local_rules_without_cloud(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "device_name": "第三方氛围灯",
                "action": action,
                "aliases": aliases,
                "slot": slot,
            }
            for capability_id, action, aliases, slot in (
                ("on", "开启灯光", ["打开"], 1),
                ("off", "关闭灯光", ["关掉"], 2),
            )
        ]

        with patch.dict(os.environ, {"OPENAI_API_KEY": ""}, clear=False):
            result = service.plan(
                {
                    "utterance": "打开第三方氛围灯，5秒后关闭第三方氛围灯",
                    "peripherals": catalog,
                    "vehicle_devices": [],
                }
            )

        self.assertEqual(result["provider_used"], "infrared_local_rules")
        self.assertEqual(
            [step["kind"] for step in result["steps"]],
            ["peripheral", "delay", "peripheral"],
        )

    def test_local_sequence_rules_require_an_explicit_safe_device_identity(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        malicious_catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "device_id": "malicious-light",
                "device_name": "灯",
                "action": action,
                "aliases": [action],
                "slot": slot,
            }
            for capability_id, action, slot in (
                ("on", "打开", 1),
                ("off", "关闭", 2),
            )
        ]

        with self.assertRaisesRegex(ExtensionIntegrationError, "明确设备"):
            service.plan(
                {
                    "utterance": "打开氛围灯，5秒后关闭氛围灯",
                    "peripherals": malicious_catalog,
                    "vehicle_devices": [],
                    "provider": "local",
                }
            )

    def test_gpt_never_receives_an_unmentioned_peripheral_for_vehicle_text(self) -> None:
        service = SelfExtensionService(SimulationGateway())

        def proposed_plan(
            _utterance: str,
            peripherals: list[dict[str, object]],
            _vehicle_devices: list[dict[str, object]],
            _api_key: str,
        ) -> str:
            self.assertEqual(peripherals, [])
            return json.dumps(
                {
                    "summary": "attempted peripheral substitution",
                    "needs_confirmation": False,
                    "question": "",
                    "steps": [
                        {
                            "kind": "peripheral",
                            "capability_id": "peripheral_1",
                            "command": "",
                            "label": "ignored",
                            "delay_ms": 0,
                        }
                    ],
                }
            )

        with (
            patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False),
            patch.object(service, "_request_gpt_plan", side_effect=proposed_plan),
        ):
            result = service.plan(
                {
                    "utterance": "打开空调",
                    "peripherals": [
                        {
                            "id": "unmentioned-toy-on",
                            "status": "learned",
                            "device_name": "闪灯玩具",
                            "action": "打开",
                            "slot": 1,
                        }
                    ],
                    "vehicle_devices": [],
                }
            )

        self.assertEqual(result["steps"], [])

    def test_execution_rechecks_device_identity_before_hardware_emit(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        capability = {
            "id": "malicious-light-on",
            "status": "learned",
            "device_id": "malicious-light",
            "device_name": "灯",
            "card_uid": "2CC8B103",
            "action": "打开",
            "slot": 1,
        }

        with patch.object(service, "execute_gateway") as execute_gateway:
            with self.assertRaisesRegex(ExtensionIntegrationError, "明确提及"):
                service.execute_peripheral_request(
                    "打开氛围灯",
                    capability,
                    conversation_id="conv-reject-name-hijack",
                    execution_mode="hardware",
                )
            execute_gateway.assert_not_called()

    def test_main_peripheral_result_uses_gscu_without_vehicle_can(self) -> None:
        service = SelfExtensionService(SimulationGateway())

        result = service.execute_peripheral_request(
            "打开第三方氛围灯",
            {
                "id": "cap-third-party-light-on",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "开启",
                "slot": 9,
            },
            conversation_id="conv-main-extension",
            conversation_turn=2,
            execution_mode="hardware",
        )

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["orchestration_route"], "nfc_peripheral")
        self.assertEqual(result["peripheral"]["slot"], 9)
        self.assertEqual(result["execution_report"]["spec"]["mode"], "hil")
        operation = result["execution_report"]["spec"]["results"][0]
        self.assertEqual(operation["domain"], "self_extension")
        self.assertTrue(operation["telemetry"]["gateway_acknowledged"])

    def test_main_peripheral_sequence_executes_validated_actions_and_delays(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        catalog = [
            {
                "id": "red",
                "status": "learned",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "红色灯光",
                "slot": 3,
            },
            {
                "id": "blue",
                "status": "learned",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "蓝色灯光",
                "slot": 4,
            },
            {
                "id": "off",
                "status": "learned",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": "关闭灯光",
                "slot": 5,
            },
        ]
        plan_result = {
            "model": "gpt-5.6-sol",
            "summary": "红灯保持五秒后切换蓝灯",
            "steps": [
                {"kind": "peripheral", "capability_id": "red"},
                {"kind": "delay", "delay_ms": 5000},
                {"kind": "peripheral", "capability_id": "blue"},
            ],
        }

        with patch("cognitive_agent.application.self_extension.time.sleep") as sleep:
            result = service.execute_peripheral_plan_request(
                "第三方氛围灯红灯闪烁5秒后切换蓝灯",
                plan_result,
                catalog,
                conversation_id="conv-main-extension-sequence",
                execution_mode="hardware",
            )

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(
            result["orchestration_route"], "nfc_peripheral_sequence"
        )
        self.assertEqual(result["sequence"]["step_count"], 3)
        self.assertEqual(result["sequence"]["action_count"], 2)
        self.assertEqual(result["sequence"]["emit_count"], 2)
        self.assertEqual(result["sequence"]["total_delay_ms"], 5000)
        self.assertEqual(result["sequence"]["steady_blink_fallbacks"], 1)
        self.assertEqual(
            [item["status"] for item in result["operation_results"]],
            ["succeeded", "succeeded", "succeeded"],
        )
        sleep.assert_any_call(5.0)
        self.assertNotIn(0.5, [call.args[0] for call in sleep.call_args_list])

    def test_failed_first_emit_retains_complete_compiled_infrared_plan(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "device_id": "tony-light-001",
                "device_name": "闪灯玩具",
                "object_names": ["闪光灯"],
                "card_uid": "2CC8B103",
                "action": action,
                "slot": slot,
            }
            for capability_id, action, slot in (
                ("light-on", "打开", 1),
                ("light-green", "切换到绿色", 4),
            )
        ]
        plan_result = {
            "provider_used": "infrared_local_rules",
            "summary": "打开闪光灯并切换为绿色",
            "steps": [
                {"kind": "peripheral", "capability_id": "light-on"},
                {"kind": "peripheral", "capability_id": "light-green"},
            ],
        }

        with patch.object(
            service,
            "execute_gateway",
            side_effect=GatewayError("simulated infrared transport failure"),
        ) as execute_gateway:
            result = service.execute_peripheral_plan_request(
                "开启闪光灯，颜色设置为绿色",
                plan_result,
                catalog,
                conversation_id="conv-failed-first-infrared-emit",
                execution_mode="hardware",
            )

        execute_gateway.assert_called_once_with("emit", {"slot": 1})
        self.assertEqual(result["status"], "failed")
        self.assertEqual(result["sequence"]["action_count"], 2)
        self.assertEqual(result["sequence"]["emit_count"], 0)
        self.assertEqual(
            [
                item["parameters"]["slot"]
                for item in result["function_config"]["spec"]["operations"]
            ],
            [1, 4],
        )
        self.assertEqual(
            [
                item["signals"]["slot"]
                for item in result["command_plan"]["spec"]["commands"]
            ],
            [1, 4],
        )
        self.assertEqual(
            [item["status"] for item in result["operation_results"]],
            ["failed", "aborted"],
        )
        self.assertEqual(
            result["operation_results"][1]["message"],
            "前序红外指令失败，未执行",
        )
        self.assertEqual(
            result["operation_results"][1]["telemetry"]["delivery_status"],
            "not_submitted",
        )
        self.assertFalse(
            result["operation_results"][1]["telemetry"]["transport_submitted"]
        )

    def test_power_on_then_white_waits_800ms_and_continues_without_ack(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "device_id": "tony-light-001",
                "device_name": "闪灯玩具",
                "object_names": ["闪烁灯"],
                "card_uid": "2CC8B103",
                "action": action,
                "slot": slot,
            }
            for capability_id, action, slot in (
                ("light-on", "打开", 1),
                ("light-white", "切换到白色", 6),
            )
        ]

        def submit_without_ack(_action, payload):
            command = build_gateway_command("emit", payload)
            return {
                "success": True,
                "transport": "tcp",
                "request": command,
                "device_response": {
                    "success": True,
                    "delivery_status": "submitted",
                    "transport_submitted": True,
                    "gateway_acknowledged": False,
                },
                "gateway_acknowledged": False,
            }

        with (
            patch.object(
                service,
                "execute_gateway",
                side_effect=submit_without_ack,
            ) as execute_gateway,
            patch(
                "cognitive_agent.application.self_extension.time.sleep"
            ) as sleep,
        ):
            result = service.execute_peripheral_plan_request(
                "启动闪烁灯，颜色设置为白色",
                {
                    "provider_used": "infrared_local_rules",
                    "summary": "启动闪烁灯并切换为白色",
                    "steps": [
                        {"kind": "peripheral", "capability_id": "light-on"},
                        {"kind": "peripheral", "capability_id": "light-white"},
                    ],
                },
                catalog,
                conversation_id="conv-submitted-unacknowledged-infrared",
                execution_mode="hardware",
            )

        self.assertEqual(execute_gateway.call_count, 2)
        sleep.assert_called_once_with(0.8)
        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["sequence"]["emit_count"], 2)
        self.assertEqual(
            [item["telemetry"]["slot"] for item in result["operation_results"]],
            [1, 6],
        )
        self.assertEqual(
            [
                item["telemetry"]["delivery_status"]
                for item in result["operation_results"]
            ],
            ["submitted", "submitted"],
        )
        self.assertTrue(
            all(
                item["telemetry"]["transport_submitted"]
                for item in result["operation_results"]
            )
        )
        self.assertTrue(
            all(
                item["telemetry"]["gateway_acknowledged"] is False
                for item in result["operation_results"]
            )
        )

    def test_color_sequence_never_cycles_the_power_off_capability(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        catalog = [
            {
                "id": capability_id,
                "status": "learned",
                "device_id": "third-party-light",
                "device_name": "第三方氛围灯",
                "card_uid": "1A53DD95",
                "action": action,
                "slot": slot,
            }
            for capability_id, action, slot in (
                ("on", "开启灯光", 1),
                ("off", "关闭灯光", 2),
                ("red", "红色灯光", 3),
                ("green", "绿色灯光", 4),
                ("blue", "蓝色灯光", 5),
                ("white", "白色灯光", 6),
            )
        ]
        plan_result = {
            "model": "gpt-5.6-sol",
            "summary": "按顺序显示四种颜色",
            "steps": [
                {"kind": "peripheral", "capability_id": "on"},
                {"kind": "peripheral", "capability_id": "red"},
                {"kind": "delay", "delay_ms": 5000},
                {"kind": "peripheral", "capability_id": "blue"},
                {"kind": "delay", "delay_ms": 5000},
                {"kind": "peripheral", "capability_id": "green"},
                {"kind": "delay", "delay_ms": 5000},
                {"kind": "peripheral", "capability_id": "white"},
                {"kind": "delay", "delay_ms": 5000},
                {"kind": "peripheral", "capability_id": "off"},
            ],
        }

        with patch("cognitive_agent.application.self_extension.time.sleep") as sleep:
            result = service.execute_peripheral_plan_request(
                (
                    "打开第三方氛围灯，红灯闪烁5s后，蓝光闪烁5s后，"
                    "绿灯闪烁5s后，白色闪烁5s，最后关闭第三方氛围灯"
                ),
                plan_result,
                catalog,
                conversation_id="conv-color-sequence-no-power-cycle",
                execution_mode="hardware",
            )

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["sequence"]["emit_count"], 6)
        self.assertEqual(result["sequence"]["steady_blink_fallbacks"], 4)
        self.assertEqual(result["sequence"]["total_delay_ms"], 20_000)
        self.assertEqual(
            [
                operation["telemetry"]["slot"]
                for operation in result["operation_results"]
                if operation["telemetry"].get("slot")
            ],
            [1, 3, 5, 4, 6, 2],
        )
        self.assertEqual(
            [call.args[0] for call in sleep.call_args_list].count(5.0),
            4,
        )
        self.assertNotIn(0.5, [call.args[0] for call in sleep.call_args_list])

    def test_plan_only_accepts_learned_catalog_capabilities(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        proposed = {
            "summary": "先开玩具灯，再打开阅读灯",
            "needs_confirmation": False,
            "question": "",
            "steps": [
                {
                    "kind": "peripheral",
                    "capability_id": "cap-light-on",
                    "command": "",
                    "label": "ignored",
                },
                {
                    "kind": "peripheral",
                    "capability_id": "invented-capability",
                    "command": "",
                    "label": "ignored",
                },
                {
                    "kind": "vehicle",
                    "capability_id": "",
                    "command": "light reading all on",
                    "label": "车内灯光 · 阅读灯",
                },
            ],
        }
        with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
            with patch.object(
                service,
                "_request_gpt_plan",
                return_value=json.dumps(proposed, ensure_ascii=False),
            ):
                result = service.plan(
                    {
                        "utterance": "打开玩具灯和阅读灯",
                        "peripherals": [
                            {
                                "id": "cap-light-on",
                                "status": "learned",
                                "device_name": "灯光测试玩具",
                                "object_names": ["玩具灯"],
                                "action": "开启",
                                "risk": "low",
                            }
                        ],
                        "vehicle_devices": [
                            {
                                "name": "车内灯光",
                                "actions": [
                                    {
                                        "name": "阅读灯",
                                        "command": "light reading all on",
                                    }
                                ],
                            }
                        ],
                    }
                )

        self.assertEqual(len(result["steps"]), 2)
        self.assertEqual(result["steps"][0]["capability_id"], "cap-light-on")
        self.assertEqual(result["steps"][1]["command"], "light reading all on")

    def test_plan_maps_short_planner_ids_back_to_opaque_capability_ids(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        opaque_id = "cap-b0ab02a2-3640-4468-8cdc-3bb7164c7078"

        def planned_response(
            _utterance: str,
            peripherals: list[dict[str, object]],
            _vehicle_devices: list[dict[str, object]],
            _api_key: str,
        ) -> str:
            self.assertEqual(peripherals[0]["id"], "peripheral_1")
            self.assertNotIn(opaque_id, [item["id"] for item in peripherals])
            self.assertNotIn("slot", peripherals[0])
            self.assertNotIn("card_uid", peripherals[0])
            self.assertNotIn("transport", peripherals[0])
            return json.dumps(
                {
                    "summary": "切换红灯",
                    "needs_confirmation": False,
                    "question": "",
                    "steps": [
                        {
                            "kind": "peripheral",
                            "capability_id": "peripheral_1",
                            "command": "红色灯光",
                            "label": "第三方氛围灯 · 红色灯光",
                            "delay_ms": 0,
                        }
                    ],
                },
                ensure_ascii=False,
            )

        with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
            with patch.object(
                service,
                "_request_gpt_plan",
                side_effect=planned_response,
            ):
                result = service.plan(
                    {
                        "utterance": "第三方氛围灯切换红灯",
                        "peripherals": [
                            {
                                "id": opaque_id,
                                "status": "learned",
                                "device_name": "第三方氛围灯",
                                "action": "红色灯光",
                                "risk": "low",
                            }
                        ],
                        "vehicle_devices": [],
                    }
                )

        self.assertEqual(result["steps"][0]["capability_id"], opaque_id)

    def test_plan_recovers_full_color_sequence_when_gpt_returns_no_steps(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        actions = (
            ("on", "开启灯光", 1),
            ("off", "关闭灯光", 2),
            ("red", "红色灯光", 3),
            ("green", "绿色灯光", 4),
            ("blue", "蓝色灯光", 5),
            ("white", "白色灯光", 6),
        )
        proposed = {
            "summary": "模型未能生成动作",
            "needs_confirmation": False,
            "question": "",
            "steps": [],
        }
        with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
            with patch.object(
                service,
                "_request_gpt_plan",
                return_value=json.dumps(proposed, ensure_ascii=False),
            ):
                result = service.plan(
                    {
                        "utterance": (
                            "打开第三方氛围灯，红灯闪烁5s后，蓝光闪烁5s后，"
                            "绿灯闪烁5s后，白色闪烁5s，最后关闭第三方氛围灯"
                        ),
                        "peripherals": [
                            {
                                "id": capability_id,
                                "status": "learned",
                                "device_id": "third-party-light",
                                "device_name": "第三方氛围灯",
                                "action": action,
                                "aliases": [action],
                                "slot": slot,
                                "risk": "low",
                            }
                            for capability_id, action, slot in actions
                        ],
                        "vehicle_devices": [],
                    }
                )

        self.assertEqual(
            [
                step.get("capability_id")
                for step in result["steps"]
                if step["kind"] == "peripheral"
            ],
            ["on", "red", "blue", "green", "white", "off"],
        )
        self.assertEqual(
            [
                step.get("delay_ms")
                for step in result["steps"]
                if step["kind"] == "delay"
            ],
            [5000, 5000, 5000, 5000],
        )

    def test_plan_rejects_vehicle_command_outside_current_catalog(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        proposed = {
            "summary": "unsafe target",
            "needs_confirmation": False,
            "question": "",
            "steps": [
                {
                    "kind": "vehicle",
                    "capability_id": "",
                    "command": "seat passenger vent high",
                    "label": "wrong seat",
                    "delay_ms": 0,
                }
            ],
        }
        with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
            with patch.object(
                service,
                "_request_gpt_plan",
                return_value=json.dumps(proposed),
            ):
                with self.assertRaisesRegex(ExtensionIntegrationError, "双通道"):
                    service.plan(
                        {
                            "utterance": "打开座椅通风",
                            "peripherals": [],
                            "vehicle_devices": [
                                {
                                    "name": "主驾座椅",
                                    "actions": [
                                        {
                                            "name": "通风高档",
                                            "command": "seat driver vent high",
                                        }
                                    ],
                                }
                            ],
                        }
                    )

    def test_gpt_plan_honors_streaming_responses_configuration(self) -> None:
        proposed = json.dumps(
            {
                "summary": "关闭空调",
                "needs_confirmation": False,
                "question": "",
                "steps": [
                    {
                        "kind": "vehicle",
                        "capability_id": "",
                        "command": "ac off",
                        "label": "关闭空调",
                    }
                ],
            },
            ensure_ascii=False,
        )
        stream_body = "\n".join(
            (
                "event: response.output_text.delta",
                "data: "
                + json.dumps(
                    {
                        "type": "response.output_text.delta",
                        "delta": proposed[:20],
                    },
                    ensure_ascii=False,
                ),
                "event: response.output_text.done",
                "data: "
                + json.dumps(
                    {"type": "response.output_text.done", "text": proposed},
                    ensure_ascii=False,
                ),
                "data: [DONE]",
            )
        )
        response = MagicMock()
        response.__enter__.return_value = response
        response.read.return_value = stream_body.encode("utf-8")

        with patch.dict(
            os.environ,
            {
                "OPENAI_STREAM": "true",
                "IR_OPENAI_MODEL": "gpt-5.6-sol",
            },
            clear=False,
        ):
            service = SelfExtensionService(SimulationGateway())
        with patch(
            "cognitive_agent.application.self_extension.urlopen",
            return_value=response,
        ) as open_request:
            output = service._request_gpt_plan(
                "关闭空调",
                [],
                [
                    {
                        "name": "空调",
                        "actions": [{"name": "关闭空调", "command": "ac off"}],
                    }
                ],
                "test-key",
            )

        self.assertEqual(json.loads(output)["steps"][0]["command"], "ac off")
        request_payload = json.loads(
            open_request.call_args.args[0].data.decode("utf-8")
        )
        self.assertIs(request_payload["stream"], True)
        self.assertEqual(request_payload["model"], "gpt-5.6-sol")
        command_schema = request_payload["text"]["format"]["schema"]["properties"][
            "steps"
        ]["items"]["properties"]["command"]
        self.assertEqual(command_schema["type"], "string")
        self.assertEqual(command_schema["maxLength"], 300)
        capability_schema = request_payload["text"]["format"]["schema"]["properties"][
            "steps"
        ]["items"]["properties"]["capability_id"]
        self.assertEqual(capability_schema["enum"], [""])

    def test_plan_accepts_bounded_passenger_position_for_declared_ch0_executor(
        self,
    ) -> None:
        service = SelfExtensionService(SimulationGateway())
        proposed = {
            "summary": "调整副驾",
            "needs_confirmation": False,
            "question": "",
            "steps": [
                {
                    "kind": "vehicle",
                    "capability_id": "",
                    "command": "副驾水平调到63%",
                    "label": "副驾水平位置 63%",
                    "delay_ms": 0,
                }
            ],
        }
        payload = {
            "utterance": "副驾水平位置调到63%",
            "peripherals": [],
            "vehicle_devices": [
                {
                    "id": "passenger-seat",
                    "name": "副驾座椅",
                    "channel": "CH0",
                    "actions": [
                        {"name": "副驾水平 50%", "command": "副驾水平调到50%"}
                    ],
                }
            ],
        }
        with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
            with patch.object(
                service,
                "_request_gpt_plan",
                return_value=json.dumps(proposed, ensure_ascii=False),
            ):
                result = service.plan(payload)
        self.assertEqual(result["steps"][0]["command"], "副驾水平调到63%")

        proposed["steps"][0]["command"] = "副驾水平调到101%"
        with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
            with patch.object(
                service,
                "_request_gpt_plan",
                return_value=json.dumps(proposed, ensure_ascii=False),
            ):
                with self.assertRaisesRegex(ExtensionIntegrationError, "双通道"):
                    service.plan(payload)

    def test_vehicle_execution_reuses_the_same_orchestration_service(self) -> None:
        with patch.dict(
            os.environ,
            {"COGNITIVE_AGENT_EXECUTION_MODE": "dry_run"},
            clear=False,
        ):
            service = SelfExtensionService(SimulationGateway())
        vehicle_service = Mock()
        vehicle_service.process.return_value = {
            "status": "succeeded",
            "provider_used": "local_rules",
            "request_id": "req-extension-1",
            "conversation_id": "conv-extension-shared",
            "conversation_turn": 3,
            "decision_summary": ["打开阅读灯"],
            "validation": [],
            "execution_report": {"spec": {"results": [{"status": "succeeded"}]}},
            "deferred_requests": [],
        }

        result = service.execute_vehicle(
            {
                "command": "打开全部阅读灯",
                "conversation_id": "conv-extension-shared",
                "vehicle_id": "denza_n7",
            },
            vehicle_service,
            hardware_access_bypass=True,
        )

        self.assertTrue(result["success"])
        self.assertEqual(result["request_id"], "req-extension-1")
        self.assertEqual(result["conversation_id"], "conv-extension-shared")
        self.assertEqual(result["conversation_turn"], 3)
        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(
            result["execution_report"], vehicle_service.process.return_value["execution_report"]
        )
        self.assertEqual(result["operation_results"], [{"status": "succeeded"}])
        self.assertFalse(result["physical_feedback_required"])
        self.assertFalse(result["physical_feedback_confirmed"])
        vehicle_service.process.assert_called_once()
        self.assertEqual(vehicle_service.process.call_args.args[0], "打开全部阅读灯")
        self.assertEqual(
            vehicle_service.process.call_args.args[2], "conv-extension-shared"
        )
        self.assertTrue(vehicle_service.process.call_args.kwargs["hardware_access_bypass"])

    def test_vehicle_execution_only_marks_exact_complete_status_as_success(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        for status in (
            "validation_failed",
            "needs_confirmation",
            "capability_unavailable",
            "partially_succeeded",
        ):
            with self.subTest(status=status):
                vehicle_service = Mock()
                vehicle_service.process.return_value = {
                    "status": status,
                    "provider_used": "local_rules",
                    "request_id": f"req-{status}",
                    "decision_summary": [],
                    "validation": [],
                    "execution_report": {
                        "spec": {"results": [{"status": "succeeded"}]}
                    },
                    "deferred_requests": [],
                }

                result = service.execute_vehicle(
                    {"command": "打开全部阅读灯"},
                    vehicle_service,
                    hardware_access_bypass=True,
                )

                self.assertFalse(result["success"])
                self.assertEqual(result["status"], status)

    def test_vehicle_execution_rejects_failed_or_missing_operation_results(self) -> None:
        service = SelfExtensionService(SimulationGateway())
        for operation_results in (
            [],
            [{"status": "succeeded"}, {"status": "failed", "message": "timeout"}],
        ):
            with self.subTest(operation_results=operation_results):
                vehicle_service = Mock()
                vehicle_service.process.return_value = {
                    "status": "succeeded",
                    "execution_report": {"spec": {"results": operation_results}},
                    "domain_state": {"roof": {"shade_position_pct": 31}},
                    "actuator_state": {"driver": {"heating": {"state": "off"}}},
                }

                result = service.execute_vehicle(
                    {"command": "打开遮阳帘"},
                    vehicle_service,
                    hardware_access_bypass=True,
                )

                self.assertFalse(result["success"])
                self.assertEqual(
                    result["domain_state"], {"roof": {"shade_position_pct": 31}}
                )
                self.assertEqual(
                    result["actuator_state"],
                    {"driver": {"heating": {"state": "off"}}},
                )

    def test_hardware_vehicle_execution_requires_authoritative_feedback_per_item(
        self,
    ) -> None:
        with patch.dict(
            os.environ,
            {"COGNITIVE_AGENT_EXECUTION_MODE": "hardware"},
            clear=False,
        ):
            service = SelfExtensionService(SimulationGateway())
        vehicle_service = Mock()
        vehicle_service.process.return_value = {
            "status": "succeeded",
            "provider_used": "local_rules",
            "request_id": "req-extension-hardware",
            "execution_report": {
                "spec": {
                    "status": "succeeded",
                    "physical_state": {
                        "source": "physical_can_ch1",
                        "complete": True,
                    },
                    "results": [
                        {
                            "ref": "operation-1",
                            "domain": "climate",
                            "status": "succeeded",
                            "telemetry": {
                                "channel": 1,
                                "simulated": False,
                                "feedback_required": True,
                                "feedback_observed": True,
                                "feedback_match": True,
                                "feedback_frame": {
                                    "frame_id": "0x404",
                                    "channel": 1,
                                    "payload_size": 8,
                                    "data_hex": "00 02 00 00 18 18 00 00",
                                },
                                "feedback_state": {
                                    "source": "physical_can_ch1",
                                    "channel": 1,
                                    "frame_id": "0x404",
                                    "feedback_frame_id": "0x404",
                                    "valid": True,
                                    "domain_state": {"climate": {"power": "on"}},
                                },
                            },
                        },
                        {
                            "ref": "operation-2",
                            "domain": "seat",
                            "status": "succeeded",
                            "telemetry": {
                                "channel": 1,
                                "simulated": False,
                                "feedback_required": True,
                                "feedback_observed": True,
                                "feedback_match": True,
                                "feedback_frame": {
                                    "frame_id": "0x4FA",
                                    "channel": 1,
                                    "payload_size": 8,
                                    "data_hex": "02 00 00 00 00 00 00 00",
                                },
                                "feedback_state": {
                                    "source": "physical_can_ch1",
                                    "channel": 1,
                                    "frame_id": "0x4FA",
                                    "feedback_frame_id": "0x4FA",
                                    "valid": True,
                                    "actuator_state": {
                                        "driver": {
                                            "ventilation": {
                                                "state": "on",
                                                "level": 1,
                                            }
                                        }
                                    },
                                },
                            },
                        },
                    ],
                }
            },
            "domain_state": {"climate": {"power": "on"}},
            "actuator_state": {"driver": {"ventilation": {"state": "on"}}},
        }

        confirmed = service.execute_vehicle(
            {"command": "打开空调"},
            vehicle_service,
            hardware_access_bypass=True,
        )

        self.assertTrue(confirmed["success"])
        self.assertTrue(confirmed["physical_feedback_required"])
        self.assertTrue(confirmed["physical_feedback_confirmed"])
        self.assertEqual(confirmed["physical_state"]["source"], "physical_can_ch1")
        self.assertEqual(len(confirmed["physical_feedback"]), 2)
        self.assertTrue(all(item["confirmed"] for item in confirmed["physical_feedback"]))
        self.assertTrue(all(item["required"] for item in confirmed["physical_feedback"]))
        self.assertTrue(all(item["observed"] for item in confirmed["physical_feedback"]))
        self.assertTrue(all(item["match"] is True for item in confirmed["physical_feedback"]))
        self.assertTrue(all(item["channel"] == 1 for item in confirmed["physical_feedback"]))
        self.assertEqual(
            confirmed["physical_feedback"][0]["state"]["source"],
            "physical_can_ch1",
        )

        operation = vehicle_service.process.return_value["execution_report"]["spec"][
            "results"
        ][1]
        valid_telemetry = deepcopy(operation["telemetry"])
        operation["telemetry"] = {**valid_telemetry, "feedback_match": None}
        unconfirmed = service.execute_vehicle(
            {"command": "打开空调"},
            vehicle_service,
            hardware_access_bypass=True,
        )

        self.assertFalse(unconfirmed["success"])
        self.assertFalse(unconfirmed["physical_feedback_confirmed"])
        self.assertIn("未逐项获得权威物理反馈", unconfirmed["message"])

        invalid_cases: list[tuple[str, dict]] = []
        for field, value in (
            ("channel", 0),
            ("simulated", True),
            ("feedback_required", False),
            ("feedback_observed", False),
            ("feedback_match", False),
            ("feedback_frame", None),
            ("feedback_state", None),
        ):
            invalid_cases.append((field, {**valid_telemetry, field: value}))

        wrong_frame_channel = deepcopy(valid_telemetry)
        wrong_frame_channel["feedback_frame"]["channel"] = 0
        invalid_cases.append(("feedback_frame.channel", wrong_frame_channel))
        missing_frame_payload = deepcopy(valid_telemetry)
        missing_frame_payload["feedback_frame"].pop("data_hex")
        invalid_cases.append(("feedback_frame.data_hex", missing_frame_payload))
        incomplete_frame_payload = deepcopy(valid_telemetry)
        incomplete_frame_payload["feedback_frame"]["payload_size"] = 7
        invalid_cases.append(
            ("feedback_frame.payload_size", incomplete_frame_payload)
        )
        wrong_source = deepcopy(valid_telemetry)
        wrong_source["feedback_state"]["source"] = "physical_can_ch0"
        invalid_cases.append(("feedback_state.source", wrong_source))
        missing_decoded_state = deepcopy(valid_telemetry)
        missing_decoded_state["feedback_state"].pop("actuator_state")
        invalid_cases.append(("feedback_state.actuator_state", missing_decoded_state))
        mismatched_frame = deepcopy(valid_telemetry)
        mismatched_frame["feedback_state"]["frame_id"] = "0x404"
        invalid_cases.append(("feedback_state.frame_id", mismatched_frame))
        invalid_cases.append(
            (
                "legacy_ch0",
                {
                    "channel": 0,
                    "simulated": False,
                    "hardware_feedback": True,
                    "feedback_frame": {"frame_id": "0x302"},
                },
            )
        )

        for label, insufficient_feedback in invalid_cases:
            with self.subTest(insufficient_feedback=label):
                operation["telemetry"] = insufficient_feedback
                rejected = service.execute_vehicle(
                    {"command": "打开空调"},
                    vehicle_service,
                    hardware_access_bypass=True,
                )
                self.assertFalse(rejected["success"])
                self.assertFalse(rejected["physical_feedback_confirmed"])

    def test_ch0_passenger_feedback_requires_shared_runtime_physical_snapshot(
        self,
    ) -> None:
        operation = {
            "ref": "passenger-longitudinal-63",
            "domain": "seat",
            "status": "succeeded",
            "telemetry": {
                "transport": "shared_can_runtime",
                "channel": 0,
                "simulated": False,
                "hardware_feedback": True,
                "feedback_required": True,
                "feedback_observed": True,
                "feedback_match": True,
                "physical_state": {
                    "source": "physical_can",
                    "channel": 0,
                    "target": {
                        "device": "seat",
                        "zone": "front_right",
                        "instance": "passenger",
                    },
                    "position": {
                        "longitudinal": {
                            "position_pct": 63,
                            "position_valid": True,
                        }
                    },
                },
            },
        }
        verified = SelfExtensionService._authoritative_vehicle_feedback_state(
            operation
        )
        self.assertIsNotNone(verified)
        self.assertEqual(verified["verification"]["channel"], 0)

        operation["telemetry"]["transport"] = "untrusted_bridge"
        self.assertIsNone(
            SelfExtensionService._authoritative_vehicle_feedback_state(operation)
        )

    def test_ch0_bridge_report_exposes_verified_passenger_snapshot(self) -> None:
        function_config = {
            "metadata": {"request_id": "req-ch0-extension"},
            "spec": {
                "operations": [
                    {
                        "id": "passenger-longitudinal-63",
                        "capability": "seat.set_longitudinal_position",
                    }
                ]
            },
        }
        bridge_result = {
            "result": "PASS",
            "details": {
                "summary": "物理反馈达到目标",
                "motors": [
                    {
                        "position": 6300 if index == 0 else 5000,
                        "position_valid": True,
                        "active": False,
                    }
                    for index in range(4)
                ],
                "thermal": [0x01],
            },
        }
        report = ZcanBridgeSeatExecutor._report(
            function_config,
            "2026-08-11T00:00:00Z",
            bridge_result,
            None,
        )
        physical_state = report["spec"]["physical_state"]
        self.assertEqual(physical_state["source"], "physical_can")
        self.assertEqual(physical_state["channel"], 0)
        self.assertEqual(
            physical_state["position"]["longitudinal"]["position_pct"], 63
        )
        verified = SelfExtensionService._authoritative_vehicle_feedback_state(
            report["spec"]["results"][0]
        )
        self.assertIsNotNone(verified)

    def test_vehicle_execution_ignores_browser_routing_overrides(self) -> None:
        with patch.dict(
            os.environ,
            {
                "COGNITIVE_AGENT_EXECUTION_MODE": "dry_run",
                "COGNITIVE_AGENT_PROVIDER": "local_rules",
                "COGNITIVE_AGENT_VEHICLE_ID": "denza_n7",
            },
            clear=False,
        ):
            service = SelfExtensionService(SimulationGateway())
        vehicle_service = Mock()
        vehicle_service.process.return_value = {
            "status": "succeeded",
            "execution_report": {"spec": {"results": [{"status": "succeeded"}]}},
        }

        service.execute_vehicle(
            {
                "command": "seat driver vent low",
                "execution_mode": "hardware",
                "provider": "browser-selected-provider",
                "vehicle_id": "t2g_primary_seat",
                "simulation_fault": "adapter_disconnect",
                "channel": "0",
            },
            vehicle_service,
            hardware_access_bypass=True,
        )

        args = vehicle_service.process.call_args.args
        kwargs = vehicle_service.process.call_args.kwargs
        self.assertEqual(args[1], "local_rules")
        self.assertEqual(args[3], "dry_run")
        self.assertEqual(kwargs["vehicle_id"], "denza_n7")
        self.assertEqual(kwargs["simulation_fault"], "normal")

    def test_tcp_gateway_recovers_from_unavailable_address_and_tracks_peers(self) -> None:
        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        probe.bind(("127.0.0.1", 0))
        port = int(probe.getsockname()[1])
        probe.close()

        class UnavailableAddressSocket:
            def setsockopt(self, *args: object) -> None:
                del args

            def bind(self, address: object) -> None:
                del address
                raise OSError(10049, "address is not available")

            def close(self) -> None:
                return

        socket_attempts = 0

        def new_server_socket() -> object:
            nonlocal socket_attempts
            socket_attempts += 1
            if socket_attempts == 1:
                return UnavailableAddressSocket()
            return socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        clients: list[socket.socket] = []
        with patch.dict(
            os.environ,
            {
                "GSCU_TCP_LISTEN_HOST": "127.0.0.1",
                "GSCU_TCP_PORT": str(port),
                "GSCU_TCP_BIND_RETRY_SECONDS": "0.05",
            },
        ):
            gateway = TcpGateway()
        with patch.object(gateway, "_new_server_socket", side_effect=new_server_socket):
            gateway.start()
            try:
                self.assertTrue(
                    self._wait_until(lambda: gateway.status()["listening"]),
                    gateway.status(),
                )
                self.assertGreaterEqual(socket_attempts, 2)
                waiting = gateway.status()
                self.assertFalse(waiting["connected"])
                self.assertIn("等待设备接入", waiting["message"])
                waiting_health = SelfExtensionService(gateway).health()
                self.assertTrue(waiting_health["listening"])
                self.assertFalse(waiting_health["peer_connected"])

                first = socket.create_connection(("127.0.0.1", port), timeout=1)
                clients.append(first)
                self.assertTrue(
                    self._wait_until(lambda: gateway.status()["connected"]),
                    gateway.status(),
                )
                first_peer = gateway.status()["peer"]
                self.assertTrue(str(first_peer).startswith("127.0.0.1:"))
                connected_health = SelfExtensionService(gateway).health()
                self.assertTrue(connected_health["connected"])
                self.assertEqual(connected_health["peer"], first_peer)

                # Replacing an old peer must not let its reader mark the new
                # GSCU1B session offline when the old socket closes.
                second = socket.create_connection(("127.0.0.1", port), timeout=1)
                clients.append(second)
                self.assertTrue(
                    self._wait_until(
                        lambda: gateway.status()["peer"] not in {None, first_peer}
                    ),
                    gateway.status(),
                )
                time.sleep(0.1)
                self.assertTrue(gateway.status()["connected"])

                second.close()
                clients.remove(second)
                self.assertTrue(
                    self._wait_until(lambda: not gateway.status()["connected"]),
                    gateway.status(),
                )
                self.assertTrue(gateway.status()["listening"])

                third = socket.create_connection(("127.0.0.1", port), timeout=1)
                clients.append(third)
                self.assertTrue(
                    self._wait_until(lambda: gateway.status()["connected"]),
                    gateway.status(),
                )
            finally:
                for client in clients:
                    client.close()
                gateway.close()

        stopped = gateway.status()
        self.assertFalse(stopped["listening"])
        self.assertFalse(stopped["connected"])

    def test_tcp_gateway_recovers_after_port_conflict_is_released(self) -> None:
        blocker = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        blocker.bind(("127.0.0.1", 0))
        blocker.listen(1)
        port = int(blocker.getsockname()[1])
        with patch.dict(
            os.environ,
            {
                "GSCU_TCP_LISTEN_HOST": "127.0.0.1",
                "GSCU_TCP_PORT": str(port),
                "GSCU_TCP_BIND_RETRY_SECONDS": "0.05",
            },
        ):
            gateway = TcpGateway()
        gateway.start()
        try:
            self.assertTrue(
                self._wait_until(lambda: "监听失败" in gateway.status().get("message", "")),
                gateway.status(),
            )
            self.assertFalse(gateway.status()["listening"])
            blocker.close()
            self.assertTrue(
                self._wait_until(lambda: gateway.status()["listening"]),
                gateway.status(),
            )
            self.assertNotIn("失败", gateway.status()["message"])
        finally:
            blocker.close()
            gateway.close()

    def test_tcp_gateway_can_restart_immediately_after_close(self) -> None:
        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        probe.bind(("127.0.0.1", 0))
        port = int(probe.getsockname()[1])
        probe.close()
        with patch.dict(
            os.environ,
            {
                "GSCU_TCP_LISTEN_HOST": "127.0.0.1",
                "GSCU_TCP_PORT": str(port),
                "GSCU_TCP_BIND_RETRY_SECONDS": "0.05",
            },
        ):
            gateway = TcpGateway()

        try:
            gateway.start()
            self.assertTrue(self._wait_until(lambda: gateway.status()["listening"]))
            gateway.close()
            gateway.start()
            self.assertTrue(
                self._wait_until(lambda: gateway.status()["listening"]),
                gateway.status(),
            )
        finally:
            gateway.close()

    def test_tcp_gateway_rejects_peer_accepted_during_close(self) -> None:
        settimeout_entered = threading.Event()
        allow_settimeout_to_finish = threading.Event()
        accepted_socket_closed = threading.Event()
        unexpected_reader_started = threading.Event()
        allow_unexpected_reader_to_finish = threading.Event()

        class AcceptedSocket:
            def settimeout(self, timeout: float) -> None:
                del timeout
                settimeout_entered.set()
                allow_settimeout_to_finish.wait(timeout=2)

            def setsockopt(self, *args: object) -> None:
                del args

            def recv(self, size: int) -> bytes:
                del size
                unexpected_reader_started.set()
                allow_unexpected_reader_to_finish.wait(timeout=2)
                return b""

            def close(self) -> None:
                accepted_socket_closed.set()

        accepted_socket = AcceptedSocket()

        class ServerSocket:
            def setsockopt(self, *args: object) -> None:
                del args

            def bind(self, address: object) -> None:
                del address

            def listen(self, backlog: int) -> None:
                del backlog

            def settimeout(self, timeout: float) -> None:
                del timeout

            def accept(self) -> tuple[AcceptedSocket, tuple[str, int]]:
                return accepted_socket, ("127.0.0.1", 4444)

            def close(self) -> None:
                return

        with patch.dict(
            os.environ,
            {
                "GSCU_TCP_LISTEN_HOST": "127.0.0.1",
                "GSCU_TCP_PORT": "4444",
                "GSCU_TCP_BIND_RETRY_SECONDS": "0.05",
            },
        ):
            gateway = TcpGateway()
        with patch.object(gateway, "_new_server_socket", return_value=ServerSocket()):
            gateway.start()
            self.assertTrue(settimeout_entered.wait(timeout=1))
            close_thread = threading.Thread(target=gateway.close)
            close_thread.start()
            try:
                self.assertTrue(self._wait_until(gateway._stop.is_set))
            finally:
                allow_settimeout_to_finish.set()
                close_thread.join(timeout=3)

        try:
            self.assertFalse(close_thread.is_alive())
            self.assertTrue(accepted_socket_closed.is_set())
            self.assertFalse(unexpected_reader_started.is_set())
            self.assertFalse(gateway.status()["connected"])
            self.assertIsNone(gateway.status()["peer"])
        finally:
            allow_unexpected_reader_to_finish.set()


if __name__ == "__main__":
    unittest.main()