From 85b4740d29a2bd29099faa6e066d39a38186b0fc Mon Sep 17 00:00:00 2001 From: vee Date: Mon, 3 Aug 2026 09:25:38 +0800 Subject: [PATCH] feat: serialize public hardware execution --- apps/web-console/app.js | 29 ++++-- apps/web-console/index.html | 2 +- apps/web-console/service-worker.js | 4 +- deploy/server/README.md | 16 ++++ .../application/seat_service.py | 47 +++++++++- .../infrastructure/can/__init__.py | 11 ++- .../infrastructure/can/hardware.py | 56 ++++++++++- src/cognitive_agent/server.py | 25 ++++- tests/e2e/test_seat_web_api.py | 92 ++++++++++++++++++- .../test_hardware_execution_coordinator.py | 42 +++++++++ 10 files changed, 308 insertions(+), 16 deletions(-) create mode 100644 tests/unit/test_hardware_execution_coordinator.py 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()