Cognitive_Agent/tests/e2e/test_seat_web_api.py

from __future__ import annotations

import json
import os
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from http.cookiejar import CookieJar
from http.server import ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from cognitive_agent.application.codex_assistant import CodexAssistantService
from cognitive_agent.application.mode_profile import ModeProfileRegistry
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
from cognitive_agent.application.self_extension import (
    GatewayError,
    SelfExtensionService,
    SimulationGateway,
    partition_active_peripheral_intent,
)
from cognitive_agent.auth import AuthConfig, AuthRuntime, hash_password
from cognitive_agent.infrastructure.can.access import HardwareAccessController
from cognitive_agent.infrastructure.can.gateway import CanGatewayError
from cognitive_agent.infrastructure.llm.openai_responses import OpenAIIntentError
from cognitive_agent.infrastructure.speech.understanding import (
    UnderstandingTranscriptProcessor,
)
from cognitive_agent.server import SeatDemoHandler


class SeatWebApiTests(unittest.TestCase):
    username = "test-user"
    password = "public-demo-test-password"

    @classmethod
    def setUpClass(cls) -> None:
        cls.temporary_directory = tempfile.TemporaryDirectory()
        cls.extension_temporary_directory = tempfile.TemporaryDirectory()
        # Web regression tests must never inherit a developer's live bench
        # enablement, especially when local debug mode intentionally skips the
        # per-request approval token.
        with patch.dict(
            os.environ,
            {
                "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
                "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
                "COGNITIVE_AGENT_WIRED_PASSENGER_CAN_ENABLED": "false",
            },
            clear=False,
        ):
            service = SeatDemoService(REPOSITORY_ROOT)
        service.artifact_root = Path(cls.temporary_directory.name)
        cls.service = service
        cls.original_service = SeatDemoHandler.service
        SeatDemoHandler.service = service
        cls.original_extension = SeatDemoHandler.extension
        SeatDemoHandler.extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                cls.extension_temporary_directory.name, "gscu-ir-catalog.sqlite3"
            ),
        )
        cls.original_auth = SeatDemoHandler.auth
        SeatDemoHandler.auth = AuthRuntime(
            AuthConfig(
                enabled=True,
                username=cls.username,
                password_hash=hash_password(
                    cls.password,
                    salt=b"0123456789abcdef",
                    iterations=100_000,
                ),
                session_secret=b"t" * 64,
                session_ttl_seconds=3600,
                login_attempts=5,
                login_window_seconds=300,
                requests_per_minute=120,
                trust_proxy_headers=False,
            )
        )
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), SeatDemoHandler)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()
        cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def tearDownClass(cls) -> None:
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=2)
        SeatDemoHandler.service = cls.original_service
        SeatDemoHandler.extension = cls.original_extension
        SeatDemoHandler.auth = cls.original_auth
        cls.extension_temporary_directory.cleanup()
        cls.temporary_directory.cleanup()

    def setUp(self) -> None:
        self.original_hardware_access = self.service.hardware_access
        self.service.hardware_access = HardwareAccessController()
        self.cookie_jar = CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cookie_jar)
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/auth/login",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {"username": self.username, "password": self.password}
            ).encode("utf-8"),
        )
        with self.opener.open(request, timeout=3) as response:
            payload = json.loads(response.read().decode("utf-8"))
        self.assertTrue(payload["authenticated"])

    def tearDown(self) -> None:
        self.service.hardware_access = self.original_hardware_access

    def test_authentication_protects_console_and_api(self) -> None:
        unauthenticated = urllib.request.build_opener()
        with unauthenticated.open(f"{self.base_url}/", timeout=3) as response:
            login_html = response.read().decode("utf-8")
            final_url = response.geturl()

        self.assertIn("/login", final_url)
        self.assertIn("体验账号登录", login_html)
        with unauthenticated.open(f"{self.base_url}/network.js", timeout=3) as response:
            network_script = response.read().decode("utf-8")
        self.assertIn("CognitiveAgentNetwork", network_script)
        with self.assertRaises(urllib.error.HTTPError) as caught:
            unauthenticated.open(f"{self.base_url}/api/runtime", timeout=3)
        self.assertEqual(caught.exception.code, 401)

    def test_n7_mode_scene_catalog_is_public_and_read_only(self) -> None:
        unauthenticated = urllib.request.build_opener()

        with unauthenticated.open(f"{self.base_url}/n7/scenes", timeout=3) as response:
            catalog = json.loads(response.read().decode("utf-8"))

        self.assertEqual(response.status, 200)
        self.assertEqual(response.headers["Cache-Control"], "no-store")
        self.assertEqual(response.headers["Access-Control-Allow-Origin"], "*")
        self.assertEqual(catalog["service"], "n7-scenes")
        self.assertEqual(catalog["schema_version"], 1)
        self.assertEqual(catalog["vehicle_id"], "denza_n7")
        self.assertGreaterEqual(catalog["scene_count"], 13)
        self.assertEqual(catalog["scene_count"], len(catalog["scenes"]))
        office = next(scene for scene in catalog["scenes"] if scene["id"] == "office")
        self.assertIn(office["origin"], {"preset", "customized"})
        self.assertTrue(office["actions"])
        self.assertIn("capability", office["actions"][0])
        self.assertNotIn("command_flow", json.dumps(catalog, ensure_ascii=False))

        preflight = urllib.request.Request(
            f"{self.base_url}/n7/scenes",
            method="OPTIONS",
        )
        with unauthenticated.open(preflight, timeout=3) as response:
            self.assertEqual(response.status, 204)
            self.assertEqual(response.headers["Access-Control-Allow-Origin"], "*")
            self.assertIn("GET", response.headers["Access-Control-Allow-Methods"])

    def test_user_mode_delete_endpoint_removes_only_user_profiles(self) -> None:
        original_registry = self.service.mode_profiles
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            preset_path = root / "common" / "preset" / "office.yaml"
            preset_path.parent.mkdir(parents=True)
            preset_path.write_text(
                """
id: office
name: 办公模式
operations:
- domain: climate
  capability: climate.set_temperature
  target_scope: vehicle
  parameters:
    temperature_c: 24
""".strip()
                + "\n",
                encoding="utf-8",
            )
            registry = ModeProfileRegistry(root)
            user_mode = registry.save_mode(
                "K歌模式",
                [
                    {
                        "domain": "ambient_light",
                        "capability": "ambient_light.set_power",
                        "target_scope": "vehicle",
                        "parameters": {"state": "on"},
                    }
                ],
                vehicle_id="denza_n7",
            )
            self.service.mode_profiles = registry
            try:
                request = urllib.request.Request(
                    f"{self.base_url}/api/mode-profiles/{user_mode.id}?vehicle_id=denza_n7",
                    method="DELETE",
                )
                with self.opener.open(request, timeout=3) as response:
                    payload = json.loads(response.read().decode("utf-8"))
                self.assertEqual(response.status, 200)
                self.assertEqual(payload["deleted_mode_profile"]["name"], "K歌模式")
                self.assertFalse(Path(user_mode.source).exists())

                protected = urllib.request.Request(
                    f"{self.base_url}/api/mode-profiles/office?vehicle_id=denza_n7",
                    method="DELETE",
                )
                with self.assertRaises(urllib.error.HTTPError) as caught:
                    self.opener.open(protected, timeout=3)
                self.assertEqual(caught.exception.code, 403)
                error = json.loads(caught.exception.read().decode("utf-8"))
                self.assertEqual(error["error"], "preset_mode_protected")
                self.assertTrue(preset_path.is_file())
            finally:
                self.service.mode_profiles = original_registry

    def test_mode_card_editor_endpoint_persists_generated_can_flow(self) -> None:
        original_registry = self.service.mode_profiles
        original_artifact_root = self.service.artifact_root
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            preset_root = base / "presets"
            user_root = base / "user-modes"
            preset = preset_root / "vehicles" / "denza_n7" / "office.yaml"
            preset.parent.mkdir(parents=True)
            preset.write_text(
                """
id: office
name: 办公模式
scope:
  vehicle_ids: [denza_n7]
operations:
- domain: climate
  capability: climate.set_temperature
  target_scope: passenger
  parameters:
    temperature_c: 24
""".strip()
                + "\n",
                encoding="utf-8",
            )
            request_id = "req-mode-editor-api-001"
            instruction = "副驾空调温度设置为24度"
            artifact_root = base / "runs"
            request_dir = artifact_root / request_id
            request_dir.mkdir(parents=True)
            result = {
                "request_id": request_id,
                "utterance": instruction,
                "status": "succeeded",
                "vehicle": {"id": "denza_n7"},
                "execution_mode": "dry_run",
                "control_transport": "wireless_mqtt",
                "function_configs": {
                    "native_climate": {
                        "spec": {
                            "domain": "climate",
                            "operations": [
                                {
                                    "id": "passenger_temperature",
                                    "capability": "climate.set_temperature",
                                    "target": {"zone": "passenger"},
                                    "parameters": {"temperature_c": 24},
                                }
                            ],
                        }
                    }
                },
                "command_plans": {
                    "native_climate": {
                        "spec": {
                            "commands": [
                                {
                                    "message": "Native_0x1DE",
                                    "signals": {
                                        "frame_id": "0x1DE",
                                        "data_hex": "00 00 00 00 00 18 00 00",
                                    },
                                }
                            ]
                        }
                    }
                },
                "execution_report": {"spec": {"results": []}},
            }
            (request_dir / "result.json").write_text(
                json.dumps(result, ensure_ascii=False), encoding="utf-8"
            )
            self.service.mode_profiles = ModeProfileRegistry(
                preset_root, user_root=user_root
            )
            self.service.artifact_root = artifact_root
            try:
                request = urllib.request.Request(
                    f"{self.base_url}/api/mode-profiles/office",
                    method="POST",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps(
                        {
                            "name": "专注办公模式",
                            "instruction": instruction,
                            "source_request_id": request_id,
                            "vehicle_id": "denza_n7",
                        },
                        ensure_ascii=False,
                    ).encode("utf-8"),
                )
                with self.opener.open(request, timeout=3) as response:
                    payload = json.loads(response.read().decode("utf-8"))
                self.assertEqual(response.status, 200)
                self.assertEqual(payload["storage"], "persistent_yaml")
                self.assertEqual(payload["mode_profile"]["origin"], "customized")
                self.assertEqual(payload["mode_profile"]["instruction"], instruction)
                self.assertTrue(
                    (user_root / "vehicles" / "denza_n7" / "user" / "office.yaml").is_file()
                )

                create_request = urllib.request.Request(
                    f"{self.base_url}/api/mode-profiles",
                    method="POST",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps(
                        {
                            "name": "通勤模式",
                            "instruction": instruction,
                            "source_request_id": request_id,
                            "vehicle_id": "denza_n7",
                        },
                        ensure_ascii=False,
                    ).encode("utf-8"),
                )
                with self.opener.open(create_request, timeout=3) as response:
                    created = json.loads(response.read().decode("utf-8"))
                self.assertEqual(response.status, 201)
                self.assertEqual(created["mode_profile"]["origin"], "user")
                self.assertEqual(created["mode_profile"]["name"], "通勤模式")
                self.assertTrue(
                    (
                        user_root
                        / "vehicles"
                        / "denza_n7"
                        / "user"
                        / f"{created['mode_profile']['id']}.yaml"
                    ).is_file()
                )
            finally:
                self.service.mode_profiles = original_registry
                self.service.artifact_root = original_artifact_root

    def test_office_mode_generation_and_card_execution_reuse_saved_flow(self) -> None:
        original_registry = self.service.mode_profiles
        original_artifact_root = self.service.artifact_root
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        instruction = (
            "空调温度设置为25度，风速设置为1档，"
            "启动闪烁灯，颜色设置为红色"
        )
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            preset_root = base / "presets"
            user_root = base / "users"
            preset = preset_root / "vehicles" / "denza_n7" / "office.yaml"
            preset.parent.mkdir(parents=True)
            preset.write_text(
                """
id: office
name: 办公模式
icon: briefcase
scope:
  vehicle_ids: [denza_n7]
operations:
- domain: climate
  capability: climate.set_temperature
  target_scope: driver
  parameters:
    temperature_c: 24
""".strip()
                + "\n",
                encoding="utf-8",
            )
            extension = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=base / "office-mode-ir.sqlite3",
            )
            extension.sync_peripheral_catalog(
                [
                    {
                        "id": "ir-slot:tony-light-001:1",
                        "status": "learned",
                        "enabled": True,
                        "device_id": "tony-light-001",
                        "device_name": "闪灯玩具",
                        "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                        "card_uid": "2CC8B103",
                        "action": "打开",
                        "action_names": ["打开", "开启", "启动"],
                        "risk": "low",
                        "slot": 1,
                    },
                    {
                        "id": "ir-slot:tony-light-001:3",
                        "status": "learned",
                        "enabled": True,
                        "device_id": "tony-light-001",
                        "device_name": "闪灯玩具",
                        "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                        "card_uid": "2CC8B103",
                        "action": "切换到红色",
                        "action_names": [
                            "切换到红色",
                            "切换为红色",
                            "设置为红色",
                        ],
                        "risk": "low",
                        "slot": 3,
                    },
                ]
            )
            self.service.mode_profiles = ModeProfileRegistry(
                preset_root,
                user_root=user_root,
            )
            self.service.artifact_root = base / "runs"
            SeatDemoHandler.extension = extension
            SeatDemoHandler.nfc = SimpleNamespace(
                snapshot=lambda: {
                    "revision": 1,
                    "reader": {"online": True},
                    "card": {"action": "present", "uid": "2CC8B103"},
                }
            )
            try:
                with self.opener.open(
                    f"{self.base_url}/api/vehicle-display-sync", timeout=3
                ) as response:
                    display_before_compile = json.loads(
                        response.read().decode("utf-8")
                    )
                generate_request = urllib.request.Request(
                    f"{self.base_url}/api/requests/preview",
                    method="POST",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps(
                        {
                            "utterance": instruction,
                            "provider": "local",
                            "conversation_id": "conv-office-mode-generate",
                            "vehicle_id": "denza_n7",
                            "execution_mode": "dry_run",
                            "control_transport": "wired_can",
                            "client_type": "mode_editor_compile",
                        },
                        ensure_ascii=False,
                    ).encode("utf-8"),
                )
                with self.opener.open(generate_request, timeout=3) as response:
                    generated = json.loads(response.read().decode("utf-8"))
                with self.opener.open(
                    f"{self.base_url}/api/vehicle-display-sync", timeout=3
                ) as response:
                    display_after_compile = json.loads(
                        response.read().decode("utf-8")
                    )
                self.assertEqual(
                    display_after_compile["task"],
                    display_before_compile["task"],
                )

                save_request = urllib.request.Request(
                    f"{self.base_url}/api/mode-profiles/office",
                    method="POST",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps(
                        {
                            "name": "办公模式",
                            "instruction": instruction,
                            "source_request_id": generated["request_id"],
                            "vehicle_id": "denza_n7",
                        },
                        ensure_ascii=False,
                    ).encode("utf-8"),
                )
                with self.opener.open(save_request, timeout=3) as response:
                    saved = json.loads(response.read().decode("utf-8"))

                execute_request = urllib.request.Request(
                    f"{self.base_url}/api/mode-profiles/office/execute",
                    method="POST",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps(
                        {
                            "conversation_id": "conv-office-mode-execute",
                            "vehicle_id": "denza_n7",
                            "execution_mode": "dry_run",
                            "control_transport": "wired_can",
                            "client_type": "web_console_mode_card",
                        },
                        ensure_ascii=False,
                    ).encode("utf-8"),
                )
                with patch.object(
                    self.service,
                    "_interpret",
                    side_effect=AssertionError(
                        "saved mode card must not reinterpret natural language"
                    ),
                ), self.opener.open(execute_request, timeout=3) as response:
                    executed = json.loads(response.read().decode("utf-8"))
            finally:
                self.service.mode_profiles = original_registry
                self.service.artifact_root = original_artifact_root
                SeatDemoHandler.extension = original_extension
                SeatDemoHandler.nfc = original_nfc

        generated_commands = [
            (
                command.get("signals", {}).get("debug_command")
                or (
                    f"infrared emit {command.get('signals', {}).get('slot')}"
                    if command.get("message") == "infrared.emit"
                    else None
                )
            )
            for plan in generated.get("command_plans", {}).values()
            for command in plan.get("spec", {}).get("commands", [])
        ]
        self.assertEqual(
            generated_commands,
            [
                "ac temp driver 25",
                "ac fan 1",
                "infrared emit 1",
                "infrared emit 3",
            ],
        )
        self.assertEqual(saved["mode_profile"]["command_flow"]["execution_steps"], [])
        self.assertEqual(executed["status"], "succeeded")
        self.assertEqual(executed["utterance"], "办公模式")
        self.assertEqual(executed["provider_used"], "saved_mode_command_flow")
        self.assertEqual(executed["orchestration_route"], "saved_mode_command_flow")
        self.assertEqual(
            [item["slot"] for item in executed["infrared_actions"]],
            [1, 3],
        )
        self.assertTrue(
            all(
                item["parse_source"] == "saved_mode_command_flow"
                for item in executed["infrared_actions"]
            )
        )

    def test_saved_mode_accepts_prefixed_self_extension_flow_key(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            extension = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "prefixed-mode-ir.sqlite3"),
            )
            extension.sync_peripheral_catalog(
                [
                    {
                        "id": "ir-slot:tony-light-001:1",
                        "status": "learned",
                        "enabled": True,
                        "device_id": "tony-light-001",
                        "device_name": "闪灯玩具",
                        "card_uid": "2CC8B103",
                        "action": "打开",
                        "slot": 1,
                    }
                ]
            )
            handler = object.__new__(SeatDemoHandler)
            handler.extension = extension
            handler.nfc = SimpleNamespace(
                snapshot=lambda: {
                    "card": {"action": "present", "uid": "2CC8B103"}
                }
            )
            profile = SimpleNamespace(
                command_flow={
                    "function_configs": {
                        "local_self_extension": {
                            "spec": {
                                "domain": "self_extension",
                                "operations": [
                                    {
                                        "capability": "infrared.emit",
                                        "target": {"device": "tony-light-001"},
                                        "parameters": {"action": "打开", "slot": 1},
                                    }
                                ],
                            }
                        }
                    }
                }
            )

            entries = handler._saved_mode_infrared_entries(profile)

        self.assertEqual([item["capability"]["slot"] for item in entries], [1])
        self.assertEqual(
            entries[0]["capability"]["parse_source"],
            "saved_mode_command_flow",
        )

    def test_head_unit_display_sync_tracks_tasks_and_usb_catalog(self) -> None:
        task_request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "副驾有点热，有点闷",
                    "provider": "local",
                    "conversation_id": "conv-head-unit-display-sync",
                    "vehicle_id": "denza_n7",
                    "execution_mode": "dry_run",
                    "client_type": "head_unit",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        with self.opener.open(task_request, timeout=3) as response:
            task_result = json.loads(response.read().decode("utf-8"))

        peripheral = {
            "id": "cap-usb-light-on",
            "device_id": "usb-light-1",
            "device_name": "USB 氛围灯条",
            "device_type": "light",
            "type_label": "灯光外设",
            "action": "开启灯带",
        }
        extension_request = urllib.request.Request(
            f"{self.base_url}/api/extension/display-sync",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {"device": {"id": "usb-light-1"}, "peripherals": [peripheral]},
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        with self.opener.open(extension_request, timeout=3) as response:
            extension_result = json.loads(response.read().decode("utf-8"))
        with self.opener.open(
            f"{self.base_url}/api/vehicle-display-sync", timeout=3
        ) as response:
            snapshot = json.loads(response.read().decode("utf-8"))

        self.assertEqual(task_result["status"], "succeeded")
        self.assertTrue(extension_result["success"])
        self.assertEqual(snapshot["task"]["utterance"], "副驾有点热，有点闷")
        self.assertEqual(snapshot["task"]["result"]["request_id"], task_result["request_id"])
        self.assertEqual(snapshot["extension"]["catalog"], [peripheral])
        self.assertEqual(snapshot["extension"]["last_action"]["type"], "hardware_detected")
        self.assertIn("extension_health", snapshot)

    def test_static_console_and_runtime_endpoint(self) -> None:
        with self.opener.open(f"{self.base_url}/", timeout=3) as response:
            html = response.read().decode("utf-8")
        with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-131", timeout=3) as response:
            app_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/gesture-control.js?v=android-pad-9", timeout=3
        ) as response:
            gesture_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/console-v2.js?v=local-first-66", timeout=3
        ) as response:
            console_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/console-v2.css?v=delivery-state-68", timeout=3
        ) as response:
            console_styles = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/speech-input.js?v=sensevoice-13", timeout=3
        ) as response:
            speech_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/speech-input.css?v=sensevoice-7", timeout=3
        ) as response:
            speech_styles = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/speech-audio-worklet.js?v=sensevoice-5", timeout=3
        ) as response:
            speech_worklet = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/self-extension.js?v=gscu-tcp-43", timeout=3
        ) as response:
            extension_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/self-extension.css?v=gscu-tcp-43", timeout=3
        ) as response:
            extension_styles = response.read().decode("utf-8").replace("\r\n", "\n")
        with self.opener.open(
            f"{self.base_url}/assistant-chat.js?v=codex-assistant-27", timeout=3
        ) as response:
            assistant_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/energy-monitor.js?v=energy-monitor-4", timeout=3
        ) as response:
            energy_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/energy-monitor.css?v=energy-monitor-4", timeout=3
        ) as response:
            energy_styles = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/service-worker.js", timeout=3
        ) as response:
            service_worker = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/api/extension/health", timeout=3
        ) as response:
            extension_health = json.loads(response.read().decode("utf-8"))
        with self.opener.open(
            f"{self.base_url}/api/extension/nfc/status", timeout=3
        ) as response:
            nfc_status = json.loads(response.read().decode("utf-8"))
        with self.opener.open(f"{self.base_url}/api/runtime", timeout=3) as response:
            runtime = json.loads(response.read().decode("utf-8"))
        with self.opener.open(f"{self.base_url}/api/speech/status", timeout=3) as response:
            speech_status = json.loads(response.read().decode("utf-8"))

        self.assertIn("511工程-智能座舱", html)
        self.assertIn('id="seatTwinCanvas"', html)
        self.assertIn('data-vehicle-model="byd-tang-2024-reference"', html)
        self.assertIn("整车孪生界面", html)
        self.assertIn("<strong>车内灯光</strong>", html)
        self.assertIn('id="interiorLightPower"', html)
        self.assertIn('id="ambientBrightness"', html)
        self.assertIn('id="ambientColorIndex"', html)
        self.assertIn('data-function="massage"', html)
        self.assertIn("CMD 0x1DE · FB 0x404 · CH1", html)
        self.assertIn("CMD 0x431 · FB 0x41A · CH1", html)
        self.assertIn("CMD 0x431 · FB 0x388 · CH1", html)
        self.assertNotIn("透明车体 · 实体设备", html)
        self.assertIn('src="/app.js?v=dual-seat-131"', html)
        self.assertIn('href="/styles.css?v=dual-seat-23"', html)
        self.assertIn('src="/gesture-control.js?v=android-pad-9"', html)
        self.assertNotIn('class="vehicle-system-hud"', html)
        self.assertIn('src="/console-v2.js?v=local-first-66"', html)
        self.assertIn('href="/console-v2.css?v=delivery-state-68"', html)
        self.assertIn('src="/speech-input.js?v=sensevoice-13"', html)
        self.assertIn('href="/speech-input.css?v=sensevoice-7"', html)
        self.assertIn('id="speechDictationButton"', html)
        self.assertIn('class="speech-waveform"', html)
        self.assertIn('<body class="console-v2">', html)
        self.assertIn('id="v2BootScreen"', html)
        self.assertIn('正在加载车机控制与实时状态', html)
        self.assertIn('src="/self-extension.js?v=gscu-tcp-43"', html)
        self.assertIn('href="/self-extension.css?v=gscu-tcp-43"', html)
        self.assertNotIn("self-calibration", html)
        self.assertNotIn("self-repair", html)
        self.assertNotIn("self-evolution", html)
        self.assertNotIn("滚动查看更多", html)
        self.assertIn('aria-label="发送消息"', html)
        self.assertIn('id="assistantTranscript"', html)
        self.assertIn('href="/assistant-chat.css?v=codex-assistant-26"', html)
        self.assertIn('src="/assistant-chat.js?v=codex-assistant-27"', html)
        self.assertIn('href="/energy-monitor.css?v=energy-monitor-4"', html)
        self.assertIn('src="/energy-monitor.js?v=energy-monitor-4"', html)
        self.assertNotIn('"admin"', console_script)
        self.assertNotIn("v2RoleSelect", console_script)
        self.assertIn('sidebarToggle.id = "v2SidebarToggle"', console_script)
        self.assertIn('id = "v2ModuleSwitcher"', console_script)
        self.assertIn(
            'panelHeading.classList.add("v2-operation-heading", "v2-module-heading")',
            console_script,
        )
        self.assertIn('if (title) title.textContent = "AI助手"', console_script)
        self.assertIn('id="v2PanelControlDock"', console_script)
        self.assertIn('sidebarCollapsed: true', console_script)
        self.assertIn('setSidePanelVisibility("devices", false)', console_script)
        self.assertIn('setSidePanelVisibility("extensions", false)', console_script)
        self.assertIn('setDrawerOpen(true)', assistant_script)
        self.assertIn('($("#v2PanelControlDock") || document.body).append(trigger)', assistant_script)
        self.assertIn('trigger.id = "energyMonitorToggle"', energy_script)
        self.assertIn('document.body.append(layer)', energy_script)
        self.assertIn('.energy-monitor-layer', energy_styles)
        self.assertNotIn('assistantDrawerClose', assistant_script)
        self.assertIn('fetch("/api/vehicle-display-sync"', console_script)
        self.assertIn('document.body.classList.add("v2-passive-display", "v2-interactive-sync")', console_script)
        self.assertIn('input.readOnly = false', console_script)
        self.assertIn('function buildTwinDeviceInteractions()', console_script)
        self.assertIn('id="v2SyncLogList"', console_script)
        self.assertIn('deviceToggle.dataset.panelToggle = "devices"', console_script)
        self.assertIn('extensionToggle.dataset.panelToggle = "extensions"', console_script)
        self.assertIn('id="vehicleExtensionSync"', console_script)
        self.assertNotIn(
            'if (ui.activeModule === "self-extension" && ui.role !== "user")',
            console_script,
        )
        self.assertNotIn(
            'if (ui.activeModule === "self-extension" && role !== "user"',
            console_script,
        )
        self.assertIn('511工程-智能座舱', console_script)
        self.assertIn('网页控制链路', console_script)
        self.assertIn('["self-orchestration", "self-extension"]', console_script)
        self.assertNotIn('label: "自诊断"', console_script)
        self.assertNotIn('runtimeControls.id = "v2RuntimeControls"', console_script)
        self.assertIn('function prepareRuntimeControls(originalForm)', console_script)
        self.assertIn('executionField.hidden = false', console_script)
        self.assertIn('insertVisibleControl(executionField, "v2-sync-execution-field")', console_script)
        self.assertIn('insertVisibleControl(simulationFaultField, "v2-sync-simulation-fault-field")', console_script)
        self.assertNotIn('class="v2-profile"', console_script)
        self.assertNotIn('class="v2-control-switch"', console_script)
        self.assertNotIn('class="v2-config-stack"', console_script)
        self.assertNotIn('id="v2GestureModelValue"', console_script)
        self.assertNotIn('id="v2VehicleState"', console_script)
        self.assertIn('actionRail.append(cameraToggle)', console_script)
        self.assertNotIn("function enhanceConfiguredDropdowns()", console_script)
        self.assertNotIn('modeChips?.addEventListener("wheel"', console_script)
        self.assertIn('classList.toggle("v2-twin-active", definition.index === 0)', console_script)
        self.assertIn('label: "执行结果与硬件回读"', console_script)
        self.assertIn("function keepRunButtonAvailable()", console_script)
        self.assertIn('window.setTimeout(() => $("#v2BootScreen")?.remove(), 220)', console_script)
        self.assertNotIn("function requestUserSafeExecutionMode()", console_script)
        self.assertNotIn('dryRunButton.click()', console_script)
        self.assertIn('const DEFAULT_EXECUTION_MODE = "hardware"', app_script)
        self.assertIn('preferredExecutionMode: DEFAULT_EXECUTION_MODE', app_script)
        self.assertIn('selectExecutionMode(modeAvailable ? preferredMode : "dry_run", { remember: false })', app_script)
        self.assertIn('class="mode-option active" id="hardwareMode"', html)
        self.assertIn('grid-template-columns: clamp(245px, 18.5vw, 310px)', console_styles)
        self.assertNotIn('.v2-control-switch {', console_styles)
        self.assertIn('.v2-module-switcher {', console_styles)
        self.assertIn('grid-template-columns: repeat(2, minmax(0, 1fr))', console_styles)
        self.assertIn(
            '.console-v2 .v2-module-heading.v2-module-heading h1',
            console_styles,
        )
        self.assertIn('font-size: clamp(24px, 2vw, 31px)', console_styles)
        self.assertIn('--v2-module-title-block: clamp(12px, 2vh, 22px)', console_styles)
        self.assertIn('--v2-module-title-inline: clamp(20px, 2.3vw, 34px)', console_styles)
        self.assertIn(
            'grid-template-rows: 36px clamp(236px, 33dvh, 246px) minmax(0, 1fr)',
            console_styles,
        )
        self.assertIn(
            'color-mix(in srgb, var(--v2-accent-soft) 86%, white)',
            console_styles,
        )
        self.assertNotIn(
            '.cabin-channel-legend [data-channel="ch0"]',
            console_styles,
        )
        self.assertIn('margin-inline: calc(0px - var(--v2-main-pad))', console_styles)
        self.assertIn('background: color-mix(in srgb, var(--v2-accent) 30%, var(--v2-accent-soft))', console_styles)
        self.assertIn('width: clamp(48px, 45%, 72px)', console_styles)
        self.assertIn('font-weight: 800', console_styles)
        self.assertIn('white-space: nowrap', console_styles)
        self.assertNotIn('.v2-gesture-model-field', console_styles)
        self.assertIn('.v2-inline-gesture .v2-camera-toggle', console_styles)
        self.assertIn('.v2-inline-gesture .v2-gesture-action-rail', console_styles)
        self.assertIn('grid-template-rows: 42px 10px 42px', console_styles)
        self.assertIn('background: var(--v2-accent);', console_styles)
        self.assertNotIn("gestureArmButton", console_styles)
        self.assertIn('.v2-inline-gesture #gestureStopButton', console_styles)
        self.assertIn('.v2-inline-gesture .v2-gesture-result-stack', console_styles)
        self.assertIn('.v2-inline-gesture .v2-gesture-execution-slot', console_styles)
        self.assertIn('.v2-page-rail', console_styles)
        self.assertNotIn('.v2-dropdown-menu', console_styles)
        self.assertIn('.v2-edge-fade-left', console_styles)
        self.assertIn('.v2-edge-fade-right', console_styles)
        self.assertIn('.mode-group-divider', console_styles)
        self.assertIn('.joint-twin-shell > .cabin-band', console_styles)
        self.assertIn('.console-v2 .v2-page-index', console_styles)
        self.assertIn('.joint-twin-shell .twin-readout', console_styles)
        self.assertIn('.joint-seat-panel .thermal-states', console_styles)
        self.assertIn('.joint-twin-shell .domain-readouts', console_styles)
        self.assertIn('grid-template-columns: repeat(2, minmax(0, 1fr))', console_styles)
        self.assertNotIn('.v2-config-section', console_styles)
        self.assertNotIn('.console-v2 .v2-config-card {', console_styles)
        self.assertIn('.console-v2.v2-sidebar-collapsed .workspace', console_styles)
        self.assertNotIn('id="hardwareAccessField"', html)
        self.assertNotIn('id="hardwareClientLabel"', html)
        self.assertNotIn('id="requestHardwareAccess"', html)
        self.assertNotIn('id="releaseHardwareAccess"', html)
        self.assertNotIn('id="approvalField"', html)
        self.assertNotIn('id="approvalToken"', html)
        self.assertNotIn("approval_token: approvalToken", app_script)
        self.assertNotIn("授权并真实执行", app_script)
        self.assertIn('const dynamicOrder = new Map([["rocking_horse", 0]', app_script)
        self.assertIn('divider.className = "mode-group-divider"', app_script)
        self.assertIn("function isHardwareExecutionReady()", app_script)
        self.assertIn("function hardwareExecutionIssue(", app_script)
        self.assertNotIn(
            'if (mode === "hardware" && !hardwareExecutionReady) return false',
            app_script,
        )
        self.assertIn('data-control-transport="wired_can"', html)
        self.assertIn('data-control-transport="wireless_mqtt"', html)
        self.assertIn('<div class="mode-rail" aria-label="功能模式">', html)
        self.assertIn('id="modeRailToggle"', html)
        self.assertIn('id="modeRailContent"', html)
        self.assertIn('role="region"', html)
        self.assertIn('aria-label="功能模式列表"', html)
        self.assertIn('<span>AI 新建</span>', html)
        self.assertIn('class="mode-rail-identity"', html)
        self.assertIn('aria-expanded="true"', html)
        self.assertIn('模拟下发', html)
        self.assertIn('真实下发', html)
        self.assertNotIn('id="simulationMode"', html)
        self.assertIn('control_transport: state.controlTransport', app_script)
        self.assertIn('button.dataset.connectionState = connectionState', app_script)
        self.assertIn('信号链路${connectionState === "online" ? "正常" : "断开"}', app_script)
        self.assertIn('const transportField = $("#vehicleTransportField", originalForm)', console_script)
        self.assertNotIn('runtimeControls.append(transportField)', console_script)
        self.assertIn('insertVisibleControl(transportField, "v2-sync-transport-field")', console_script)
        self.assertNotIn('class="v2-sync-link-stack"', console_script)
        self.assertNotIn('id="v2HeadUnitState"', console_script)
        self.assertIn('.v2-sync-execution-field .execution-mode-options', console_styles)
        self.assertIn(
            'grid-template-columns: repeat(2, minmax(0, 1fr))',
            console_styles,
        )
        self.assertNotIn('class="v2-sync-protocol"', console_script)
        self.assertIn('.console-v2 .v2-transport-field .vehicle-transport-options', console_styles)
        self.assertIn('.v2-sync-transport-field .mode-option', console_styles)
        self.assertIn('.v2-mode-rail > .mode-rail-content[hidden]', console_styles)
        self.assertIn('font-size: var(--text-control) !important', console_styles)
        self.assertIn('overflow-wrap: anywhere', console_styles)
        self.assertIn('button.disabled = false', app_script)
        self.assertIn('button.setAttribute("aria-pressed", String(selected))', app_script)
        self.assertNotIn(
            'availableTransports.find((item) => item.state_configured)?.id',
            app_script,
        )
        self.assertIn("hardwareButton.disabled = false", app_script)
        self.assertIn(
            'hardwareButton.dataset.executionState = hardwareExecutionReady ? "available" : "blocked"',
            app_script,
        )
        self.assertNotIn(
            'id="hardwareMode" data-execution-mode="hardware" disabled',
            html,
        )
        self.assertIn('const seat = twin.selected === "driver" ? "driver" : "passenger"', app_script)
        self.assertNotIn('data-mode-command="seat driver vent low"', app_script)
        self.assertIn("CH0 副驾座椅与 CH1 主驾座椅独立收发、独立反馈", app_script)
        self.assertIn("CH1 主驾实车测试不启用无闭环的座椅位置手势", app_script)
        self.assertNotIn("车端 CAN 未启用，其他领域保持虚拟执行", app_script)
        self.assertNotIn('id="hardwareNote"', html)
        self.assertNotIn('id="hardwareAliasNote"', html)
        self.assertNotIn('id="vehicleTransportNote"', html)
        self.assertIn("下发控制指令", console_script)
        self.assertIn('new Set(["mode_editor_compile"])', console_script)
        self.assertIn("function extractReferenceWindowPanes", app_script)
        self.assertIn("function collectReferenceFrontLampMeshes", app_script)
        self.assertIn("function emphasizeReferenceWindowPanes", app_script)
        self.assertIn("function configureActiveSeatDepthLayers", app_script)
        self.assertIn("function addTwinGlassDepthMask", app_script)
        self.assertIn("function applyTwinVehicleTheme", app_script)
        self.assertIn("leadingMaxZ", app_script)
        self.assertIn("function updateReferenceWindowOutlineLayers", app_script)
        self.assertIn("vehicle.traverse((object) => object.layers.set(1))", app_script)
        self.assertIn('const categoryPriority = ["body", "glass", "wheel"', app_script)
        self.assertIn("function styleReferencePassiveGlass", app_script)
        self.assertIn("function styleReferencePassiveRearExterior", app_script)
        self.assertIn('twinRole: "reference_passive_rear_bumper"', app_script)
        self.assertNotIn('"cognitive-agent:demo-role-change"', console_script)
        self.assertIn('resultStack.id = "v2GestureResultStack"', console_script)
        self.assertIn('executionSlot.id = "v2GestureExecutionSlot"', console_script)
        self.assertIn('statusTarget.append(requestStatus)', console_script)
        self.assertIn('root.id = "v2SelfExtension"', extension_script)
        self.assertIn('data-extension-mode="study"', extension_script)
        self.assertIn('data-extension-mode="control"', extension_script)
        self.assertIn('外设玩具能力学习', extension_script)
        self.assertIn('外设玩具与车端联动', extension_script)
        self.assertGreaterEqual(extension_script.count('class="v2-module-heading"'), 2)
        self.assertNotIn('class="section-kicker"', extension_script)
        self.assertIn('id="extensionExecuteButton"', extension_script)
        self.assertIn('data-lucide="play"', extension_script)
        self.assertNotIn('id="extensionControlHealth"', extension_script)
        self.assertIn('data-extension-connected="${Boolean(selected)}"', extension_script)
        self.assertIn('<small>红外学习器</small>', extension_script)
        self.assertIn('id="extensionStudyHealthState"', extension_script)
        self.assertIn('id="extensionVehicleState">CHECK</span>', extension_script)
        self.assertIn('id="extensionResetButton"', extension_script)
        self.assertNotIn('id="extensionConnectionButton"', extension_script)
        self.assertNotIn('id="extensionStatusDialog"', extension_script)
        self.assertIn('/api/extension/${path}', extension_script)
        self.assertIn("HEALTH_POLL_INTERVAL_MS", extension_script)
        self.assertIn("HEALTH_REQUEST_TIMEOUT_MS", extension_script)
        self.assertIn("healthState.listening", extension_script)
        self.assertIn("healthState.peer", extension_script)
        self.assertIn('healthState.message || "GSCU1B 尚未连接"', extension_script)
        self.assertIn('window.addEventListener("pageshow"', extension_script)
        self.assertIn('[data-module="self-extension"]', extension_styles)
        self.assertNotIn('[data-module="self-extension"] .v2-role-select', extension_styles)
        self.assertIn('.v2-extension-composer .primary-action {', extension_styles)
        self.assertNotIn('--v2-accent:', extension_styles)
        self.assertIn('.v2-sidebar-collapsed[data-module="self-extension"] .v2-extension-sidebar', extension_styles)
        self.assertIn('.v2-extension-capability span {', extension_styles)
        self.assertIn('.v2-extension-parameter-list > span {', extension_styles)
        self.assertIn('.v2-extension-vehicle-card > footer {', extension_styles)
        self.assertIn('.v2-extension-twin-host > .twin-band {', extension_styles)
        self.assertIn('.v2-extension-type-card[data-extension-connected="true"]', extension_styles)
        self.assertIn('.v2-extension-type-card[data-extension-connected="false"]', extension_styles)
        self.assertIn(
            '.v2-extension-control\n.v2-extension-learned-action {\n  --extension-card-color: var(--v2-accent);',
            extension_styles,
        )
        self.assertIn('grid-template-columns: repeat(3, minmax(0, 1fr))', extension_styles)
        self.assertTrue(extension_health["success"])
        self.assertIn(extension_health["transport"], {"tcp", "mqtt", "simulation"})
        self.assertIn("nfc", extension_health)
        self.assertTrue(nfc_status["success"])
        self.assertIn("reader", nfc_status)
        self.assertIn("card", nfc_status)
        self.assertIn("legacy_profiles", nfc_status)
        self.assertIn("legacy_bridge_addresses", nfc_status)
        self.assertIn("function emphasizeReferenceFrontLampMeshes", app_script)
        self.assertIn("function createReferenceFrontLampHighlights", app_script)
        self.assertIn("referenceFrontLampMaterials", app_script)
        self.assertIn("targets.windowPositions", app_script)
        self.assertIn(
            'const REFERENCE_VEHICLE_ASSET_VERSION = "77a0fae7"', app_script
        )
        self.assertIn(
            "const REFERENCE_VEHICLE_LOAD_TIMEOUT_MS = 30000", app_script
        )
        self.assertIn(
            'referenceVehicleAssetUrl("byd-tang-2024.fbx")', app_script
        )
        self.assertIn("const stageHasLayout = () =>", app_script)
        self.assertIn("window.requestAnimationFrame(initSeatTwin)", app_script)
        self.assertIn('clientType = "web_console"', app_script)
        self.assertIn("client_type: clientType", app_script)
        self.assertIn("function executionVerificationStatus(result = {})", app_script)
        self.assertIn('"submitted_unverified"', app_script)
        self.assertIn("指令已提交，等待状态确认", app_script)
        self.assertIn("指令已提交 · 等待状态确认", console_script)
        self.assertIn("CAN 指令已提交，等待车辆状态确认", console_script)
        self.assertIn('canvas.addEventListener("webglcontextlost"', app_script)
        self.assertIn('"cognitive-agent:module-change"', app_script)
        self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v282"', service_worker)
        self.assertIn('/console-v2.css?v=delivery-state-68', service_worker)
        self.assertIn('/console-v2.js?v=local-first-66', service_worker)
        self.assertIn('/assistant-chat.css?v=codex-assistant-26', service_worker)
        self.assertIn('/assistant-chat.js?v=codex-assistant-27', service_worker)
        self.assertIn('/energy-monitor.css?v=energy-monitor-4', service_worker)
        self.assertIn('/energy-monitor.js?v=energy-monitor-4', service_worker)
        self.assertIn('"/app.js?v=dual-seat-131"', service_worker)
        self.assertIn('"/speech-input.js?v=sensevoice-13"', service_worker)
        self.assertIn('"/speech-input.css?v=sensevoice-7"', service_worker)
        self.assertIn('"/speech-audio-worklet.js?v=sensevoice-5"', service_worker)
        self.assertIn('fetch("/api/speech/preview"', speech_script)
        self.assertIn('fetch("/api/speech/transcribe"', speech_script)
        self.assertIn('"X-Speech-Utterance-Id"', speech_script)
        self.assertIn('"X-Speech-Preview-Sequence"', speech_script)
        self.assertIn(
            '"X-Speech-Wake-Word-Required": inputMode === "wake" ? "1" : "0"',
            speech_script,
        )
        self.assertIn("result.assistant_text", speech_script)
        self.assertIn("AudioWorkletNode", speech_script)
        self.assertIn('registerProcessor("speech-capture-processor"', speech_worklet)
        self.assertIn("navigator.mediaDevices.getUserMedia", speech_script)
        self.assertIn("PREFERRED_MICROPHONE", speech_script)
        self.assertIn("ENDPOINT_SILENCE_MS", speech_script)
        self.assertNotIn("input.dataset.speechExecutionEligible =", speech_script)
        self.assertIn("lastAppliedPreviewSequence", speech_script)
        self.assertIn("result.stale === true", speech_script)
        self.assertIn("wakeWord.detected === true", speech_script)
        self.assertIn("wakeWord.remainder ?? result.live_text", speech_script)
        self.assertIn("status.wake_word.primary", speech_script)
        self.assertIn("speechExecutionEligible", app_script)
        self.assertIn("speech_utterance_id", app_script)
        self.assertIn("speechUtteranceId", speech_script)
        self.assertIn('const DEFAULT_WAKE_WORD = "你好，小迪"', speech_script)
        self.assertIn('replaceInput(liveText, "partial")', speech_script)
        self.assertIn('speech_semantic: "语音语义"', console_script)
        self.assertIn("语音意图已识别，但当前车辆能力不支持执行", app_script)
        self.assertIn("语音控制参数不完整，请补充后再执行", app_script)
        self.assertIn(".speech-dictation.is-listening", speech_styles)
        self.assertEqual(speech_status["model"], "SenseVoiceSmall")
        self.assertIn('"/assets/pets/self-extension-cat.png?v=3"', service_worker)
        self.assertIn(".then(() => self.skipWaiting())", service_worker)
        self.assertNotIn("const cached = await caches.match(request)", service_worker)
        self.assertNotIn("self-evolution", service_worker)
        self.assertNotIn("self-calibration", service_worker)
        self.assertNotIn("self-repair", service_worker)
        self.assertNotIn("n7_seat.glb", service_worker)
        self.assertIn("hardware_access_required", app_script)
        self.assertIn('button.setAttribute("aria-label", "关闭摄像头")', gesture_script)
        self.assertIn("function refreshCameraAvailability", gesture_script)
        self.assertIn("navigator.mediaDevices.enumerateDevices()", gesture_script)
        self.assertIn('addEventListener?.("devicechange", handleCameraDeviceChange)', gesture_script)
        self.assertIn('case "NotFoundError":', gesture_script)
        self.assertIn('case "NotAllowedError":', gesture_script)
        self.assertIn('case "NotReadableError":', gesture_script)
        self.assertIn('case "OverconstrainedError":', gesture_script)
        self.assertNotIn("sendSimulatedGesture", gesture_script)
        self.assertNotIn("gestureDebug", gesture_script)
        self.assertIn("HARDWARE_REQUEST_TIMEOUT_MS = 15 * 60 * 1000", app_script)
        self.assertIn(
            "requestTimeoutForExecutionMode(executionMode)", app_script
        )
        self.assertIn('error.code === "hardware_busy"', app_script)
        self.assertIn("BENCH BUSY", app_script)
        self.assertIn("FBXLoader.js", html)
        self.assertIn('data-seat-layout="five-seat"', html)

        self.assertIn('data-active-seat="front-right"', html)
        self.assertNotIn('placeholder="请输入服务端配置令牌"', html)
        self.assertNotIn('class="twin-callout-layer"', html)
        self.assertNotIn('class="twin-callout', html)
        self.assertNotIn('data-callout=', html)
        self.assertIn('data-twin-function="ventilation"', html)
        self.assertIn('data-twin-function="heating"', html)
        self.assertIn('class="twin-level-meter"', html)
        self.assertIn('id="presetIndicator"', html)
        self.assertIn('id="vehicleSelect"', html)
        self.assertIn('id="saveModeToggle"', html)
        self.assertIn("AI 新建", html)
        self.assertIn('aria-label="快捷功能模式，可横向滚动"', html)
        self.assertIn('id="modeRailContent"', html)
        self.assertIn('aria-label="功能模式列表"', html)
        self.assertNotIn('id="modeRailPrev"', html)
        self.assertNotIn('id="modeRailNext"', html)
        self.assertGreater(html.index('id="saveModeToggle"'), html.index('id="modeRailContent"'))
        self.assertIn('id="modeGalleryToggle"', html)
        self.assertIn('id="modeGallery"', html)
        self.assertIn('id="modeGalleryGroups"', html)
        self.assertIn("管理功能模式", html)
        self.assertIn('id="modeEditorBackdrop"', html)
        self.assertIn('id="modeEditorName"', html)
        self.assertIn('id="modeEditorInstruction"', html)
        self.assertIn('id="modeEditorExecute"', html)
        self.assertIn('id="modeEditorComplete"', html)
        self.assertIn('id="modeEditorFlowPreview"', html)
        self.assertIn('class="mode-editor-flow-preview empty"', html)
        self.assertNotIn('id="saveModeSource"', html)
        self.assertNotIn("主驾 24℃", html)
        populate_modes = app_script[
            app_script.index("function populateModeChips()")
            : app_script.index("function setModeExecutionState")
        ]
        self.assertNotIn("ch1_body_can_driver_only", populate_modes)
        self.assertIn("function populateModeGallery", app_script)
        self.assertNotIn("function updateModeRailNavigation", app_script)
        self.assertNotIn("function scrollModeRail", app_script)
        self.assertIn('className = "mode-gallery-card"', app_script)
        self.assertIn("function createModeGalleryAddCard", app_script)
        self.assertIn('button.dataset.modeCreate = "true"', app_script)
        self.assertIn("function openModeEditorForCreate", app_script)
        self.assertIn('apiRequest(endpoint', app_script)
        self.assertIn('? "/api/mode-profiles"', app_script)
        self.assertIn("#modeGallery button[data-mode-profile-id]", app_script)
        self.assertIn("function setModeExecutionState", app_script)
        self.assertNotIn("function requestCommandAndWait", app_script)
        self.assertIn("已永久保存到模式卡片", app_script)
        self.assertLess(
            app_script.index('id: "static"'),
            app_script.index('id: "dynamic"'),
        )
        gallery_groups = app_script[
            app_script.index("const MODE_GALLERY_GROUPS")
            : app_script.index("const MODE_DOMAIN_LABELS")
        ]
        self.assertNotIn('id: "personal"', gallery_groups)
        self.assertNotIn('id: "frequent"', gallery_groups)
        self.assertNotIn('id: "comfort"', gallery_groups)
        self.assertIn("function isDynamicModeProfile", app_script)
        self.assertIn('if (group.id === "custom") grid.append(createModeGalleryAddCard())', app_script)
        self.assertIn('button.dataset.modeOrigin === "user"', app_script)
        self.assertIn('appendGroupLabel("自定义模式", "custom")', app_script)
        self.assertIn('appendDivider("custom")', app_script)
        self.assertIn('deleteUserModeProfile(deleteButton.dataset.modeDelete)', app_script)
        self.assertIn('method: "DELETE"', app_script)
        self.assertIn('deleteButton.dataset.modeDelete = mode.id', app_script)
        self.assertIn('editButton.dataset.modeEdit = mode.id', app_script)
        self.assertIn("function openModeEditor", app_script)
        self.assertIn("function executeModeEditorInstruction", app_script)
        self.assertIn("function completeModeEditor", app_script)
        self.assertIn("function modeEditorReadableCommands", app_script)
        self.assertIn("function modeEditorMessagesForOperation", app_script)
        self.assertIn("function modeEditorAppendCommandCard", app_script)
        self.assertIn('title.textContent = flow?.control_transport === "wireless_mqtt" ? "MQTT 消息序列"', app_script)
        self.assertIn("message.payload", app_script)
        self.assertIn("source_request_id: editor.sourceRequestId", app_script)
        self.assertIn('method: "POST"', app_script)
        self.assertIn('data-domain-panel="climate"', html)
        self.assertIn('data-domain-panel="roof"', html)
        self.assertIn('data-domain-panel="ambient_light"', html)
        self.assertIn('data-domain-panel="window"', html)
        self.assertIn('data-domain-panel="front_light"', html)
        self.assertIn('data-domain-panel="vehicle_status"', html)
        self.assertIn('id="vehicleGear"', html)
        self.assertIn('id="vehicleSpeed"', html)
        self.assertNotIn('data-domain-panel="control_link"', html)
        self.assertIn('id="cabinStateTitle">整车设备状态</h3>', html)
        self.assertIn('id="cabinCarouselPrev"', app_script)
        self.assertIn('id="cabinCarouselNext"', app_script)
        self.assertIn('aria-label="左右浏览设备卡片"', app_script)
        self.assertNotIn('class="cabin-category-tabs"', app_script)
        self.assertNotIn('class="section-kicker"', html)
        self.assertIn('.joint-twin-shell > .twin-band {', console_styles)
        self.assertIn('grid-template-rows: 62px minmax(0, 1fr)', console_styles)
        self.assertIn('grid-template-columns: repeat(2, minmax(0, 1fr))', console_styles)
        self.assertIn('--cabin-cards-per-view: 2', console_styles)
        self.assertIn('const cabinCardsPerView = () => 2', app_script)
        self.assertIn('scroll-snap-type: x mandatory', console_styles)
        self.assertIn('data-device-power-value', app_script)
        self.assertIn('查看${name}功率详情', app_script)
        self.assertIn('name: "车内灯光", measurable: false', app_script)
        self.assertIn('.device-power-entry.unavailable', console_styles)
        self.assertIn('powerDrawer.id = "devicePowerDrawer"', app_script)
        self.assertIn('window.updateDevicePowerTelemetry = updatePowerTelemetry', app_script)
        self.assertIn('["longitudinal_motor", "水平电机"]', app_script)
        self.assertIn('["cushion_motor", "坐盆电机"]', app_script)
        self.assertIn('["backrest_motor", "座椅靠背电机"]', app_script)
        self.assertIn('["legrest_motor", "腿托电机"]', app_script)
        self.assertIn('["ventilation_control", "通风控制"]', app_script)
        self.assertIn('["heating_control", "加热控制"]', app_script)
        self.assertIn('["compressor", "压缩机"]', app_script)
        self.assertIn('["electronic_fan", "电子风扇"]', app_script)
        self.assertIn('["blower", "鼓风机"]', app_script)
        self.assertIn('["thermal_management_module", "空调热管理集成模块"]', app_script)
        self.assertIn('["compressor_controller", "压缩机电控"]', app_script)
        self.assertIn('["electronic_cooling_pump", "电控冷却水泵"]', app_script)
        self.assertIn('["front_left_window", "左前车窗"]', app_script)
        self.assertIn('["front_right_window", "右前车窗"]', app_script)
        self.assertIn('["rear_left_window", "左后车窗"]', app_script)
        self.assertIn('["rear_right_window", "右后车窗"]', app_script)
        self.assertIn('["front_left_combination_lamp", "左前组合灯"]', app_script)
        self.assertIn('["front_right_combination_lamp", "右前组合灯"]', app_script)
        self.assertIn('name: "主驾座椅", measurable: true, components: []', app_script)
        self.assertIn('name: "遮阳帘", measurable: true, components: []', app_script)
        self.assertIn("仅支持设备总功率", app_script)
        self.assertIn('.device-power-drawer {', console_styles)
        self.assertNotIn('class="roof-visual"', html)
        self.assertNotIn('class="ambient-visual"', html)
        self.assertIn('id="driverAirDirection"', html)
        self.assertIn('id="passengerAirDirection"', html)
        self.assertNotIn('id="transmittanceValue"', html)
        self.assertNotIn('id="roofMotorState"', html)
        self.assertIn('id="ambientBrightness"', html)
        self.assertNotIn('class="example-group"', html)
        self.assertNotIn('id="safetyStrip"', html)
        self.assertNotIn('class="pipeline"', html)
        self.assertNotIn('class="topbar"', html)
        self.assertNotIn('data-axis="height"', html)
        self.assertEqual(runtime["domain"], "cabin")
        self.assertEqual(
            runtime["hardware_execution_allowed"],
            (
                runtime["hardware"]["configured"]
                or runtime["vehicle_can"]["configured"]
                or runtime["vehicle_mqtt"]["configured"]
            ),
        )
        self.assertEqual(runtime["hardware"]["supported_targets"], ["front_right"])
        self.assertEqual(
            runtime["hardware"]["concurrency_policy"],
            "single_flight_fail_fast",
        )
        self.assertEqual(
            runtime["hardware"]["access_policy"],
            "local_approval_idle_lease",
        )
        self.assertEqual(
            runtime["hardware"]["access_inactivity_timeout_seconds"], 180
        )
        self.assertFalse(runtime["hardware"]["busy"])
        self.assertEqual(
            runtime["hardware"]["supported_seat_instances"], ["passenger"]
        )
        self.assertGreaterEqual(runtime["matrix"]["messages"], 25)
        self.assertGreaterEqual(runtime["matrix"]["signals"], 128)
        self.assertEqual(runtime["motion_position_range_pct"], [0, 100])
        self.assertEqual(runtime["default_vehicle_id"], "denza_n7")
        self.assertEqual(runtime["vehicle"]["name"], "腾势 N7")
        self.assertIn(
            "denza_n7",
            {vehicle["id"] for vehicle in runtime["vehicles"]},
        )
        self.assertIn(
            "zero_gravity",
            {mode["id"] for mode in runtime["modes"]},
        )
        modes_by_id = {mode["id"]: mode for mode in runtime["modes"]}
        self.assertEqual(modes_by_id["rocking_horse"]["behavior"], "cyclic")
        self.assertEqual(modes_by_id["dynamic_cradle"]["behavior"], "cyclic")
        self.assertEqual(modes_by_id["office"]["behavior"], "static")
        self.assertTrue(
            all(
                mode["behavior"] in {"static", "cyclic"}
                for mode in runtime["modes"]
            )
        )
        self.assertEqual(
            runtime["vehicle"]["seat"]["motors"]["longitudinal"]["range"],
            215,
        )
        self.assertEqual(
            runtime["vehicle"]["seat"]["motors"]["longitudinal"]["reference_zero"],
            "forwardmost",
        )
        self.assertEqual(
            runtime["vehicle"]["seat"]["motors"]["cushion_tilt"]["reference_zero"],
            "highest",
        )
        capability_names = {item["name"] for item in runtime["capabilities"]}
        self.assertIn("seat.save_posture_preset", capability_names)
        self.assertIn("seat.apply_posture_preset", capability_names)
        self.assertIn("climate.set_temperature", capability_names)
        self.assertIn("roof.set_transmittance", capability_names)
        self.assertIn("ambient_light.set_illumination", capability_names)
        passenger_seat_capabilities = [
            item
            for item in runtime["capabilities"]
            if item["name"].startswith("seat.")
            and not item["name"].startswith("seat.set_driver_")
            and item["name"] != "seat.recall_driver_memory_native"
        ]
        driver_seat_capabilities = [
            item
            for item in runtime["capabilities"]
            if item["name"].startswith("seat.set_driver_")
            or item["name"] == "seat.recall_driver_memory_native"
        ]
        self.assertTrue(passenger_seat_capabilities)
        self.assertTrue(
            all(
                item["target_scopes"] == ["passenger"]
                for item in passenger_seat_capabilities
            )
        )
        self.assertTrue(driver_seat_capabilities)
        self.assertTrue(
            all(
                item["target_scopes"] == ["driver", "front_left"]
                for item in driver_seat_capabilities
            )
        )

    def test_speech_transcription_endpoint_forwards_wav_without_persisting_it(self) -> None:
        class SpeechStub:
            received = b""

            @classmethod
            def transcribe_wav(
                cls,
                content: bytes,
                *,
                utterance_id: str | None = None,
                require_wake_word: bool = False,
            ) -> dict[str, object]:
                self.assertFalse(require_wake_word)
                cls.received = content
                return {
                    "utterance_id": utterance_id,
                    "text": "打开空调到二十四度。",
                    "language": "zh",
                    "duration_ms": 800,
                    "inference_ms": 92,
                    "model": "SenseVoiceSmall",
                }

        original_speech = SeatDemoHandler.speech
        SeatDemoHandler.speech = SpeechStub()
        try:
            wav_content = b"RIFF-speech-test"
            request = urllib.request.Request(
                f"{self.base_url}/api/speech/transcribe",
                method="POST",
                headers={
                    "Content-Type": "audio/wav",
                    "Accept": "application/json",
                    "X-Speech-Utterance-Id": "e2e-speech-1",
                },
                data=wav_content,
            )
            with self.opener.open(request, timeout=3) as response:
                payload = json.loads(response.read().decode("utf-8"))
                permissions_policy = response.headers["Permissions-Policy"]
        finally:
            SeatDemoHandler.speech = original_speech

        self.assertEqual(SpeechStub.received, wav_content)
        self.assertEqual(payload["text"], "打开空调到二十四度。")
        self.assertEqual(payload["utterance_id"], "e2e-speech-1")
        self.assertIn("microphone=(self)", permissions_policy)

    def test_speech_utterance_id_preserves_structured_slots_into_intent_ir(self) -> None:
        understanding = UnderstandingTranscriptProcessor(REPOSITORY_ROOT).process(
            "把副驾腿托向上调整到最大位置"
        ).to_dict()
        understanding["utterance_id"] = "e2e-speech-bridge"

        class SpeechStub:
            @staticmethod
            def cached_result(utterance_id: str) -> dict[str, object]:
                if utterance_id != "e2e-speech-bridge":
                    raise AssertionError("unexpected speech utterance id")
                return understanding

        original_speech = SeatDemoHandler.speech
        SeatDemoHandler.speech = SpeechStub()
        try:
            request = urllib.request.Request(
                f"{self.base_url}/api/requests",
                method="POST",
                headers={"Content-Type": "application/json", "Accept": "application/json"},
                data=json.dumps(
                    {
                        "utterance": understanding["clean_text"],
                        "speech_utterance_id": "e2e-speech-bridge",
                        "client_type": "web_console",
                        "provider": "openai",
                        "conversation_id": "conv-e2e-speech-bridge",
                        "execution_mode": "dry_run",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )
            with self.opener.open(request, timeout=3) as response:
                payload = json.loads(response.read().decode("utf-8"))
        finally:
            SeatDemoHandler.speech = original_speech

        self.assertEqual(payload["status"], "succeeded")
        self.assertEqual(payload["provider_used"], "speech_semantic")
        self.assertEqual(
            payload["speech_understanding"]["raw_text"],
            "把副驾腿托向上调整到最大位置",
        )
        action = payload["intent_ir"]["spec"]["actions"][0]
        self.assertEqual(action["capability"], "seat.set_legrest_height")
        self.assertEqual(action["target"]["instance"], "passenger")
        self.assertEqual(action["parameters"], {"position_pct": 100})
        self.assertEqual(action["provenance"]["parser"], "speech_semantic")

    def test_roof_shade_speech_crosses_server_bridge_as_native_can(self) -> None:
        cases = (
            ("把遮阳帘打开", {"motion": "open"}),
            ("遮阳帘打开一半", {"motion": "half"}),
        )
        for index, (utterance, parameters) in enumerate(cases):
            with self.subTest(utterance=utterance):
                utterance_id = f"e2e-roof-shade-{index}"
                understanding = UnderstandingTranscriptProcessor(
                    REPOSITORY_ROOT
                ).process(utterance).to_dict()
                understanding["utterance_id"] = utterance_id

                def cached_result(
                    requested_id: str,
                    *,
                    expected_id: str = utterance_id,
                    result: dict[str, object] = understanding,
                ) -> dict[str, object]:
                    if requested_id != expected_id:
                        raise AssertionError("unexpected speech utterance id")
                    return result

                original_speech = SeatDemoHandler.speech
                SeatDemoHandler.speech = SimpleNamespace(cached_result=cached_result)
                try:
                    request = urllib.request.Request(
                        f"{self.base_url}/api/requests",
                        method="POST",
                        headers={
                            "Content-Type": "application/json",
                            "Accept": "application/json",
                        },
                        data=json.dumps(
                            {
                                "utterance": understanding["clean_text"],
                                "speech_utterance_id": utterance_id,
                                "client_type": "web_console",
                                "provider": "openai",
                                "conversation_id": f"conv-e2e-roof-shade-{index}",
                                "execution_mode": "dry_run",
                            },
                            ensure_ascii=False,
                        ).encode("utf-8"),
                    )
                    with self.opener.open(request, timeout=3) as response:
                        payload = json.loads(response.read().decode("utf-8"))
                finally:
                    SeatDemoHandler.speech = original_speech

                self.assertEqual(payload["status"], "succeeded")
                self.assertEqual(payload["provider_used"], "speech_semantic")
                self.assertEqual(payload["deferred_requests"], [])
                action = payload["intent_ir"]["spec"]["actions"][0]
                self.assertEqual(action["capability"], "roof.set_shade_motion")
                self.assertEqual(action["target"]["zone"], "vehicle")
                self.assertEqual(action["parameters"], parameters)
                self.assertEqual(action["provenance"]["parser"], "speech_semantic")

    def test_self_extension_vehicle_success_contract_is_strict(self) -> None:
        with self.opener.open(
            f"{self.base_url}/self-extension.js?v=extension-44", timeout=3
        ) as response:
            extension_script = response.read().decode("utf-8")
        with self.opener.open(
            f"{self.base_url}/app.js?v=dual-seat-131", timeout=3
        ) as response:
            app_script = response.read().decode("utf-8")

        self.assertIn('code: "SEAT · CH1 BODY CAN"', extension_script)
        self.assertIn('code: "SEAT · CH0 T2G CAN"', extension_script)
        self.assertIn('id: "passenger-seat"', extension_script)
        self.assertIn('parameters: ["通风 0–3 档", "加热 0–2 档"', extension_script)
        self.assertIn('command: "seat driver vent low"', extension_script)
        self.assertIn('command: "seat driver vent high"', extension_script)
        self.assertIn('command: "seat driver vent off"', extension_script)
        self.assertIn('command: "seat driver heat off"', extension_script)
        self.assertIn('command: "ac off"', extension_script)
        self.assertIn('command: "light indoor off"', extension_script)
        self.assertIn('command: "light reading all off"', extension_script)
        self.assertIn('command: "light ambient off"', extension_script)
        self.assertIn('command: `window ${positionName} position ${position}`', extension_script)
        self.assertNotIn('command: "shade stop"', extension_script)
        self.assertNotIn('command: "door vehicle lock"', extension_script)
        self.assertNotIn('command: "door vehicle unlock"', extension_script)
        self.assertNotIn('command: "door rear unlock"', extension_script)
        self.assertNotIn('code: "OUTLET · 4 ZONES"', extension_script)
        self.assertNotIn('command: "打开全部车窗"', extension_script)
        self.assertNotIn('command: "打开副驾座椅通风"', extension_script)
        self.assertNotIn('command: "把副驾座椅向后调一点"', extension_script)
        self.assertIn('result?.status === "succeeded"', extension_script)
        self.assertIn("everyOperationSucceeded", extension_script)
        self.assertIn("everyHardwareOperationConfirmed", extension_script)
        self.assertIn("result?.physical_feedback_confirmed === true", extension_script)
        self.assertIn("cognitive-agent:extension-vehicle-result", extension_script)
        self.assertIn("MASSAGE", extension_script)
        self.assertIn("INTERIOR", extension_script)
        self.assertIn("AMBIENT", extension_script)
        self.assertIn(
            "body: JSON.stringify({ command, conversation_id: sharedConversationId() })",
            extension_script,
        )
        self.assertIn('window.addEventListener("cognitive-agent:result"', extension_script)
        self.assertIn('id="extensionVehicleStateSummary"', extension_script)
        self.assertIn('id="extensionTwinHost"', extension_script)
        self.assertIn('data-extension-current-action', extension_script)
        self.assertIn('function syncSharedTwinMount()', extension_script)
        self.assertIn('function updateVehicleCardExecution(', extension_script)
        self.assertNotIn("/api/hardware/seat-state", extension_script)
        self.assertIn(
            'window.addEventListener("cognitive-agent:extension-vehicle-result"',
            app_script,
        )
        self.assertIn("applyAuthoritativeVehicleState", app_script)
        self.assertIn('payload.conversation_id !== state.conversationId', app_script)
        self.assertIn(
            'target_policy === "ch1_body_can_driver_only"',
            app_script,
        )
        self.assertIn("async function synchronizePassengerSeatState()", app_script)
        self.assertIn("/api/hardware/seat-state", app_script)
        self.assertIn("async function synchronizeDriverSeatState()", app_script)
        self.assertIn("/api/hardware/vehicle-state", app_script)
        self.assertNotIn("passenger: mergedActuatorState.driver", app_script)
        self.assertNotIn("approvalText", app_script)

    def test_self_extension_vehicle_endpoint_preserves_unsuccessful_feedback(
        self,
    ) -> None:
        service_result = {
            "status": "partially_succeeded",
            "provider_used": "local_rules",
            "request_id": "req-extension-partial",
            "conversation_id": "conv-extension-shared-e2e",
            "conversation_turn": 4,
            "decision_summary": ["空调完成，遮阳帘反馈超时。"],
            "validation": [],
            "execution_report": {
                "spec": {
                    "status": "failed",
                    "physical_state": {
                        "source": "physical_can_ch1",
                        "complete": False,
                    },
                    "results": [
                        {
                            "ref": "climate-1",
                            "domain": "climate",
                            "status": "succeeded",
                            "telemetry": {
                                "simulated": False,
                                "feedback_observed": True,
                                "feedback_match": True,
                                "feedback_frame": {"frame_id": "0x404"},
                            },
                        },
                        {
                            "ref": "roof-1",
                            "domain": "roof",
                            "status": "failed",
                            "message": "反馈超时",
                            "telemetry": {"simulated": False},
                        },
                    ],
                }
            },
            "domain_state": {"climate": {"power": "on"}},
            "actuator_state": {"driver": {"heating": {"state": "off"}}},
            "deferred_requests": [],
        }
        request = urllib.request.Request(
            f"{self.base_url}/api/extension/vehicle/execute",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "command": "打开空调并关闭遮阳帘",
                    "conversation_id": "conv-extension-shared-e2e",
                }
            ).encode("utf-8"),
        )
        with patch.object(self.service, "process", return_value=service_result):
            with self.opener.open(request, timeout=3) as response:
                result = json.loads(response.read().decode("utf-8"))

        self.assertFalse(result["success"])
        self.assertEqual(result["status"], "partially_succeeded")
        self.assertEqual(result["conversation_id"], "conv-extension-shared-e2e")
        self.assertEqual(result["conversation_turn"], 4)
        self.assertEqual(result["execution_report"], service_result["execution_report"])
        self.assertEqual(result["domain_state"], service_result["domain_state"])
        self.assertEqual(result["actuator_state"], service_result["actuator_state"])
        self.assertEqual(result["physical_state"]["source"], "physical_can_ch1")
        self.assertEqual(len(result["physical_feedback"]), 2)
        self.assertFalse(result["physical_feedback_confirmed"])

    def test_self_extension_gateway_uses_unified_authenticated_service(self) -> None:
        original_extension = SeatDemoHandler.extension
        SeatDemoHandler.extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name, "gateway-auth.sqlite3"
            ),
        )
        try:
            with self.opener.open(
                f"{self.base_url}/api/extension/health", timeout=3
            ) as response:
                health = json.loads(response.read().decode("utf-8"))
            request = urllib.request.Request(
                f"{self.base_url}/api/extension/gscu/emit",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps({"slot": 9}).encode("utf-8"),
            )
            with self.opener.open(request, timeout=3) as response:
                result = json.loads(response.read().decode("utf-8"))
        finally:
            SeatDemoHandler.extension = original_extension

        self.assertTrue(health["success"])
        self.assertEqual(health["transport"], "simulation")
        self.assertTrue(health["connected"])
        self.assertTrue(result["success"])
        self.assertEqual(result["transport"], "simulation")
        self.assertFalse(result["physical_state_confirmed"])

    def test_self_extension_tcp_configuration_endpoint_updates_runtime(self) -> None:
        original_extension = SeatDemoHandler.extension
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name, "tcp-config.sqlite3"
            ),
        )
        SeatDemoHandler.extension = extension
        try:
            with tempfile.TemporaryDirectory() as directory:
                with patch.dict(
                    os.environ,
                    {"GSCU_RUNTIME_CONFIG_PATH": f"{directory}/gateway.json"},
                    clear=False,
                ):
                    request = urllib.request.Request(
                        f"{self.base_url}/api/extension/gscu/config",
                        method="POST",
                        headers={"Content-Type": "application/json"},
                        data=json.dumps(
                            {"listen_host": "127.0.0.1", "port": 45444}
                        ).encode("utf-8"),
                    )
                    with self.opener.open(request, timeout=3) as response:
                        result = json.loads(response.read().decode("utf-8"))
                    persisted = json.loads(
                        Path(directory, "gateway.json").read_text(encoding="utf-8")
                    )
        finally:
            extension.close()
            SeatDemoHandler.extension = original_extension

        self.assertTrue(result["success"])
        self.assertEqual(result["transport"], "tcp")
        self.assertEqual(result["listen_host"], "127.0.0.1")
        self.assertEqual(result["port"], 45444)
        self.assertEqual(
            persisted,
            {"transport": "tcp", "listen_host": "127.0.0.1", "port": 45444},
        )

    def test_infrared_slot_catalog_api_and_confirmed_erase_all(self) -> None:
        original_extension = SeatDemoHandler.extension
        with tempfile.TemporaryDirectory() as directory:
            extension = SelfExtensionService(
                SimulationGateway(),
                catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
            )
            SeatDemoHandler.extension = extension
            try:
                for slot, name in ((12, "电视开关"), (3, "空调制冷")):
                    request = urllib.request.Request(
                        f"{self.base_url}/api/extension/gscu/slots/{slot}",
                        method="PUT",
                        headers={"Content-Type": "application/json"},
                        data=json.dumps({"name": name, "source": "tcp"}).encode("utf-8"),
                    )
                    with self.opener.open(request, timeout=3) as response:
                        self.assertEqual(response.status, 200)

                structured_request = urllib.request.Request(
                    f"{self.base_url}/api/extension/gscu/slots/5",
                    method="PUT",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps(
                        {
                            "device_id": "toy-light-001",
                            "object_names": ["闪烁灯", "闪灯玩具"],
                            "action_names": ["切换为蓝色", "设置为蓝色"],
                            "source": "tcp",
                        },
                        ensure_ascii=False,
                    ).encode("utf-8"),
                )
                with self.opener.open(structured_request, timeout=3) as response:
                    structured = json.loads(response.read().decode("utf-8"))
                self.assertEqual(structured["slot"]["device_id"], "toy-light-001")
                self.assertEqual(
                    structured["slot"]["object_names"], ["闪烁灯", "闪灯玩具"]
                )
                self.assertEqual(
                    structured["slot"]["action_names"],
                    ["切换为蓝色", "设置为蓝色"],
                )

                with self.opener.open(
                    f"{self.base_url}/api/extension/gscu/slots", timeout=3
                ) as response:
                    catalog = json.loads(response.read().decode("utf-8"))
                self.assertEqual(
                    [item["slot"] for item in catalog["slots"]], [3, 5, 12]
                )

                delete_request = urllib.request.Request(
                    f"{self.base_url}/api/extension/gscu/slots/3",
                    method="DELETE",
                )
                with self.opener.open(delete_request, timeout=3) as response:
                    deleted = json.loads(response.read().decode("utf-8"))
                self.assertTrue(deleted["removed"])
                self.assertFalse(deleted["device_signal_erased"])

                erase_request = urllib.request.Request(
                    f"{self.base_url}/api/extension/gscu/erase-all",
                    method="POST",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps({"confirm": "ERASE_ALL"}).encode("utf-8"),
                )
                with self.opener.open(erase_request, timeout=3) as response:
                    erased = json.loads(response.read().decode("utf-8"))
                self.assertTrue(erased["catalog_cleared"])
                self.assertEqual(erased["catalog_removed_count"], 2)

                with self.opener.open(
                    f"{self.base_url}/api/extension/gscu/slots", timeout=3
                ) as response:
                    empty_catalog = json.loads(response.read().decode("utf-8"))
                self.assertEqual(empty_catalog["slots"], [])
            finally:
                SeatDemoHandler.extension = original_extension

    def test_main_request_routes_live_nfc_peripheral_to_gscu(self) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        SeatDemoHandler.extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name, "main-nfc-route.sqlite3"
            ),
        )
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "revision": 1,
                "reader": {"online": True},
                "card": {"action": "present", "uid": "1A53DD95"},
            }
        )
        catalog_request = urllib.request.Request(
            f"{self.base_url}/api/extension/display-sync",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "peripherals": [
                        {
                            "id": "cap-third-party-light-on",
                            "status": "learned",
                            "device_id": "third-party-light",
                            "device_name": "第三方氛围灯",
                            "card_uid": "1A53DD95",
                            "action": "开启",
                            "aliases": ["开灯"],
                            "slot": 9,
                        }
                    ]
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "打开第三方氛围灯",
                    "provider": "local",
                    "conversation_id": "conv-main-nfc-extension",
                    "execution_mode": "hardware",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        try:
            with self.opener.open(catalog_request, timeout=3) as response:
                self.assertEqual(response.status, 200)
            with patch.object(
                self.service,
                "process",
                side_effect=AssertionError("vehicle orchestration must not run"),
            ):
                with self.opener.open(request, timeout=3) as response:
                    result = json.loads(response.read().decode("utf-8"))
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["orchestration_route"], "nfc_peripheral")
        self.assertEqual(result["peripheral"]["name"], "第三方氛围灯")
        self.assertEqual(result["peripheral"]["slot"], 9)
        operation = result["execution_report"]["spec"]["results"][0]
        self.assertEqual(operation["domain"], "self_extension")
        self.assertTrue(operation["telemetry"]["gateway_acknowledged"])

    def test_mode_creation_cannot_be_claimed_by_live_nfc_peripheral(self) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        with tempfile.TemporaryDirectory() as directory:
            gateway = SimulationGateway()
            extension = SelfExtensionService(
                gateway,
                catalog_db_path=Path(directory, "mode-create-peripheral.sqlite3"),
            )
            extension.sync_peripheral_catalog(
                [
                    {
                        "id": "ir-slot:flashing-toy:1",
                        "status": "learned",
                        "enabled": True,
                        "device_id": "flashing-toy",
                        "device_name": "闪灯玩具",
                        "object_names": ["闪灯玩具"],
                        "card_uid": "2CC8B103",
                        "action": "打开",
                        "action_names": ["打开", "开启"],
                        "risk": "low",
                        "slot": 1,
                    }
                ]
            )
            nfc_snapshot = {
                "revision": 1,
                "reader": {"online": True},
                "card": {"action": "present", "uid": "2CC8B103"},
            }
            utterance = "创建一个打开闪灯玩具模式"
            partition = partition_active_peripheral_intent(
                utterance,
                extension.list_peripheral_capabilities(),
                nfc_snapshot,
            )
            self.assertTrue(partition["peripheral_actions"])

            SeatDemoHandler.extension = extension
            SeatDemoHandler.nfc = SimpleNamespace(snapshot=lambda: nfc_snapshot)
            request = urllib.request.Request(
                f"{self.base_url}/api/requests/preview",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps(
                    {
                        "utterance": utterance,
                        "provider": "local",
                        "conversation_id": "conv-mode-create-peripheral-boundary",
                        "execution_mode": "hardware",
                        "client_type": "web_console_assistant",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )
            try:
                with patch.object(gateway, "send", wraps=gateway.send) as send:
                    with self.opener.open(request, timeout=3) as response:
                        result = json.loads(response.read().decode("utf-8"))
                self.assertEqual(send.call_count, 0)
            finally:
                extension.close()
                SeatDemoHandler.extension = original_extension
                SeatDemoHandler.nfc = original_nfc

        self.assertEqual(result["execution_mode"], "dry_run")
        self.assertFalse(result["hardware_execution_allowed"])
        self.assertNotEqual(result.get("orchestration_route"), "nfc_peripheral")
        self.assertNotIn("peripheral", result)

    def test_all_mode_configuration_phrases_are_control_plane_only(self) -> None:
        for utterance in (
            "创建一个打开闪灯玩具模式",
            "固化为打开闪灯玩具模式",
            "修改打开闪灯玩具模式，空调调整到24度",
            "恢复打开闪灯玩具模式初始配置",
        ):
            with self.subTest(utterance=utterance):
                self.assertTrue(
                    self.service.is_mode_profile_configuration_request(utterance)
                )
        self.assertFalse(
            self.service.is_mode_profile_configuration_request("进入小憩模式")
        )

    def test_speech_routes_exact_asr_text_to_learned_peripheral_and_ui_config(
        self,
    ) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        original_speech = SeatDemoHandler.speech
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name,
                "speech-nfc-route.sqlite3",
            ),
        )
        extension.sync_peripheral_catalog(
            [
                {
                    "id": "cap-third-party-light-on",
                    "status": "learned",
                    "enabled": True,
                    "device_id": "third-party-light",
                    "device_name": "第三方氛围灯",
                    "object_names": ["第三方氛围灯"],
                    "card_uid": "1A53DD95",
                    "action": "开启",
                    "action_names": ["打开", "开启"],
                    "slot": 9,
                }
            ]
        )
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "revision": 1,
                "reader": {"online": True},
                "card": {"action": "present", "uid": "1A53DD95"},
            }
        )
        understanding = UnderstandingTranscriptProcessor(REPOSITORY_ROOT).process(
            "打开第三方氛围灯"
        ).to_dict()
        understanding["utterance_id"] = "e2e-speech-peripheral"

        class SpeechStub:
            @staticmethod
            def transcribe_wav(
                content: bytes,
                *,
                utterance_id: str | None = None,
                require_wake_word: bool = False,
            ) -> dict[str, object]:
                if require_wake_word:
                    raise AssertionError("manual transcription unexpectedly required wake word")
                if content != b"RIFF-speech-peripheral":
                    raise AssertionError("unexpected audio payload")
                return {**understanding, "utterance_id": utterance_id}

            @staticmethod
            def cached_result(utterance_id: str) -> dict[str, object]:
                if utterance_id != "e2e-speech-peripheral":
                    raise AssertionError("unexpected speech utterance id")
                return dict(understanding)

        SeatDemoHandler.speech = SpeechStub()
        try:
            transcription_request = urllib.request.Request(
                f"{self.base_url}/api/speech/transcribe",
                method="POST",
                headers={
                    "Content-Type": "audio/wav",
                    "X-Speech-Utterance-Id": "e2e-speech-peripheral",
                },
                data=b"RIFF-speech-peripheral",
            )
            with self.opener.open(transcription_request, timeout=3) as response:
                transcription = json.loads(response.read().decode("utf-8"))

            wrong_target_request = urllib.request.Request(
                f"{self.base_url}/api/requests",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps(
                    {
                        "utterance": understanding["clean_text"],
                        "speech_utterance_id": "e2e-speech-peripheral",
                        "provider": "local",
                        "conversation_id": "conv-e2e-speech-wrong-target",
                        "execution_mode": "dry_run",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )
            with self.assertRaises(urllib.error.HTTPError) as rejected:
                self.opener.open(wrong_target_request, timeout=3)
            self.assertEqual(rejected.exception.code, 409)
            rejected_payload = json.loads(
                rejected.exception.read().decode("utf-8")
            )
            self.assertEqual(rejected_payload["error"], "speech_text_changed")

            control_request = urllib.request.Request(
                f"{self.base_url}/api/requests",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps(
                    {
                        "utterance": transcription["routing_text"],
                        "speech_utterance_id": "e2e-speech-peripheral",
                        "provider": "openai",
                        "conversation_id": "conv-e2e-speech-peripheral",
                        "execution_mode": "dry_run",
                        "client_type": "web_console",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )
            with self.opener.open(control_request, timeout=3) as response:
                result = json.loads(response.read().decode("utf-8"))
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc
            SeatDemoHandler.speech = original_speech

        self.assertEqual(transcription["routing_text"], "打开第三方氛围灯")
        self.assertTrue(
            transcription["peripheral_routing"]["execution_eligible"]
        )
        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["orchestration_route"], "nfc_peripheral")
        self.assertEqual(set(result["function_configs"]), {"self_extension"})
        operation = result["function_configs"]["self_extension"]["spec"][
            "operations"
        ][0]
        self.assertEqual(operation["capability"], "infrared.emit")
        self.assertEqual(operation["parameters"]["slot"], 9)
        self.assertEqual(operation["parse_source"], "local_rule")
        self.assertEqual(
            result["speech_understanding"]["routing"],
            "validated_peripheral_text",
        )

    def test_main_request_combines_explicit_ir_clause_with_existing_can_parser(self) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name, "mixed-nfc-can.sqlite3"
            ),
        )
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "revision": 1,
                "reader": {"online": True},
                "card": {"action": "present", "uid": "2CC8B103"},
            }
        )
        catalog_request = urllib.request.Request(
            f"{self.base_url}/api/extension/display-sync",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "peripherals": [
                        {
                            "id": "toy-001-light-on",
                            "status": "learned",
                            "enabled": True,
                            "device_id": "toy-001",
                            "device_name": "闪灯玩具",
                            "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                            "card_uid": "2CC8B103",
                            "action": "开启",
                            "action_names": ["打开", "开启", "启动"],
                            "risk": "low",
                            "slot": 1,
                        },
                        {
                            "id": "toy-001-light-red",
                            "status": "learned",
                            "enabled": True,
                            "device_id": "toy-001",
                            "device_name": "闪灯玩具",
                            "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                            "card_uid": "2CC8B103",
                            "action": "切换到红色",
                            "action_names": ["切换到红色", "切换为红色", "设置为红色"],
                            "risk": "low",
                            "slot": 3,
                        },
                    ]
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Accept": "application/x-ndjson",
            },
            data=json.dumps(
                {
                    "utterance": "空调温度设置到23度，打开闪烁灯",
                    "provider": "local",
                    "conversation_id": "conv-mixed-nfc-can",
                    "execution_mode": "dry_run",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        try:
            with self.opener.open(catalog_request, timeout=3) as response:
                synced = json.loads(response.read().decode("utf-8"))
            with self.opener.open(request, timeout=3) as response:
                events = [
                    json.loads(line)
                    for line in response.read().decode("utf-8").splitlines()
                    if line.strip()
                ]
                result = events[-1]["payload"]
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        self.assertEqual(synced["learned_count"], 2)
        self.assertEqual(events[-1]["type"], "result")
        self.assertTrue(any(item["type"] == "validated_commands" for item in events))
        self.assertLess(
            next(
                index
                for index, item in enumerate(events)
                if item["type"] == "validated_commands"
            ),
            len(events) - 1,
        )
        validated_payloads = [
            item["payload"]
            for item in events
            if item["type"] == "validated_commands"
        ]
        self.assertTrue(validated_payloads)
        self.assertTrue(
            all(
                {"native_climate", "self_extension"}.issubset(
                    payload["function_configs"]
                )
                for payload in validated_payloads
            )
        )
        first_validated = validated_payloads[0]
        self.assertTrue(
            {"native_climate", "self_extension"}.issubset(
                first_validated["command_plans"]
            )
        )
        validated_operations = [
            operation
            for configuration in first_validated["function_configs"].values()
            for operation in configuration["spec"].get("operations", [])
        ]
        self.assertEqual(
            {operation["capability"] for operation in validated_operations},
            {"climate.set_temperature", "infrared.emit"},
        )
        self.assertTrue(
            all(
                operation.get("parse_source") == "local_rule"
                for operation in validated_operations
            )
        )
        validated_commands = [
            command
            for plan in first_validated["command_plans"].values()
            for command in plan["spec"].get("commands", [])
        ]
        climate_command = next(
            command
            for command in validated_commands
            if command.get("signals", {}).get("debug_command")
            == "ac temp driver 23"
        )
        self.assertEqual(climate_command["signals"]["frame_id"], "0x1DE")
        self.assertEqual(
            climate_command["signals"]["data_hex"],
            "00 00 00 00 17 00 00 00",
        )
        infrared_command = next(
            command
            for command in validated_commands
            if command.get("message") == "infrared.emit"
        )
        self.assertEqual(infrared_command["signals"]["slot"], 1)
        assistant_projection = CodexAssistantService._validated_control_payload(
            first_validated
        )
        self.assertIsNotNone(assistant_projection)
        projected_commands = assistant_projection["validated_control_commands"]
        projected_by_capability = {
            item["operation"]["capability"]: item for item in projected_commands
        }
        self.assertEqual(
            projected_by_capability["climate.set_temperature"]["actual_command"],
            "ac temp driver 23",
        )
        self.assertEqual(
            projected_by_capability["climate.set_temperature"]["frames"][0][
                "value"
            ],
            "0x1DE · 00 00 00 00 17 00 00 00",
        )
        self.assertEqual(
            projected_by_capability["infrared.emit"]["actual_command"],
            "infrared emit 1",
        )
        serialized_projection = json.dumps(
            assistant_projection, ensure_ascii=False
        )
        self.assertNotIn("command_plans", serialized_projection)
        self.assertNotIn("data_hex", serialized_projection)
        compile_trace = next(
            event
            for event in first_validated["trace_events"]
            if event["stage"] == "compile"
        )
        validated_traces = first_validated["trace_events"][
            : first_validated["trace_events"].index(compile_trace) + 1
        ]
        required_validation_stages = {
            "intent_parse",
            "context_resolve",
            "domain_route",
            "function_generate",
            "schema_validate",
            "compile",
        }
        actual_validation_stages = {
            event["stage"] for event in validated_traces
        }
        self.assertTrue(
            required_validation_stages.issubset(actual_validation_stages),
            sorted(actual_validation_stages),
        )
        self.assertTrue(
            all(
                isinstance(event.get("duration_ms"), int)
                and event["duration_ms"] >= 0
                for event in validated_traces
            )
        )
        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["orchestration_route"], "unified_can_infrared")
        self.assertEqual(
            result["provider_used"], "local_rules+infrared_local_rules"
        )
        self.assertEqual([item["slot"] for item in result["infrared_actions"]], [1])
        self.assertEqual(
            result["infrared_actions"][0]["parse_source"], "local_rule"
        )
        self.assertIn("self_extension", result["function_configs"])
        self.assertTrue(
            any(key != "self_extension" for key in result["function_configs"])
        )
        self.assertTrue(
            any(
                item.get("domain") == "self_extension"
                for item in result["operation_results"]
            )
        )

    def test_hardware_can_failure_keeps_infrared_in_validated_command_event(
        self,
    ) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name,
                "mixed-hardware-validation.sqlite3",
            ),
        )
        extension.sync_peripheral_catalog(
            [
                {
                    "id": "ir-slot:tony-light-001:1",
                    "status": "learned",
                    "enabled": True,
                    "device_id": "tony-light-001",
                    "device_name": "闪灯玩具",
                    "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                    "card_uid": "2CC8B103",
                    "action": "打开",
                    "action_names": ["打开", "开启", "启动"],
                    "risk": "low",
                    "slot": 1,
                },
                {
                    "id": "ir-slot:tony-light-001:4",
                    "status": "learned",
                    "enabled": True,
                    "device_id": "tony-light-001",
                    "device_name": "闪灯玩具",
                    "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                    "card_uid": "2CC8B103",
                    "action": "切换到绿色",
                    "action_names": ["切换到绿色", "切换为绿色", "设置为绿色"],
                    "risk": "low",
                    "slot": 4,
                },
            ]
        )
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "reader": {"online": True},
                "card": {"action": "present", "uid": "2CC8B103"},
            }
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Accept": "application/x-ndjson",
            },
            data=json.dumps(
                {
                    "utterance": (
                        "打开空调，温度设置为24度，风速调整为3档，"
                        "打开遮阳帘，开启闪光灯，颜色设置为绿色"
                    ),
                    "provider": "local",
                    "conversation_id": "conv-hardware-mixed-validation",
                    "execution_mode": "hardware",
                    "control_transport": "wired_can",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )

        def fail_can_after_validation(*_args, **kwargs):
            callback = kwargs.get("_on_commands_validated")
            self.assertTrue(callable(callback))
            callback(
                {
                    "request_id": "req-can-validation-only",
                    "utterance": "打开空调，温度设置为24度，风速调整为3档，打开遮阳帘",
                    "status": "processing",
                    "execution_mode": "hardware",
                    "control_transport": "wired_can",
                    "provider_used": "local_rules",
                    "decision_summary": [],
                    "function_configs": {
                        "native_climate": {
                            "spec": {
                                "operations": [
                                    {
                                        "id": "vehicle_set_power_1",
                                        "capability": "climate.set_power",
                                        "target": {"zone": "vehicle"},
                                        "parameters": {"state": "on"},
                                        "parse_source": "local_rule",
                                    }
                                ]
                            }
                        }
                    },
                    "command_plans": {
                        "native_climate": {"spec": {"commands": []}}
                    },
                    "operation_results": [],
                    "execution_report": None,
                }
            )
            raise CanGatewayError("simulated CAN transport failure")

        try:
            with patch.object(
                self.service,
                "process",
                side_effect=fail_can_after_validation,
            ):
                with self.opener.open(request, timeout=3) as response:
                    events = [
                        json.loads(line)
                        for line in response.read().decode("utf-8").splitlines()
                        if line.strip()
                    ]
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        self.assertEqual(events[-1]["type"], "error")
        self.assertEqual(events[-1]["error"], "can_transport_failed")
        validated = next(
            item["payload"]
            for item in events
            if item["type"] == "validated_commands"
        )
        self.assertEqual(validated["status"], "processing")
        self.assertEqual(
            set(validated["function_configs"]),
            {"native_climate", "self_extension"},
        )
        self.assertEqual(
            [item["slot"] for item in validated["infrared_actions"]],
            [1, 4],
        )

    def test_disabled_nfc_ir_command_does_not_discard_allowed_can_and_ir(
        self,
    ) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name,
                "mixed-disabled-nfc-command.sqlite3",
            ),
        )
        extension.sync_peripheral_catalog(
            [
                {
                    "id": "ir-slot:tony-light-001:1",
                    "status": "learned",
                    "enabled": True,
                    "device_id": "tony-light-001",
                    "device_name": "闪灯玩具",
                    "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                    "card_uid": "2CC8B103",
                    "action": "打开",
                    "action_names": ["打开", "开启", "启动"],
                    "risk": "low",
                    "slot": 1,
                },
                {
                    "id": "ir-slot:tony-light-001:5",
                    "status": "learned",
                    "enabled": False,
                    "device_id": "tony-light-001",
                    "device_name": "闪灯玩具",
                    "object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
                    "card_uid": "2CC8B103",
                    "action": "切换到蓝色",
                    "action_names": ["切换到蓝色", "切换为蓝色", "设置为蓝色"],
                    "risk": "low",
                    "slot": 5,
                },
            ]
        )
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "revision": 1,
                "reader": {"online": True},
                "card": {"action": "present", "uid": "2CC8B103"},
            }
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Accept": "application/x-ndjson",
            },
            data=json.dumps(
                {
                    "utterance": (
                        "打开空调，温度设置为24度，风速调整为3档，"
                        "开启闪光灯，颜色设置为蓝色"
                    ),
                    "provider": "openai",
                    "conversation_id": "conv-disabled-nfc-command",
                    "execution_mode": "dry_run",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        try:
            with patch.object(
                extension,
                "plan",
                side_effect=AssertionError(
                    "a disabled exact command must not be sent to the cloud"
                ),
            ):
                with self.opener.open(request, timeout=3) as response:
                    events = [
                        json.loads(line)
                        for line in response.read().decode("utf-8").splitlines()
                        if line.strip()
                    ]
                    result = events[-1]["payload"]
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        self.assertEqual(events[-1]["type"], "result")
        self.assertEqual(result["status"], "partially_succeeded")
        self.assertEqual(result["orchestration_route"], "unified_can_infrared")
        self.assertEqual(
            [item["slot"] for item in result["infrared_actions"]], [1]
        )
        self.assertEqual(
            [item["slot"] for item in result["denied_peripheral_actions"]],
            [5],
        )
        self.assertEqual(
            result["denied_peripheral_actions"][0]["reason"],
            "nfc_command_disabled",
        )
        debug_commands = [
            command["signals"]["debug_command"]
            for plan in result["command_plans"].values()
            for command in plan["spec"].get("commands", [])
            if command.get("signals", {}).get("debug_command")
        ]
        self.assertEqual(
            debug_commands,
            ["ac on", "ac temp driver 24", "ac fan 3"],
        )
        infrared_slots = [
            command["signals"]["slot"]
            for command in result["command_plans"]["self_extension"]["spec"][
                "commands"
            ]
        ]
        self.assertEqual(infrared_slots, [1])

    def test_generic_vehicle_light_text_never_routes_to_ir_alias(self) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name, "vehicle-light-not-ir.sqlite3"
            ),
        )
        with self.assertRaisesRegex(GatewayError, "车端设备保留名称"):
            extension.sync_peripheral_catalog(
                [
                    {
                        "id": "toy-001-light-on",
                        "status": "learned",
                        "device_id": "toy-001",
                        "device_name": "玩具 001",
                        "device_aliases": ["氛围灯"],
                        "card_uid": "2CC8B103",
                        "action": "开启闪灯",
                        "aliases": ["打开"],
                        "risk": "low",
                        "slot": 1,
                    }
                ]
            )
        self.assertEqual(extension.list_peripheral_capabilities(), [])
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "reader": {"online": True},
                "card": {"action": "present", "uid": "2CC8B103"},
            }
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "打开氛围灯，空调24度",
                    "provider": "local",
                    "conversation_id": "conv-vehicle-light-not-ir",
                    "execution_mode": "dry_run",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        try:
            with self.opener.open(request, timeout=3) as response:
                result = json.loads(response.read().decode("utf-8"))
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        self.assertNotIn("infrared_actions", result)
        self.assertNotEqual(result.get("orchestration_route"), "unified_can_infrared")
        self.assertNotIn("self_extension", result["function_configs"])
        self.assertTrue(result["function_configs"])

    def test_mixed_preview_reports_cloud_selected_infrared_capability(self) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name,
                "mixed-cloud-nfc-can.sqlite3",
            ),
        )
        extension.sync_peripheral_catalog(
            [
                {
                    "id": "toy-001-light-red",
                    "status": "learned",
                    "enabled": True,
                    "device_id": "toy-001",
                    "device_name": "闪灯玩具",
                    "object_names": ["闪灯玩具", "闪烁灯"],
                    "card_uid": "2CC8B103",
                    "action": "切换到红色",
                    "action_names": ["切换到红色", "切换为红色"],
                    "slot": 3,
                }
            ]
        )
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "reader": {"online": True},
                "card": {"action": "present", "uid": "2CC8B103"},
            }
        )
        request = urllib.request.Request(
            f"{self.base_url}/api/requests/preview",
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Accept": "application/x-ndjson",
            },
            data=json.dumps(
                {
                    "utterance": "把闪烁灯调成喜庆颜色，打开空调",
                    "provider": "openai",
                    "conversation_id": "conv-mixed-cloud-ir",
                    "execution_mode": "dry_run",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        plan_result = {
            "model": "gpt-5.6-luna",
            "summary": "喜庆颜色对应已学习的红色动作。",
            "needs_confirmation": False,
            "steps": [
                {"kind": "peripheral", "capability_id": "toy-001-light-red"}
            ],
        }
        try:
            with patch.object(extension, "plan", return_value=plan_result) as planner:
                with self.opener.open(request, timeout=3) as response:
                    events = [
                        json.loads(line)
                        for line in response.read().decode("utf-8").splitlines()
                        if line.strip()
                    ]
                    result = events[-1]["payload"]
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        planner.assert_called_once()
        self.assertEqual(result["orchestration_route"], "unified_can_infrared")
        self.assertIn("gpt:gpt-5.6-luna", result["provider_used"])
        self.assertEqual(
            result["infrared_actions"][0]["parse_source"], "cloud_model"
        )
        self.assertFalse(result["supplement_required"])

    def test_main_request_executes_gpt_nfc_sequence_without_vehicle_can(self) -> None:
        original_extension = SeatDemoHandler.extension
        original_nfc = SeatDemoHandler.nfc
        extension = SelfExtensionService(
            SimulationGateway(),
            catalog_db_path=Path(
                self.extension_temporary_directory.name, "main-nfc-sequence.sqlite3"
            ),
        )
        SeatDemoHandler.extension = extension
        SeatDemoHandler.nfc = SimpleNamespace(
            snapshot=lambda: {
                "revision": 1,
                "reader": {"online": True},
                "card": {"action": "present", "uid": "1A53DD95"},
            }
        )
        capabilities = [
            {
                "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),
                ("red", "红色灯光", 2),
                ("off", "关闭灯光", 3),
            )
        ]
        catalog_request = urllib.request.Request(
            f"{self.base_url}/api/extension/display-sync",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {"peripherals": capabilities}, ensure_ascii=False
            ).encode("utf-8"),
        )
        utterance = "打开第三方氛围灯，红灯闪烁5s后，最后关闭第三方氛围灯"
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": utterance,
                    "provider": "local",
                    "conversation_id": "conv-main-nfc-sequence",
                    "execution_mode": "hardware",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        plan_result = {
            "model": "gpt-5.6-sol",
            "summary": "按顺序切换第三方氛围灯",
            "needs_confirmation": False,
            "steps": [
                {"kind": "peripheral", "capability_id": "on"},
                {"kind": "peripheral", "capability_id": "red"},
                {"kind": "delay", "delay_ms": 5000},
                {"kind": "peripheral", "capability_id": "off"},
            ],
        }
        try:
            with self.opener.open(catalog_request, timeout=3) as response:
                self.assertEqual(response.status, 200)
            with patch.object(
                self.service,
                "process",
                side_effect=AssertionError("vehicle orchestration must not run"),
            ), patch.object(
                extension, "plan", return_value=plan_result
            ) as planner, patch(
                "cognitive_agent.application.self_extension.time.sleep"
            ):
                with self.opener.open(request, timeout=3) as response:
                    result = json.loads(response.read().decode("utf-8"))
        finally:
            SeatDemoHandler.extension = original_extension
            SeatDemoHandler.nfc = original_nfc

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(
            result["orchestration_route"], "nfc_peripheral_sequence"
        )
        self.assertEqual(result["sequence"]["step_count"], 4)
        self.assertEqual(result["sequence"]["action_count"], 3)
        self.assertEqual(result["sequence"]["emit_count"], 3)
        self.assertEqual(result["sequence"]["steady_blink_fallbacks"], 1)
        self.assertEqual(result["sequence"]["total_delay_ms"], 5000)
        self.assertEqual(len(result["operation_results"]), 4)
        planner.assert_called_once()
        self.assertEqual(planner.call_args.args[0]["utterance"], utterance)

    def test_post_request_returns_complete_dry_run(self) -> None:
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "副驾有点热，有点闷",
                    "provider": "local",
                    "conversation_id": "conv-e2e-browser",
                    "vehicle_id": "denza_n7",
                }
            ).encode("utf-8"),
        )
        with self.opener.open(request, timeout=3) as response:
            result = json.loads(response.read().decode("utf-8"))

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["function_config"]["kind"], "FunctionConfig")
        self.assertEqual(result["command_plan"]["kind"], "CommandPlan")
        self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run")
        self.assertEqual(result["conversation_id"], "conv-e2e-browser")
        self.assertEqual(result["vehicle"]["id"], "denza_n7")
        self.assertEqual(result["actuator_state"]["passenger"]["ventilation"]["level"], 1)

    def test_relative_backrest_phrase_uses_current_conversation_position(self) -> None:
        conversation_id = "conv-e2e-relative-backrest"
        for utterance in ("副驾靠背角度调到130度", "副驾靠背向前移动10度"):
            request = urllib.request.Request(
                f"{self.base_url}/api/requests",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps(
                    {
                        "utterance": utterance,
                        "provider": "local",
                        "conversation_id": conversation_id,
                        "vehicle_id": "denza_n7",
                        "execution_mode": "dry_run",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )
            with self.opener.open(request, timeout=3) as response:
                result = json.loads(response.read().decode("utf-8"))

        self.assertEqual(result["status"], "succeeded")
        operation = result["function_config"]["spec"]["operations"][0]
        self.assertEqual(operation["capability"], "seat.set_backrest_recline")
        self.assertEqual(operation["parameters"]["position_pct"], 36)
        self.assertEqual(operation["parameters"]["angle_deg"], 120)
        self.assertEqual(operation["motion"]["direction"], "decrease")
        self.assertEqual(operation["motion"]["delta_angle_deg"], 10.0)

    def test_timed_and_oversized_backrest_requests_use_endpoint_fallbacks(self) -> None:
        conversation_id = "conv-e2e-timed-endpoint-fallbacks"

        def submit(utterance: str) -> dict:
            request = urllib.request.Request(
                f"{self.base_url}/api/requests",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps(
                    {
                        "utterance": utterance,
                        "provider": "local",
                        "conversation_id": conversation_id,
                        "vehicle_id": "denza_n7",
                        "execution_mode": "dry_run",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )
            with self.opener.open(request, timeout=3) as response:
                return json.loads(response.read().decode("utf-8"))

        submit("副驾靠背归零")
        timed = submit("副驾靠背向后移动70秒")

        self.assertEqual(timed["status"], "succeeded")
        timed_operation = timed["function_config"]["spec"]["operations"][0]
        self.assertEqual(timed_operation["parameters"]["position_pct"], 100)
        self.assertEqual(timed_operation["motion"]["mode"], "timed")
        self.assertEqual(timed_operation["motion"]["requested_duration_ms"], 70_000)
        self.assertEqual(timed_operation["motion"]["applied_duration_ms"], 30_000)
        self.assertTrue(timed_operation["motion"]["duration_clamped"])
        self.assertIn("最大允许时长为30秒", "\n".join(timed["decision_summary"]))

        submit("副驾靠背角度调到130度")
        overflow = submit("副驾靠背向后移动70度")

        self.assertEqual(overflow["status"], "succeeded")
        overflow_operation = overflow["function_config"]["spec"]["operations"][0]
        self.assertEqual(overflow_operation["parameters"]["position_pct"], 100)
        self.assertEqual(overflow_operation["parameters"]["angle_deg"], 163)
        self.assertTrue(overflow_operation["motion"]["range_clamped"])
        self.assertEqual(
            overflow_operation["motion"]["requested_delta_angle_deg"], 70
        )
        self.assertIn("备用目标", "\n".join(overflow["decision_summary"]))

    def test_open_zero_gravity_phrase_matches_complete_mode(self) -> None:
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "打开零重力模式",
                    "provider": "local",
                    "conversation_id": "conv-e2e-open-zero-gravity",
                    "vehicle_id": "denza_n7",
                    "execution_mode": "dry_run",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )
        with self.opener.open(request, timeout=3) as response:
            result = json.loads(response.read().decode("utf-8"))

        self.assertEqual(result["status"], "succeeded")
        self.assertIn("zero_gravity", {item["id"] for item in result["matched_modes"]})
        operations = result["function_config"]["spec"]["operations"]
        self.assertEqual(len(operations), 4)
        self.assertEqual(
            {
                operation["capability"]: operation["parameters"]["position_pct"]
                for operation in operations
            },
            {
                "seat.set_longitudinal_position": 100,
                "seat.set_cushion_tilt": 100,
                "seat.set_backrest_recline": 81,
                "seat.set_legrest_height": 94,
            },
        )
        self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run")

    def test_hardware_state_endpoint_returns_physical_passenger_snapshot(self) -> None:
        synchronized = {
            "status": "succeeded",
            "conversation_id": "conv-e2e-physical-state",
            "source": "physical_can",
            "complete": True,
            "missing": [],
            "actuator_state": {
                "passenger": {
                    "position": {
                        "longitudinal": {"position_pct": 0},
                        "cushion_tilt": {"position_pct": 0},
                        "backrest_recline": {"position_pct": 0},
                        "legrest_height": {"position_pct": 0},
                    }
                }
            },
        }
        with patch.object(
            self.service,
            "synchronize_hardware_state",
            return_value=synchronized,
        ) as synchronize:
            with self.opener.open(
                f"{self.base_url}/api/hardware/seat-state?"
                "conversation_id=conv-e2e-physical-state&vehicle_id=denza_n7",
                timeout=3,
            ) as response:
                payload = json.loads(response.read().decode("utf-8"))

        self.assertEqual(payload["source"], "physical_can")
        self.assertEqual(
            payload["actuator_state"]["passenger"]["position"]["longitudinal"]["position_pct"],
            0,
        )
        synchronize.assert_called_once_with(
            "conv-e2e-physical-state",
            "denza_n7",
            control_transport="wired_can",
            hardware_client_id=None,
            hardware_access_bypass=True,
        )

    def test_wireless_passenger_snapshot_uses_mqtt_can0(self) -> None:
        physical_state = {
            "position": {
                "longitudinal": {
                    "position_pct": 42,
                    "position_valid": True,
                    "motion_state": "stopped",
                    "fault": False,
                }
            },
            "captured_at": "2026-08-18T00:00:00Z",
            "complete": True,
            "missing": [],
        }
        executor = SimpleNamespace(read_physical_state=lambda: dict(physical_state))
        with (
            patch.object(
                self.service,
                "vehicle_mqtt_settings",
                SimpleNamespace(state_configured=True),
            ),
            patch.object(
                self.service._vehicle_mqtt_transport,
                "bus",
                return_value=object(),
            ) as mqtt_bus,
            patch.object(
                self.service,
                "_new_hardware_executor",
                return_value=executor,
            ) as new_executor,
            patch.object(
                self.service.hardware_execution,
                "reserve",
                side_effect=AssertionError(
                    "wireless cached state must not reserve the physical bench"
                ),
            ),
        ):
            payload = self.service.synchronize_hardware_state(
                "conv-e2e-wireless-seat-state",
                "denza_n7",
                control_transport="wireless_mqtt",
                hardware_access_bypass=True,
            )
            bus_factory = new_executor.call_args.args[0]
            bus_factory()
            mqtt_bus.assert_called_once_with(channel=0, replay_latest=True)
            self.assertTrue(new_executor.call_args.kwargs["remote_transport"])
        self.assertEqual(payload["source"], "wireless_mqtt_ch0")
        self.assertEqual(payload["control_transport"], "wireless_mqtt")
        self.assertEqual(
            payload["actuator_state"]["passenger"]["position"]["longitudinal"]["position_pct"],
            42,
        )

    def test_wireless_vehicle_snapshot_does_not_reserve_physical_bench(self) -> None:
        snapshot = SimpleNamespace(
            feedback_states=(),
            observed_ids=frozenset(),
            captured_at="2026-08-18T00:00:00Z",
            capture_sequence=17,
        )
        with (
            patch.object(
                self.service,
                "vehicle_mqtt_settings",
                SimpleNamespace(state_configured=True),
            ),
            patch.object(
                self.service._vehicle_mqtt_transport,
                "snapshot",
                return_value=snapshot,
            ),
            patch.object(
                self.service.hardware_execution,
                "reserve",
                side_effect=AssertionError(
                    "wireless cached state must not reserve the physical bench"
                ),
            ),
        ):
            payload = self.service.synchronize_vehicle_state(
                "conv-e2e-wireless-vehicle-state",
                "denza_n7",
                control_transport="wireless_mqtt",
                hardware_access_bypass=True,
            )

        self.assertEqual(payload["source"], "wireless_mqtt_ch1")
        self.assertEqual(payload["capture_sequence"], 17)

    def test_gpt_http_failure_falls_back_without_returning_502(self) -> None:
        original_api_key = self.service.openai.api_key
        self.service.openai.api_key = "test-key"
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "打开副驾座椅通风，同时播放音乐",
                    "provider": "openai",
                    "conversation_id": "conv-e2e-gpt-fallback",
                    "execution_mode": "dry_run",
                }
            ).encode("utf-8"),
        )
        try:
            with patch.object(
                self.service.openai,
                "interpret",
                side_effect=OpenAIIntentError("invalid_api_key"),
            ):
                with self.opener.open(request, timeout=3) as response:
                    result = json.loads(response.read().decode("utf-8"))
                    self.assertEqual(response.status, 201)
        finally:
            self.service.openai.api_key = original_api_key

        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(result["provider_used"], "local_rules_fallback")
        self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run")

    def test_two_stage_http_api_returns_local_commands_before_cloud_supplement(
        self,
    ) -> None:
        original_api_key = self.service.openai.api_key
        self.service.openai.api_key = "two-stage-http-test-key"
        preview_request = urllib.request.Request(
            f"{self.base_url}/api/requests/preview",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "空调温度设置为24度，播放音乐",
                    "provider": "openai",
                    "conversation_id": "conv-e2e-two-stage-http",
                    "execution_mode": "dry_run",
                    "reasoning_effort": "medium",
                    "client_type": "web_console",
                }
            ).encode("utf-8"),
        )
        model_output = {
            "detected_domains": ["ambient_light"],
            "requests": [
                {
                    "domain": "ambient_light",
                    "capability": "ambient_light.set_power",
                    "target_scope": "vehicle",
                    "operation": "set",
                    "duration_ms": None,
                    "parameters": [{"name": "state", "value": "on"}],
                    "reason": "补充播放场景氛围灯。",
                }
            ],
            "ambiguities": [],
            "decision_summary": ["仅补充本地未解析片段。"],
            "confidence": 0.9,
        }
        try:
            with patch.object(
                self.service.openai,
                "interpret",
                side_effect=AssertionError("preview must not invoke GPT"),
            ):
                with self.opener.open(preview_request, timeout=3) as response:
                    preview = json.loads(response.read().decode("utf-8"))
            self.assertEqual(response.status, 201)
            local_operation = next(
                operation
                for config in preview["function_configs"].values()
                for operation in config["spec"]["operations"]
            )
            self.assertEqual(local_operation["parse_source"], "local_rule")
            self.assertTrue(preview["supplement_required"])

            supplement_request = urllib.request.Request(
                f"{self.base_url}/api/requests/{preview['request_id']}/supplement",
                method="POST",
                headers={"Content-Type": "application/json"},
                data=json.dumps(
                    {
                        "provider": "openai",
                        "reasoning_effort": "medium",
                        "client_type": "web_console",
                    }
                ).encode("utf-8"),
            )
            with patch.object(
                self.service.openai,
                "interpret",
                return_value=model_output,
            ) as interpret:
                with self.opener.open(supplement_request, timeout=3) as response:
                    result = json.loads(response.read().decode("utf-8"))
            self.assertEqual(response.status, 201)
            self.assertEqual(interpret.call_args.args[0], "播放音乐")
        finally:
            self.service.openai.api_key = original_api_key

        self.assertEqual(result["phase"], "two_stage_complete")
        sources = {
            operation["capability"]: operation["parse_source"]
            for config in result["function_configs"].values()
            for operation in config["spec"]["operations"]
        }
        self.assertEqual(sources["climate.set_temperature"], "local_rule")
        self.assertEqual(sources["ambient_light.set_power"], "cloud_supplement")

    def test_preview_streams_validated_commands_before_terminal_result(self) -> None:
        request = urllib.request.Request(
            f"{self.base_url}/api/requests/preview",
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Accept": "application/x-ndjson",
            },
            data=json.dumps(
                {
                    "utterance": "空调24度、风量2档",
                    "provider": "local",
                    "conversation_id": "conv-e2e-command-stream",
                    "execution_mode": "dry_run",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )

        with self.opener.open(request, timeout=3) as response:
            events = [
                json.loads(line)
                for line in response.read().decode("utf-8").splitlines()
                if line.strip()
            ]

        self.assertEqual(
            response.headers.get_content_type(),
            "application/x-ndjson",
        )
        self.assertEqual(
            [event["type"] for event in events],
            ["validated_commands", "result"],
        )
        validated = events[0]["payload"]
        self.assertEqual(validated["status"], "processing")
        self.assertIsNone(validated["execution_report"])
        self.assertTrue(validated["command_plans"])
        self.assertEqual(events[1]["payload"]["status"], "succeeded")

    def test_stream_failure_uses_cli_outcome_and_updates_display_sync_terminal_state(
        self,
    ) -> None:
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Accept": "application/x-ndjson",
            },
            data=json.dumps(
                {
                    "utterance": "打开副驾座椅通风",
                    "provider": "local",
                    "conversation_id": "conv-e2e-stream-can-failure",
                    "execution_mode": "hardware",
                    "client_type": "web_console",
                },
                ensure_ascii=False,
            ).encode("utf-8"),
        )

        with patch.object(
            self.service,
            "process",
            side_effect=CanGatewayError("VCI_OpenDevice failed"),
        ):
            with self.opener.open(request, timeout=3) as response:
                events = [
                    json.loads(line)
                    for line in response.read().decode("utf-8").splitlines()
                    if line.strip()
                ]
        with self.opener.open(
            f"{self.base_url}/api/vehicle-display-sync", timeout=3
        ) as response:
            snapshot = json.loads(response.read().decode("utf-8"))

        self.assertEqual(events, [{
            "type": "error",
            "error": "can_transport_failed",
            "message": "VCI_OpenDevice failed",
            "outcome": "failed_or_blocked",
        }])
        self.assertEqual(snapshot["task"]["status"], "failed")
        self.assertEqual(snapshot["task"]["utterance"], "打开副驾座椅通风")
        self.assertEqual(
            snapshot["task"]["error"],
            {
                "code": "can_transport_failed",
                "message": "VCI_OpenDevice failed",
                "outcome": "failed_or_blocked",
            },
        )

    def test_frontend_suite_covers_every_physical_seat_capability(self) -> None:
        suite_path = (
            REPOSITORY_ROOT
            / "tests"
            / "fixtures"
            / "seat_frontend_capabilities.json"
        )
        suite = json.loads(suite_path.read_text(encoding="utf-8"))
        required = set(suite["required_hardware_capabilities"])
        covered: set[str] = set()
        original_api_key = self.service.openai.api_key
        self.service.openai.api_key = "frontend-suite-test-key"
        try:
            for case in suite["cases"]:
                with self.subTest(case=case["id"]):
                    requests = []
                    for expected in case["operations"]:
                        covered.add(expected["capability"])
                        requests.append(
                            {
                                "domain": "seat",
                                "capability": expected["capability"],
                                "target_scope": expected["target"],
                                "operation": "set",
                                "parameters": [
                                    {"name": name, "value": value}
                                    for name, value in expected["parameters"].items()
                                ],
                                "reason": f"frontend test case {case['id']}",
                            }
                        )
                    model_plan = {
                        "detected_domains": ["seat"],
                        "requests": requests,
                        "ambiguities": [],
                        "decision_summary": [f"执行前端测试用例 {case['id']}。"],
                        "confidence": 0.99,
                    }
                    request = urllib.request.Request(
                        f"{self.base_url}/api/requests",
                        method="POST",
                        headers={"Content-Type": "application/json"},
                        data=json.dumps(
                            {
                                "utterance": case["utterance"],
                                "provider": "openai",
                                "conversation_id": f"conv-frontend-suite-{case['id'].lower()}",
                                "execution_mode": "dry_run",
                                "vehicle_id": "denza_n7",
                            },
                            ensure_ascii=False,
                        ).encode("utf-8"),
                    )
                    with patch.object(
                        self.service.openai,
                        "interpret",
                        return_value=model_plan,
                    ):
                        try:
                            with self.opener.open(request, timeout=3) as response:
                                result = json.loads(response.read().decode("utf-8"))
                        except urllib.error.HTTPError as exc:
                            detail = exc.read().decode("utf-8", errors="replace")
                            self.fail(f"frontend case {case['id']} returned {exc.code}: {detail}")

                    self.assertEqual(result["status"], "succeeded")
                    operations = result["function_config"]["spec"]["operations"]
                    self.assertEqual(len(operations), len(case["operations"]))
                    self.assertEqual(
                        result["execution_report"]["spec"]["mode"], "dry_run"
                    )
                    self.assertEqual(
                        len(result["execution_report"]["spec"]["results"]),
                        len(case["operations"]),
                    )
                    for expected, actual in zip(case["operations"], operations):
                        self.assertEqual(actual["capability"], expected["capability"])
                        self.assertEqual(
                            actual["target"]["instance"], expected["target"]
                        )
                        for name, value in expected["parameters"].items():
                            self.assertEqual(actual["parameters"][name], value)
        finally:
            self.service.openai.api_key = original_api_key

        self.assertEqual(covered, required)
        safety_ids = {item["id"] for item in suite["safety_cases"]}
        self.assertEqual(
            safety_ids,
            {"SAFE-01", "SAFE-02", "SAFE-03", "RECOVERY-01"},
        )

    def test_hardware_request_is_rejected_when_server_gate_is_locked(self) -> None:
        request = urllib.request.Request(
            f"{self.base_url}/api/requests",
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(
                {
                    "utterance": "副驾座椅加热开到一档",
                    "provider": "local",
                    "conversation_id": "conv-e2e-hardware-locked",
                    "execution_mode": "hardware",
                    "approval_token": "must-not-be-persisted",
                }
            ).encode("utf-8"),
        )
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.opener.open(request, timeout=3)

        self.assertEqual(caught.exception.code, 403)
        payload = json.loads(caught.exception.read().decode("utf-8"))
        self.assertEqual(payload["error"], "hardware_safety_gate")
        artifact_text = "\n".join(
            path.read_text(encoding="utf-8")
            for path in Path(self.temporary_directory.name).rglob("*")
            if path.is_file()
        )
        self.assertNotIn("must-not-be-persisted", artifact_text)

    def test_local_decision_point_selects_one_public_terminal_and_can_stop_it(self) -> None:
        client_a = "hwc-e2e-client-a1"
        client_b = "hwc-e2e-client-b2"

        def post(path: str, payload: dict, *, public: bool = False):
            headers = {"Content-Type": "application/json"}
            if public:
                headers["X-Forwarded-For"] = "203.0.113.25"
            request = urllib.request.Request(
                f"{self.base_url}{path}",
                method="POST",
                headers=headers,
                data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            )
            return self.opener.open(request, timeout=3)

        with post(
            "/api/hardware/access/request",
            {"client_id": client_a, "label": "用户电脑 A"},
            public=True,
        ) as response:
            first = json.loads(response.read().decode("utf-8"))
            self.assertEqual(response.status, 202)
        with post(
            "/api/hardware/access/request",
            {"client_id": client_b, "label": "用户电脑 B"},
            public=True,
        ) as response:
            second = json.loads(response.read().decode("utf-8"))
        self.assertEqual(first["queue_position"], 1)
        self.assertEqual(second["queue_position"], 2)

        with self.opener.open(
            f"{self.base_url}/api/local/hardware-access", timeout=3
        ) as response:
            local_snapshot = json.loads(response.read().decode("utf-8"))
        self.assertEqual(len(local_snapshot["pending"]), 2)

        with post(
            "/api/local/hardware-access/approve", {"client_id": client_b}
        ) as response:
            approved = json.loads(response.read().decode("utf-8"))
        self.assertEqual(approved["active"]["client_id"], client_b)

        with self.assertRaises(urllib.error.HTTPError) as remote_admin:
            request = urllib.request.Request(
                f"{self.base_url}/api/local/hardware-access",
                headers={"X-Forwarded-For": "203.0.113.25"},
            )
            self.opener.open(request, timeout=3)
        self.assertEqual(remote_admin.exception.code, 403)

        def hardware_request(client_id: str, conversation: str):
            return post(
                "/api/requests",
                {
                    "utterance": "关闭副驾座椅加热",
                    "provider": "local",
                    "conversation_id": conversation,
                    "execution_mode": "hardware",
                    "hardware_client_id": client_id,
                },
                public=True,
            )

        fake_result = {"status": "succeeded", "request_id": "req-approved-client"}
        with patch.object(
            self.service, "_process_once", return_value=fake_result
        ) as process_once:
            with hardware_request(client_b, "conv-approved-client") as response:
                self.assertEqual(response.status, 201)
            with self.assertRaises(urllib.error.HTTPError) as unapproved:
                hardware_request(client_a, "conv-unapproved-client")
            denied_payload = json.loads(unapproved.exception.read().decode("utf-8"))
            self.assertEqual(unapproved.exception.code, 403)
            self.assertEqual(denied_payload["error"], "hardware_access_required")
            self.assertEqual(denied_payload["access_state"], "pending")
            self.assertEqual(process_once.call_count, 1)

        with post(
            "/api/hardware/access/release",
            {"client_id": client_b},
            public=True,
        ) as response:
            released = json.loads(response.read().decode("utf-8"))
        self.assertEqual(released["state"], "none")
        self.assertIsNone(self.service.hardware_access.admin_snapshot()["active"])

    def test_local_hardware_decision_page_is_not_exposed_through_proxy(self) -> None:
        with self.opener.open(
            f"{self.base_url}/hardware-control", timeout=3
        ) as response:
            html = response.read().decode("utf-8")
        self.assertIn("真实台架本地决策台", html)

        proxied = urllib.request.Request(
            f"{self.base_url}/hardware-control",
            headers={"X-Forwarded-For": "203.0.113.25"},
        )
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.opener.open(proxied, timeout=3)
        self.assertEqual(caught.exception.code, 403)

    def test_overlapping_hardware_request_returns_conflict_without_entering_can(self) -> None:
        entered = threading.Event()
        release = threading.Event()
        first_result: dict[str, object] = {}
        cookie_header = "; ".join(
            f"{cookie.name}={cookie.value}" for cookie in self.cookie_jar
        )

        def request(conversation_id: str) -> urllib.request.Request:
            return urllib.request.Request(
                f"{self.base_url}/api/requests",
                method="POST",
                headers={
                    "Content-Type": "application/json",
                    "Cookie": cookie_header,
                },
                data=json.dumps(
                    {
                        "utterance": "关闭副驾座椅加热",
                        "provider": "local",
                        "conversation_id": conversation_id,
                        "execution_mode": "hardware",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
            )

        def occupy_bench(*_args: object, **_kwargs: object) -> dict[str, object]:
            entered.set()
            self.assertTrue(release.wait(timeout=3))
            return {"status": "succeeded", "request_id": "req-exclusive-owner"}

        def submit_owner() -> None:
            with urllib.request.urlopen(
                request("conv-exclusive-owner"), timeout=5
            ) as response:
                first_result["http_status"] = response.status
                first_result["payload"] = json.loads(
                    response.read().decode("utf-8")
                )

        with patch.object(
            self.service,
            "_process_once",
            side_effect=occupy_bench,
        ) as process_once:
            owner = threading.Thread(target=submit_owner)
            owner.start()
            self.assertTrue(entered.wait(timeout=2))
            with self.assertRaises(urllib.error.HTTPError) as caught:
                urllib.request.urlopen(
                    request("conv-exclusive-contender"), timeout=3
                )
            payload = json.loads(caught.exception.read().decode("utf-8"))
            self.assertEqual(caught.exception.code, 409)
            self.assertEqual(caught.exception.headers["Retry-After"], "5")
            self.assertEqual(payload["error"], "hardware_busy")
            self.assertIn("未进入CAN链路", payload["message"])
            self.assertEqual(payload["retry_after_seconds"], 5)
            self.assertEqual(process_once.call_count, 1)
            state_request = urllib.request.Request(
                f"{self.base_url}/api/hardware/seat-state?"
                "conversation_id=conv-exclusive-state-reader&vehicle_id=denza_n7",
                headers={"Cookie": cookie_header},
            )
            with self.assertRaises(urllib.error.HTTPError) as state_caught:
                urllib.request.urlopen(state_request, timeout=3)
            state_payload = json.loads(
                state_caught.exception.read().decode("utf-8")
            )
            self.assertEqual(state_caught.exception.code, 409)
            self.assertEqual(state_payload["error"], "hardware_busy")
            release.set()
            owner.join(timeout=5)

        self.assertFalse(owner.is_alive())
        self.assertEqual(first_result["http_status"], 201)
        self.assertFalse(self.service.hardware_execution.public_info()["busy"])


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