${nfcPet ? renderDetectedPetCard(nfcPet) : ""}
- ${devices.length ? devices.map((device) => `
-
-
- ${device.actions.slice(0, 8).map((action) => `${escapeHtml(action)}`).join("")}
- `).join("") : nfcPet ? "" : `
+ ${devices.length ? devices.map((device) => {
+ const expanded = expandedPeripheralDeviceId === device.id;
+ return `
+
+
+ ${expanded ? `
+ ${device.actions.map((action) => ``).join("")}
+
` : ""}
+ `;
+ }).join("") : nfcPet ? "" : `
等待 NFC 外设靠近检测到已绑定的 NFC 后，外设将在本次运行期间持续显示
`}
diff --git a/apps/web-console/service-worker.js b/apps/web-console/service-worker.js
index 90e81fb..310c2d0 100644
--- a/apps/web-console/service-worker.js
+++ b/apps/web-console/service-worker.js
@@ -11,8 +11,8 @@ const APP_SHELL = [
"/pwa.js?v=android-pad-3",
"/console-v2.css?v=delivery-state-69",
"/console-v2.js?v=local-first-79-cinema-status",
- "/self-extension.css?v=pet-settings-54-dialog-size",
- "/self-extension.js?v=nfc-session-58",
+ "/self-extension.css?v=peripheral-cards-55",
+ "/self-extension.js?v=peripheral-cards-59",
"/speech-input.css?v=sensevoice-7",
"/speech-input.js?v=sensevoice-13",
"/speech-audio-worklet.js?v=sensevoice-5",
diff --git a/src/cognitive_agent/application/extension_toy_publisher.py b/src/cognitive_agent/application/extension_toy_publisher.py
index 3296438..1bfd414 100644
--- a/src/cognitive_agent/application/extension_toy_publisher.py
+++ b/src/cognitive_agent/application/extension_toy_publisher.py
@@ -6,6 +6,7 @@ import json
import re
import threading
import time
+import uuid
from collections.abc import Callable, Mapping
from copy import deepcopy
from datetime import UTC, datetime
@@ -157,6 +158,7 @@ class NfcExtensionToyPublisher:
self._thread: threading.Thread | None = None
self._revision = 0
self._status: dict[str, Any] = {
+ "session_id": None,
"state": "stopped",
"topic": EXTENSION_TOY_MQTT_TOPIC,
"qos": 1,
@@ -168,6 +170,7 @@ class NfcExtensionToyPublisher:
"active_extension_names": [],
"active_target_scopes": [],
"detected_nfc_ids": [],
+ "detection_count": 0,
"message_count": 0,
"burst_frame_attempt_count": 0,
"burst_publish_attempt_count": 0,
@@ -188,12 +191,14 @@ class NfcExtensionToyPublisher:
return
self._stop.clear()
self._set_status(
+ session_id=uuid.uuid4().hex,
state="waiting_for_nfc",
active_nfc_id=None,
active_catalog_revision=None,
active_extension_names=[],
active_target_scopes=[],
detected_nfc_ids=[],
+ detection_count=0,
message_count=0,
last_error=None,
)
@@ -211,12 +216,14 @@ class NfcExtensionToyPublisher:
thread.join(timeout=2.0)
self._thread = None
self._set_status(
+ session_id=None,
state="stopped",
active_nfc_id=None,
active_catalog_revision=None,
active_extension_names=[],
active_target_scopes=[],
detected_nfc_ids=[],
+ detection_count=0,
message_count=0,
)
@@ -339,6 +346,7 @@ class NfcExtensionToyPublisher:
def _run(self) -> None:
latched_uid = ""
+ present_uid = ""
latched_catalog_revision = ""
latched_message_signature: str | None = None
while not self._stop.is_set():
@@ -349,6 +357,16 @@ class NfcExtensionToyPublisher:
self._stop.wait(self.poll_interval_s)
continue
+ if card_uid != present_uid:
+ present_uid = card_uid
+ if card_uid:
+ with self._lock:
+ detection_count = int(self._status["detection_count"]) + 1
+ self._set_status(
+ active_nfc_id=card_uid,
+ detection_count=detection_count,
+ )
+
if card_uid:
with self._lock:
detected_nfc_ids = list(self._status["detected_nfc_ids"])
diff --git a/src/cognitive_agent/application/self_extension.py b/src/cognitive_agent/application/self_extension.py
index 322785d..9d45867 100644
--- a/src/cognitive_agent/application/self_extension.py
+++ b/src/cognitive_agent/application/self_extension.py
@@ -2976,6 +2976,48 @@ class SelfExtensionService:
},
}
+ def execute_quick_peripheral_capability(
+ self,
+ capability_id: object,
+ *,
+ conversation_id: str | None = None,
+ ) -> dict[str, Any]:
+ """Execute one durable NFC-authorized capability selected by the UI."""
+
+ normalized_id = str(capability_id or "").strip()
+ if not normalized_id or len(normalized_id) > 128:
+ raise ValueError("外设快捷控制缺少有效的能力 ID。")
+ capability = next(
+ (
+ dict(item)
+ for item in self.list_peripheral_capabilities()
+ if isinstance(item, Mapping)
+ and str(item.get("id") or "").strip() == normalized_id
+ and item.get("status") == "learned"
+ and item.get("enabled") is True
+ and bool(_normalized_card_uid(item.get("card_uid")))
+ ),
+ None,
+ )
+ if capability is None:
+ raise ExtensionIntegrationError(
+ "外设快捷控制未通过当前持久 NFC 能力目录校验。"
+ )
+ device_name = str(
+ capability.get("device_name")
+ or capability.get("device_id")
+ or "外设"
+ ).strip()
+ action_name = str(capability.get("action") or "动作").strip()
+ result = self.execute_peripheral_request(
+ f"{device_name} {action_name}",
+ capability,
+ conversation_id=conversation_id,
+ execution_mode="hardware",
+ provider_requested="web_console_quick_control",
+ )
+ return {"success": True, **result}
+
def execute_peripheral_plan_request(
self,
utterance: str,
diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py
index 413a416..bf27ab2 100755
--- a/src/cognitive_agent/server.py
+++ b/src/cognitive_agent/server.py
@@ -4554,6 +4554,21 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
)
self._json({"success": True, **sync_result})
return
+ if path == "/api/extension/peripherals/execute":
+ result = self.extension.execute_quick_peripheral_capability(
+ payload.get("capability_id"),
+ conversation_id=str(payload.get("conversation_id") or "").strip()
+ or None,
+ )
+ self.vehicle_display_sync.update_extension(
+ action={
+ "type": "peripheral_quick_control",
+ "command": result.get("utterance"),
+ },
+ result=result,
+ )
+ self._json(result)
+ return
if path == "/api/extension/assistant/plan":
self.vehicle_display_sync.update_extension(
catalog=payload.get("peripherals", payload.get("catalog", [])),
diff --git a/tests/e2e/test_seat_web_api.py b/tests/e2e/test_seat_web_api.py
index 1d6cd9c..4530d62 100755
--- a/tests/e2e/test_seat_web_api.py
+++ b/tests/e2e/test_seat_web_api.py
@@ -1024,11 +1024,11 @@ operations:
) as response:
speech_worklet = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/self-extension.js?v=nfc-session-58", timeout=3
+ f"{self.base_url}/self-extension.js?v=peripheral-cards-59", timeout=3
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/self-extension.css?v=pet-settings-54-dialog-size", timeout=3
+ f"{self.base_url}/self-extension.css?v=peripheral-cards-55", timeout=3
) as response:
extension_styles = response.read().decode("utf-8").replace("\r\n", "\n")
with self.opener.open(
@@ -1105,7 +1105,7 @@ operations:
self.assertIn('id="v2BootScreen"', template)
self.assertIn('正在加载车机控制与实时状态', template)
self.assertIn(
- f'src="{runtime_url("self-extension.js", "/self-extension.js?v=nfc-session-58")}"',
+ f'src="{runtime_url("self-extension.js", "/self-extension.js?v=peripheral-cards-59")}"',
html,
)
self.assertIn('import "../self-extension.css";', vite_entry)
diff --git a/tests/unit/test_extension_toy_publisher.py b/tests/unit/test_extension_toy_publisher.py
index 2600b10..ca7bc1d 100644
--- a/tests/unit/test_extension_toy_publisher.py
+++ b/tests/unit/test_extension_toy_publisher.py
@@ -131,6 +131,9 @@ def test_each_detection_publishes_three_frames_per_device_and_then_stops() -> No
for _at, publication in publications
) == {"toy-fan-001": 3, "toy-light-001": 3}
status = service.snapshot()
+ session_id = status["session_id"]
+ assert isinstance(session_id, str) and session_id
+ assert status["detection_count"] == 1
assert status["state"] == "burst_complete"
assert status["burst_frame_attempt_count"] == 3
assert status["burst_publish_attempt_count"] == 6
@@ -147,6 +150,8 @@ def test_each_detection_publishes_three_frames_per_device_and_then_stops() -> No
nfc["card"] = {"action": "present", "uid": "E113B203"}
assert wait_until(lambda: len(publications) == 12)
assert service.snapshot()["state"] == "burst_complete"
+ assert service.snapshot()["session_id"] == session_id
+ assert service.snapshot()["detection_count"] == 2
finally:
service.close()
@@ -172,6 +177,8 @@ def test_detected_devices_remain_latched_until_publisher_stops() -> None:
service.close()
assert service.snapshot()["detected_nfc_ids"] == []
+ assert service.snapshot()["session_id"] is None
+ assert service.snapshot()["detection_count"] == 0
def test_no_matching_ability_publishes_nothing() -> None:
diff --git a/tests/unit/test_frontend_performance_contract.py b/tests/unit/test_frontend_performance_contract.py
index 1c3ec31..0ba3f40 100644
--- a/tests/unit/test_frontend_performance_contract.py
+++ b/tests/unit/test_frontend_performance_contract.py
@@ -94,7 +94,7 @@ def test_performance_assets_have_a_new_cache_generation() -> None:
for asset in (
"/app.js?v=dual-seat-147-video-media",
"/console-v2.js?v=local-first-79-cinema-status",
- "/self-extension.js?v=nfc-session-58",
+ "/self-extension.js?v=peripheral-cards-59",
"/energy-monitor.js?v=energy-monitor-5-performance",
):
assert asset in html
diff --git a/tests/unit/test_frontend_theme_contract.py b/tests/unit/test_frontend_theme_contract.py
index 0fba051..37d72d7 100644
--- a/tests/unit/test_frontend_theme_contract.py
+++ b/tests/unit/test_frontend_theme_contract.py
@@ -47,7 +47,7 @@ def test_theme_asset_versions_are_consistent_with_the_pwa_shell() -> None:
for source_asset, cached_asset in (
('import "../styles.css"', "/styles.css?v=dual-seat-24-dialog-size"),
('import "../console-v2.css"', "/console-v2.css?v=delivery-state-69"),
- ('import "../self-extension.css"', "/self-extension.css?v=pet-settings-54-dialog-size"),
+ ('import "../self-extension.css"', "/self-extension.css?v=peripheral-cards-55"),
('import "../speech-input.css"', "/speech-input.css?v=sensevoice-7"),
('import "../codepen-glass-theme.css"', "/codepen-glass-theme.css?v=creative-cloud-6-performance"),
):
diff --git a/tests/unit/test_nfc_frontend_contract.py b/tests/unit/test_nfc_frontend_contract.py
index 56f3bb6..cf79164 100644
--- a/tests/unit/test_nfc_frontend_contract.py
+++ b/tests/unit/test_nfc_frontend_contract.py
@@ -67,13 +67,23 @@ def test_unified_console_exposes_device_id_based_nfc_authorization() -> None:
assert 'signalSource: "backend_catalog"' in script
assert '等待 NFC 外设靠近' in script
assert '检测到已绑定的 NFC 后，外设将在本次运行期间持续显示' in script
+ assert 'data-hide-peripheral-card' in script
+ assert 'data-toggle-peripheral-card' in script
+ assert 'data-run-peripheral-quick-action' in script
+ assert 'function hidePeripheralCard(deviceId, cardUid)' in script
+ assert 'function togglePeripheralCard(deviceId)' in script
+ assert 'api("peripherals/execute"' in script
+ assert 'const hiddenPeripheralDevices = new Map()' in script
+ assert 'detection_count' in script
assert '.v2-nfc-inline' in styles
assert '.v2-nfc-dialog' in styles
assert '.v2-nfc-command-list' in styles
assert '.v2-nfc-command-option' in styles
assert '.v2-nfc-select-all' in styles
- assert 'self-extension.js?v=nfc-session-58' in html
+ assert '.v2-extension-device-hide' in styles
+ assert '.v2-extension-quick-controls' in styles
+ assert 'self-extension.js?v=peripheral-cards-59' in html
assert 'import "../self-extension.css";' in vite_entry
@@ -144,6 +154,8 @@ def test_server_exposes_nfc_snapshot_through_same_origin_extension_api() -> None
assert 'SeatDemoHandler.pets.start()' in server
assert 'SeatDemoHandler.pets.close()' in server
assert 'SeatDemoHandler.nfc.close()' in server
+ assert 'path == "/api/extension/peripherals/execute"' in server
+ assert "execute_quick_peripheral_capability" in server
assert "frame-src 'self' http://*:3001" in server
diff --git a/tests/unit/test_self_extension.py b/tests/unit/test_self_extension.py
index 0965019..e1667eb 100644
--- a/tests/unit/test_self_extension.py
+++ b/tests/unit/test_self_extension.py
@@ -1403,6 +1403,40 @@ class SelfExtensionTests(unittest.TestCase):
self.assertEqual(operation["domain"], "self_extension")
self.assertTrue(operation["telemetry"]["gateway_acknowledged"])
+ def test_quick_peripheral_control_reauthorizes_durable_capability(self) -> None:
+ with TemporaryDirectory() as directory:
+ service = SelfExtensionService(
+ SimulationGateway(),
+ catalog_db_path=Path(directory, "infrared.sqlite3"),
+ )
+ service.sync_peripheral_catalog(
+ [
+ {
+ "id": "quick-light-on",
+ "status": "learned",
+ "enabled": True,
+ "device_id": "quick-light",
+ "device_name": "快捷灯",
+ "object_names": ["快捷灯"],
+ "card_uid": "2CC8B103",
+ "action": "打开",
+ "action_names": ["打开", "开启"],
+ "slot": 8,
+ }
+ ]
+ )
+
+ result = service.execute_quick_peripheral_capability(
+ "quick-light-on",
+ conversation_id="conv-quick-light",
+ )
+
+ self.assertTrue(result["success"])
+ self.assertEqual(result["execution_mode"], "hardware")
+ self.assertEqual(result["provider_requested"], "web_console_quick_control")
+ self.assertEqual(result["peripheral"]["slot"], 8)
+ service.close()
+
def test_main_peripheral_sequence_executes_validated_actions_and_delays(self) -> None:
service = SelfExtensionService(SimulationGateway())
catalog = [
diff --git a/tests/unit/test_self_extension_nfc_sync_contract.py b/tests/unit/test_self_extension_nfc_sync_contract.py
index 0dfa695..ceefd50 100644
--- a/tests/unit/test_self_extension_nfc_sync_contract.py
+++ b/tests/unit/test_self_extension_nfc_sync_contract.py
@@ -116,3 +116,18 @@ def test_homepage_only_reveals_nfc_devices_detected_in_this_run() -> None:
assert "const detectedNfcIds = new Set(" in unified
assert "detectedNfcIds.has(normalizeCardUid(device.cardUid))" in unified
assert "检测到已绑定的 NFC 后，外设将在本次运行期间持续显示" in unified
+
+
+def test_homepage_peripheral_cards_are_single_expand_and_quick_control_surfaces() -> None:
+ source = SOURCE_PATH.read_text(encoding="utf-8")
+ unified = _function_source(
+ source,
+ " function buildUnifiedExtensionSync() {",
+ "\n function init()",
+ )
+
+ assert "expandedPeripheralDeviceId === device.id" in unified
+ assert 'data-hide-peripheral-card="${escapeHtml(device.id)}"' in unified
+ assert 'data-toggle-peripheral-card="${escapeHtml(device.id)}"' in unified
+ assert 'data-run-peripheral-quick-action="${escapeHtml(action.id)}"' in unified
+ assert "hiddenPeripheralDevices.has(device.id)" in unified