feat: serialize public hardware execution

apps/web-console/app.js

@@ -2790,8 +2790,11 @@ async function loadRuntime() {
     const profileText = (hardware.firmware_profiles || []).join(" / ") || "未选择 profile";
     const flashText = hardware.auto_flash_enabled ? "自动烧录已启用" : "自动烧录未启用";
     const approvalText = hardware.approval_required ? "需授权令牌" : "调试免令牌";
+    const concurrencyText = hardware.concurrency_policy === "single_flight_fail_fast"
+      ? " · 单请求独占"
+      : "";
     $("#hardwareNote").textContent = hardware.configured
-      ? `${hardware.backend} · ${hardware.bitrate / 1000} kbit/s · ${profileText} · ${flashText} · ${approvalText}`
+      ? `${hardware.backend} · ${hardware.bitrate / 1000} kbit/s · ${profileText} · ${flashText} · ${approvalText}${concurrencyText}`
       : `真实执行已锁定：${profileText}；请完成服务端启用、适配器、固件镜像${hardware.approval_required ? "和令牌" : ""}配置`;
     $("#hardwareAliasNote").textContent = "副驾座椅 → t2g_primary_seat（统一电机、加热和通风执行器）";
     $("#autoFlashNote").textContent = hardware.auto_flash_enabled
@@ -2802,7 +2805,7 @@ async function loadRuntime() {
       ? '<i data-lucide="shield-check"></i>硬件已配置'
       : '<i data-lucide="shield-off"></i>硬件锁定';
     $("#hardwareCoverage").textContent = hardware.configured
-      ? `可选 · ${approvalText}`
+      ? `可选 · ${approvalText}${concurrencyText}`
       : "已阻断";
     $("#hardwareCoverage").className = hardware.configured ? "available" : "blocked";
     const modeAvailable = state.executionMode === "hardware"
@@ -3684,7 +3687,14 @@ async function submitRequest(event) {
       window.location.replace("/login");
       return;
     }
-    if (!response.ok) throw new Error(payload.message || payload.error || `HTTP ${response.status}`);
+    if (!response.ok) {
+      const requestError = new Error(payload.message || payload.error || `HTTP ${response.status}`);
+      requestError.code = payload.error;
+      requestError.retryAfter = Number(
+        payload.retry_after_seconds || response.headers.get("Retry-After") || 0,
+      );
+      throw requestError;
+    }
     renderResult(payload);
     if (state.executionMode === "hardware") {
       await synchronizePhysicalSeatState();
@@ -3701,9 +3711,16 @@ async function submitRequest(event) {
   } catch (error) {
     window.clearInterval(state.progressTimer);
     resetPipeline();
+    if (error.code === "hardware_busy") {
+      setRunStatus("failed", "BENCH BUSY");
+      $("#resultTitle").textContent = "真实台架正在使用";
+      const retryHint = error.retryAfter > 0 ? ` 建议 ${error.retryAfter} 秒后重试。` : "";
+      showToast(`${error.message}${retryHint}`);
+    } else {
       setRunStatus("failed", "FAILED");
       $("#resultTitle").textContent = "请求处理失败";
       showToast(error.message);
+    }
   } finally {
     button.disabled = false;
     button.classList.remove("loading");

apps/web-console/index.html

@@ -15,7 +15,7 @@
     <script src="/vendor/three/0.147.0/fflate.min.js" defer></script>
     <script src="/vendor/three/0.147.0/FBXLoader.js" defer></script>
     <script src="/network.js?v=android-pad-1" defer></script>
-    <script src="/app.js?v=seat-hil-7" defer></script>
+    <script src="/app.js?v=seat-hil-8" defer></script>
     <script type="module" src="/gesture-control.js?v=android-pad-1"></script>
     <script src="/pwa.js?v=android-pad-1" defer></script>
   </head>

apps/web-console/service-worker.js

@@ -1,4 +1,4 @@
-const CACHE_VERSION = "cognitive-agent-pad-v7";
+const CACHE_VERSION = "cognitive-agent-pad-v8";
 const SHELL_CACHE = `${CACHE_VERSION}-shell`;
 const RUNTIME_CACHE = `${CACHE_VERSION}-runtime`;
 const APP_SHELL = [
@@ -6,7 +6,7 @@ const APP_SHELL = [
   "/index.html",
   "/styles.css?v=android-pad-2",
   "/network.js?v=android-pad-1",
-  "/app.js?v=seat-hil-7",
+  "/app.js?v=seat-hil-8",
   "/gesture-control.js?v=android-pad-1",
   "/pwa.js?v=android-pad-1",
   "/manifest.webmanifest",

deploy/server/README.md

@@ -49,3 +49,19 @@ systemctl status caddy
 Never place `.env`, `.env.auth`, API keys, login passwords, or SSH credentials
 inside the image or source control. Updating the application uses the same
 source synchronization followed by `docker compose ... up -d --build`.
+
+## Public physical-bench gateway
+
+When the physical T2G bench is attached to a separate trusted workstation,
+Caddy may prefer a restricted reverse-SSH upstream on `127.0.0.1:18512` and
+retain the container on `127.0.0.1:8511` as the unavailable-bench fallback.
+The workstation service must use the same `.env.auth` as the public container;
+the SSH key must be restricted to that single reverse-listen port and must not
+be allowed to run remote commands.
+
+The physical bench is a process-wide single-flight resource. A hardware action
+owns the bench from intent planning through final physical feedback. Concurrent
+hardware actions and passive physical-state reads receive HTTP `409`
+(`hardware_busy`) with `Retry-After`; they never enter the CAN path or wait to run
+later against stale position state. Dry-run and simulated-hardware requests are
+not serialized by this gate.

src/cognitive_agent/application/seat_service.py

@@ -61,6 +61,7 @@ from cognitive_agent.infrastructure.llm.openai_responses import (  # noqa: E402
 )
 from cognitive_agent.infrastructure.can.hardware import (  # noqa: E402
     HardwareSafetyError,
+    PHYSICAL_BENCH_COORDINATOR,
     SeatHardwareSettings,
 )
 from cognitive_agent.runtime.contracts import ContractValidator  # noqa: E402
@@ -138,6 +139,7 @@ class SeatDemoService:
             repository_root / "platforms" / "t2g_seat_bench" / "calibration" / "seat.yaml",
         )
         self.hardware_settings = SeatHardwareSettings.from_env(repository_root)
+        self.hardware_execution = PHYSICAL_BENCH_COORDINATOR
         # Shared by the short-lived hardware executor instances. A process must
         # establish a fresh physical four-axis reference before trusting stored
         # Hall counts, and a firmware replacement invalidates it again.
@@ -177,6 +179,8 @@ class SeatDemoService:
             runtime.runtime_info() for runtime in self.domain_runtimes.values()
         ]
         default_vehicle = self.vehicle_profiles.default()
+        hardware_info = self.hardware_settings.public_info()
+        hardware_info.update(self.hardware_execution.public_info())
         return {
             "service": "cabin-cognitive-orchestrator",
             "domain": "cabin",
@@ -184,7 +188,7 @@ class SeatDemoService:
             "execution_mode": "dry_run",
             "execution_modes": ["dry_run", "simulated_hardware", "hardware"],
             "hardware_execution_allowed": self.hardware_settings.configured,
-            "hardware": self.hardware_settings.public_info(),
+            "hardware": hardware_info,
             "seat_position_memory_synchronized": bool(
                 self._seat_position_memory_state.get("synchronized", False)
             ),
@@ -243,6 +247,16 @@ class SeatDemoService:
         self,
         conversation_id: str,
         vehicle_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Read the bench only when no physical execution currently owns it."""
+
+        with self.hardware_execution.reserve():
+            return self._synchronize_hardware_state_once(conversation_id, vehicle_id)
+
+    def _synchronize_hardware_state_once(
+        self,
+        conversation_id: str,
+        vehicle_id: str | None = None,
     ) -> dict[str, Any]:
         """Read the bench and project it into the current passenger conversation."""
 
@@ -308,6 +322,37 @@ class SeatDemoService:
         approval_token: str | None = None,
         vehicle_id: str | None = None,
         simulation_fault: str = "normal",
+    ) -> dict[str, Any]:
+        if execution_mode != "hardware":
+            return self._process_once(
+                utterance,
+                provider,
+                conversation_id,
+                execution_mode,
+                approval_token,
+                vehicle_id,
+                simulation_fault,
+            )
+        with self.hardware_execution.reserve():
+            return self._process_once(
+                utterance,
+                provider,
+                conversation_id,
+                execution_mode,
+                approval_token,
+                vehicle_id,
+                simulation_fault,
+            )
+
+    def _process_once(
+        self,
+        utterance: str,
+        provider: str = "auto",
+        conversation_id: str | None = None,
+        execution_mode: str = "dry_run",
+        approval_token: str | None = None,
+        vehicle_id: str | None = None,
+        simulation_fault: str = "normal",
     ) -> dict[str, Any]:
         utterance = utterance.strip()
         if not utterance:

src/cognitive_agent/infrastructure/can/__init__.py

@@ -2,15 +2,24 @@
 
 from .matrix import CanMatrix, MatrixError
 from .gateway import CanBus, CanFrame, CanGatewayError, PythonCanBus, ZlgCanBridgeBus
-from .hardware import HardwareSafetyError, SeatHardwareSettings
+from .hardware import (
+    HardwareBusyError,
+    HardwareExecutionCoordinator,
+    HardwareSafetyError,
+    PHYSICAL_BENCH_COORDINATOR,
+    SeatHardwareSettings,
+)
 
 __all__ = [
     "CanBus",
     "CanFrame",
     "CanGatewayError",
     "CanMatrix",
+    "HardwareBusyError",
+    "HardwareExecutionCoordinator",
     "HardwareSafetyError",
     "MatrixError",
+    "PHYSICAL_BENCH_COORDINATOR",
     "PythonCanBus",
     "SeatHardwareSettings",
     "ZlgCanBridgeBus",

src/cognitive_agent/infrastructure/can/hardware.py

@@ -5,9 +5,12 @@ from __future__ import annotations
 import hashlib
 import hmac
 import os
+import threading
+import time
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any, Callable, Iterator
 
 from cognitive_agent.infrastructure.firmware import (
     FIRMWARE_PROTOCOL_SIGNATURE,
@@ -36,6 +39,57 @@ class HardwareSafetyError(ValueError):
     """Raised before any CAN device is opened when a safety gate fails."""
 
 
+class HardwareBusyError(RuntimeError):
+    """Raised when another request already owns the single physical bench."""
+
+    def __init__(self, retry_after_seconds: int = 5) -> None:
+        self.retry_after_seconds = retry_after_seconds
+        super().__init__(
+            "真实座椅台架正在执行其他请求。为避免多人指令交叉，本次请求未进入CAN链路；"
+            "请等待当前动作完成后重试。"
+        )
+
+
+class HardwareExecutionCoordinator:
+    """Provide process-wide, fail-fast exclusive ownership of the physical bench."""
+
+    def __init__(self, retry_after_seconds: int = 5) -> None:
+        self.retry_after_seconds = retry_after_seconds
+        self._execution_lock = threading.Lock()
+        self._state_lock = threading.Lock()
+        self._active_since: float | None = None
+
+    @contextmanager
+    def reserve(self) -> Iterator[None]:
+        if not self._execution_lock.acquire(blocking=False):
+            raise HardwareBusyError(self.retry_after_seconds)
+        with self._state_lock:
+            self._active_since = time.monotonic()
+        try:
+            yield
+        finally:
+            with self._state_lock:
+                self._active_since = None
+            self._execution_lock.release()
+
+    def public_info(self) -> dict[str, Any]:
+        with self._state_lock:
+            active_since = self._active_since
+        return {
+            "busy": active_since is not None,
+            "active_for_ms": (
+                max(0, int((time.monotonic() - active_since) * 1000))
+                if active_since is not None
+                else 0
+            ),
+            "concurrency_policy": "single_flight_fail_fast",
+            "retry_after_seconds": self.retry_after_seconds,
+        }
+
+
+PHYSICAL_BENCH_COORDINATOR = HardwareExecutionCoordinator()
+
+
 def _env_bool(name: str, default: bool = False) -> bool:
     value = os.getenv(name)
     return default if value is None else value.strip().lower() in TRUE_VALUES

src/cognitive_agent/server.py

@@ -28,7 +28,10 @@ load_local_env(REPOSITORY_ROOT / ".env.auth", override=True)
 from cognitive_agent.auth import AuthConfig, AuthRuntime, verify_password  # noqa: E402
 from cognitive_agent.application.seat_service import SeatDemoService  # noqa: E402
 from cognitive_agent.application.gesture_bridge import GestureCapabilityBridge  # noqa: E402
-from cognitive_agent.infrastructure.can.hardware import HardwareSafetyError  # noqa: E402
+from cognitive_agent.infrastructure.can.hardware import (  # noqa: E402
+    HardwareBusyError,
+    HardwareSafetyError,
+)
 from cognitive_agent.infrastructure.llm.openai_responses import OpenAIIntentError  # noqa: E402
 
 
@@ -110,6 +113,16 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
                         vehicle_id,
                     )
                 )
+            except HardwareBusyError as exc:
+                self._json(
+                    {
+                        "error": "hardware_busy",
+                        "message": str(exc),
+                        "retry_after_seconds": exc.retry_after_seconds,
+                    },
+                    HTTPStatus.CONFLICT,
+                    {"Retry-After": str(exc.retry_after_seconds)},
+                )
             except ValueError as exc:
                 self._json(
                     {"error": "invalid_request", "message": str(exc)},
@@ -189,6 +202,16 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
                 simulation_fault=str(payload.get("simulation_fault", "normal")),
             )
             self._json(result, HTTPStatus.CREATED)
+        except HardwareBusyError as exc:
+            self._json(
+                {
+                    "error": "hardware_busy",
+                    "message": str(exc),
+                    "retry_after_seconds": exc.retry_after_seconds,
+                },
+                HTTPStatus.CONFLICT,
+                {"Retry-After": str(exc.retry_after_seconds)},
+            )
         except HardwareSafetyError as exc:
             self._json(
                 {"error": "hardware_safety_gate", "message": str(exc)},

tests/e2e/test_seat_web_api.py

@@ -52,7 +52,7 @@ class SeatWebApiTests(unittest.TestCase):
                 session_ttl_seconds=3600,
                 login_attempts=5,
                 login_window_seconds=300,
-                requests_per_minute=20,
+                requests_per_minute=120,
                 trust_proxy_headers=False,
             )
         )
@@ -105,7 +105,7 @@ class SeatWebApiTests(unittest.TestCase):
     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=seat-hil-7", timeout=3) as response:
+        with self.opener.open(f"{self.base_url}/app.js?v=seat-hil-8", timeout=3) as response:
             app_script = 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"))
@@ -115,11 +115,13 @@ class SeatWebApiTests(unittest.TestCase):
         self.assertIn('data-vehicle-model="byd-tang-2024-reference"', html)
         self.assertIn("整车孪生界面", html)
         self.assertIn("透明车体 · 实体设备", html)
-        self.assertIn('src="/app.js?v=seat-hil-7"', html)
+        self.assertIn('src="/app.js?v=seat-hil-8"', html)
         self.assertIn("HARDWARE_REQUEST_TIMEOUT_MS = 15 * 60 * 1000", app_script)
         self.assertIn(
             "requestTimeoutForExecutionMode(state.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)
@@ -145,6 +147,11 @@ class SeatWebApiTests(unittest.TestCase):
             runtime["hardware"]["configured"],
         )
         self.assertEqual(runtime["hardware"]["supported_targets"], ["front_right"])
+        self.assertEqual(
+            runtime["hardware"]["concurrency_policy"],
+            "single_flight_fail_fast",
+        )
+        self.assertFalse(runtime["hardware"]["busy"])
         self.assertEqual(
             runtime["hardware"]["supported_seat_instances"], ["passenger"]
         )
@@ -517,6 +524,85 @@ class SeatWebApiTests(unittest.TestCase):
         )
         self.assertNotIn("must-not-be-persisted", artifact_text)
 
+    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()

tests/unit/test_hardware_execution_coordinator.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import threading
+import unittest
+
+from cognitive_agent.infrastructure.can.hardware import (
+    HardwareBusyError,
+    HardwareExecutionCoordinator,
+)
+
+
+class HardwareExecutionCoordinatorTests(unittest.TestCase):
+    def test_rejects_overlapping_owner_and_releases_after_completion(self) -> None:
+        coordinator = HardwareExecutionCoordinator(retry_after_seconds=7)
+        entered = threading.Event()
+        release = threading.Event()
+
+        def hold_bench() -> None:
+            with coordinator.reserve():
+                entered.set()
+                self.assertTrue(release.wait(timeout=3))
+
+        owner = threading.Thread(target=hold_bench)
+        owner.start()
+        self.assertTrue(entered.wait(timeout=2))
+
+        self.assertTrue(coordinator.public_info()["busy"])
+        with self.assertRaises(HardwareBusyError) as caught:
+            with coordinator.reserve():
+                self.fail("a concurrent request must not enter the physical bench")
+        self.assertEqual(caught.exception.retry_after_seconds, 7)
+
+        release.set()
+        owner.join(timeout=3)
+        self.assertFalse(owner.is_alive())
+        self.assertFalse(coordinator.public_info()["busy"])
+        with coordinator.reserve():
+            self.assertTrue(coordinator.public_info()["busy"])
+
+
+if __name__ == "__main__":
+    unittest.main()