diff --git a/apps/web-console/app.js b/apps/web-console/app.js
index 71ce8b2..6df02e3 100755
--- a/apps/web-console/app.js
+++ b/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() {
? '硬件已配置'
: '硬件锁定';
$("#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();
- setRunStatus("failed", "FAILED");
- $("#resultTitle").textContent = "请求处理失败";
- showToast(error.message);
+ 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");
diff --git a/apps/web-console/index.html b/apps/web-console/index.html
index c97fafd..c8c6978 100755
--- a/apps/web-console/index.html
+++ b/apps/web-console/index.html
@@ -15,7 +15,7 @@
-
+
diff --git a/apps/web-console/service-worker.js b/apps/web-console/service-worker.js
index 9cc8d4f..a308a26 100644
--- a/apps/web-console/service-worker.js
+++ b/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",
diff --git a/deploy/server/README.md b/deploy/server/README.md
index 0482c36..49e0a9c 100755
--- a/deploy/server/README.md
+++ b/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.
diff --git a/src/cognitive_agent/application/seat_service.py b/src/cognitive_agent/application/seat_service.py
index acee11e..4f5ca2d 100755
--- a/src/cognitive_agent/application/seat_service.py
+++ b/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:
diff --git a/src/cognitive_agent/infrastructure/can/__init__.py b/src/cognitive_agent/infrastructure/can/__init__.py
index 26a287e..be76876 100755
--- a/src/cognitive_agent/infrastructure/can/__init__.py
+++ b/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",
diff --git a/src/cognitive_agent/infrastructure/can/hardware.py b/src/cognitive_agent/infrastructure/can/hardware.py
index b634958..78d0ac9 100755
--- a/src/cognitive_agent/infrastructure/can/hardware.py
+++ b/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
diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py
index 8797647..48f6f4f 100755
--- a/src/cognitive_agent/server.py
+++ b/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)},
diff --git a/tests/e2e/test_seat_web_api.py b/tests/e2e/test_seat_web_api.py
index dabe4ab..ee9743e 100755
--- a/tests/e2e/test_seat_web_api.py
+++ b/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()
diff --git a/tests/unit/test_hardware_execution_coordinator.py b/tests/unit/test_hardware_execution_coordinator.py
new file mode 100644
index 0000000..5756950
--- /dev/null
+++ b/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()