Add AC_usb_* executor commands for headless USB passthrough control

Every GUI action now has an executor command so JSON action files, the
socket server, and the scheduler can drive USB passthrough without a GUI:
feature flag, ACL CRUD + export/import, local loopback list/open, and
remote (live WebRTC) list/open. Each returns a JSON-able dict; the
descriptor probe summarises the device. Stubs added to actions.pyi.

Author: JE-Chen

docs/source/Eng/doc/operations_layer/usb_passthrough_operator_guide.rst

@@ -270,6 +270,38 @@ Audit chain shows ``broken_at_id``          Someone edited ``audit.db`` directly
 ==========================================  =====================================================
 
 
+Headless control (``AC_usb_*`` commands)
+========================================
+
+Everything the GUI does is also an executor command, so JSON action
+files, the socket server, and the scheduler can drive USB passthrough
+with no GUI:
+
+================================  ============================================
+Command                           Purpose
+================================  ============================================
+``AC_usb_passthrough_enable``     Toggle the feature flag (``enabled`` bool)
+``AC_usb_passthrough_status``     Report whether passthrough is enabled
+``AC_usb_acl_list``               List ACL rules + default + integrity state
+``AC_usb_acl_add``                Add a per-device rule
+``AC_usb_acl_remove``             Remove a rule
+``AC_usb_acl_set_default``        Set the default policy (allow/deny)
+``AC_usb_acl_export`` / ``_import``  Back up / restore the ACL as JSON
+``AC_usb_loopback_list``          List ACL-visible devices over loopback
+``AC_usb_loopback_open``          Claim a local device + read its descriptor
+``AC_usb_remote_list``            List a remote host's devices (live WebRTC)
+``AC_usb_remote_open``            Claim a remote device + read its descriptor
+================================  ============================================
+
+Example JSON action::
+
+    [
+      ["AC_usb_passthrough_enable", {"enabled": true}],
+      ["AC_usb_acl_add", {"vendor_id": "1050", "product_id": "0407"}],
+      ["AC_usb_loopback_open", {"vendor_id": "1050", "product_id": "0407"}]
+    ]
+
+
 What is *not* shipped yet
 =========================
 

docs/source/Zh/doc/operations_layer/usb_passthrough_operator_guide.rst

@@ -251,6 +251,37 @@ OPEN 後 host 鍵盤停止運作                   Linux：HID 裝置被 claim 
 ==========================================  =====================================================
 
 
+無 GUI 控制（``AC_usb_*`` 指令）
+================================
+
+GUI 能做的事都有對應的 executor 指令，所以 JSON action 檔、socket
+server 與排程器都能在沒有 GUI 的情況下驅動 USB passthrough：
+
+================================  ============================================
+指令                              用途
+================================  ============================================
+``AC_usb_passthrough_enable``     切換 feature flag（``enabled`` 布林）
+``AC_usb_passthrough_status``     回報是否已啟用
+``AC_usb_acl_list``               列出 ACL 規則 + 預設 + 完整性狀態
+``AC_usb_acl_add``                新增 per-device 規則
+``AC_usb_acl_remove``             移除規則
+``AC_usb_acl_set_default``        設定預設政策（allow/deny）
+``AC_usb_acl_export`` / ``_import``  以 JSON 備份／還原 ACL
+``AC_usb_loopback_list``          經 loopback 列出 ACL 可見裝置
+``AC_usb_loopback_open``          claim 本機裝置並讀描述元
+``AC_usb_remote_list``            列出遠端主機裝置（live WebRTC）
+``AC_usb_remote_open``            claim 遠端裝置並讀描述元
+================================  ============================================
+
+JSON action 範例::
+
+    [
+      ["AC_usb_passthrough_enable", {"enabled": true}],
+      ["AC_usb_acl_add", {"vendor_id": "1050", "product_id": "0407"}],
+      ["AC_usb_loopback_open", {"vendor_id": "1050", "product_id": "0407"}]
+    ]
+
+
 尚未發布的部分
 ==============
 

je_auto_control/actions.pyi

@@ -468,6 +468,42 @@ def AC_usb_watch_start(poll_interval_s: float = ...) -> Dict[str, Any]:
 def AC_usb_watch_stop() -> Dict[str, Any]:
     ...
 
+def AC_usb_passthrough_enable(enabled: bool = ...) -> Dict[str, Any]:
+    """Toggle the USB passthrough feature flag (default off)."""
+
+def AC_usb_passthrough_status() -> Dict[str, Any]:
+    ...
+
+def AC_usb_acl_list() -> Dict[str, Any]:
+    ...
+
+def AC_usb_acl_add(vendor_id: str, product_id: str, serial: str | None = ..., allow: bool = ..., prompt_on_open: bool = ..., label: str = ...) -> Dict[str, Any]:
+    """Add a per-device rule to the USB ACL."""
+
+def AC_usb_acl_remove(vendor_id: str, product_id: str, serial: str | None = ...) -> Dict[str, Any]:
+    ...
+
+def AC_usb_acl_set_default(policy: str) -> Dict[str, Any]:
+    ...
+
+def AC_usb_acl_export(path: str) -> Dict[str, Any]:
+    ...
+
+def AC_usb_acl_import(path: str, replace: bool = ...) -> Dict[str, Any]:
+    ...
+
+def AC_usb_loopback_list() -> Dict[str, Any]:
+    """List ACL-visible devices over the in-process loopback channel."""
+
+def AC_usb_loopback_open(vendor_id: str, product_id: str, serial: str | None = ...) -> Dict[str, Any]:
+    """Claim a local device over loopback and read its descriptor."""
+
+def AC_usb_remote_list() -> Dict[str, Any]:
+    """List the remote host's devices over the live WebRTC usb channel."""
+
+def AC_usb_remote_open(vendor_id: str, product_id: str, serial: str | None = ...) -> Dict[str, Any]:
+    """Claim a remote device over the live WebRTC usb channel and probe it."""
+
 def AC_vlm_click(description: str, screen_region: List[int] | None = ..., model: str | None = ..., backend: Any = ...) -> bool:
     """Locate by description, then click the center of the match."""
 

je_auto_control/utils/executor/action_executor.py

@@ -957,6 +957,136 @@ def _usb_recent_events(since: int = 0,
     )
 
 
+# --- USB passthrough (Phase 2) ------------------------------------------
+
+
+def _usb_passthrough_enable(enabled: bool = True) -> Dict[str, Any]:
+    """Toggle the USB passthrough feature flag (default off)."""
+    from je_auto_control.utils.usb.passthrough import (
+        enable_usb_passthrough, is_usb_passthrough_enabled,
+    )
+    enable_usb_passthrough(bool(enabled))
+    return {"enabled": is_usb_passthrough_enabled()}
+
+
+def _usb_passthrough_status() -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import is_usb_passthrough_enabled
+    return {"enabled": is_usb_passthrough_enabled()}
+
+
+def _usb_acl_list() -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import UsbAcl
+    acl = UsbAcl()
+    return {
+        "default": acl.default_policy,
+        "integrity_ok": acl.integrity_ok,
+        "rules": [r.to_dict() for r in acl.list_rules()],
+    }
+
+
+def _usb_acl_add(vendor_id: str, product_id: str,
+                 serial: Optional[str] = None, allow: bool = True,
+                 prompt_on_open: bool = False, label: str = "") -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import AclRule, UsbAcl
+    acl = UsbAcl()
+    acl.add_rule(AclRule(
+        vendor_id=str(vendor_id), product_id=str(product_id),
+        serial=(None if serial is None else str(serial)),
+        label=str(label), allow=bool(allow),
+        prompt_on_open=bool(prompt_on_open),
+    ))
+    return {"added": True, "rules": len(acl.list_rules())}
+
+
+def _usb_acl_remove(vendor_id: str, product_id: str,
+                    serial: Optional[str] = None) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import UsbAcl
+    removed = UsbAcl().remove_rule(
+        vendor_id=str(vendor_id), product_id=str(product_id),
+        serial=(None if serial is None else str(serial)),
+    )
+    return {"removed": removed}
+
+
+def _usb_acl_set_default(policy: str) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import UsbAcl
+    acl = UsbAcl()
+    acl.set_default_policy(str(policy))
+    return {"default": acl.default_policy}
+
+
+def _usb_acl_export(path: str) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import UsbAcl, export_acl_to_file
+    from pathlib import Path
+    export_acl_to_file(UsbAcl(), Path(path))
+    return {"exported": True, "path": str(path)}
+
+
+def _usb_acl_import(path: str, replace: bool = False) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import UsbAcl, import_acl_from_file
+    from pathlib import Path
+    count = import_acl_from_file(UsbAcl(), Path(path), replace=bool(replace))
+    return {"imported": count}
+
+
+def _usb_descriptor_probe(client: Any, vendor_id: str, product_id: str,
+                          serial: Optional[str]) -> Dict[str, Any]:
+    """Open a claim, read the device descriptor, close — return a summary."""
+    from je_auto_control.utils.usb.passthrough import describe_descriptor
+    handle = client.open(
+        vendor_id=str(vendor_id), product_id=str(product_id),
+        serial=(None if serial is None else str(serial)),
+    )
+    try:
+        descriptor = handle.control_transfer(
+            bm_request_type=0x80, b_request=0x06, w_value=0x0100, length=18,
+        )
+    finally:
+        handle.close()
+    return {
+        "ok": True, "vendor_id": str(vendor_id), "product_id": str(product_id),
+        "descriptor_hex": descriptor.hex(),
+        "descriptor": describe_descriptor(descriptor),
+    }
+
+
+def _usb_loopback_list() -> Dict[str, Any]:
+    """List ACL-visible devices over the in-process loopback channel."""
+    from je_auto_control.utils.usb.passthrough import UsbAcl, UsbLoopback
+    with UsbLoopback(acl=UsbAcl(), viewer_id="executor") as loop:
+        return {"devices": loop.list_devices()}
+
+
+def _usb_loopback_open(vendor_id: str, product_id: str,
+                       serial: Optional[str] = None) -> Dict[str, Any]:
+    """Claim a local device over loopback and read its descriptor."""
+    from je_auto_control.utils.usb.passthrough import UsbAcl, UsbLoopback
+    with UsbLoopback(acl=UsbAcl(), viewer_id="executor") as loop:
+        return _usb_descriptor_probe(loop, vendor_id, product_id, serial)
+
+
+def _usb_remote_client() -> Any:
+    """Return the live WebRTC viewer's USB client or raise a clear error."""
+    from je_auto_control.utils.remote_desktop.registry import registry
+    client = registry.webrtc_usb_client()
+    if client is None:
+        raise RuntimeError("no live WebRTC viewer with a usb channel")
+    return client
+
+
+def _usb_remote_list() -> Dict[str, Any]:
+    """List the remote host's devices over the live WebRTC usb channel."""
+    return {"devices": _usb_remote_client().list_devices()}
+
+
+def _usb_remote_open(vendor_id: str, product_id: str,
+                     serial: Optional[str] = None) -> Dict[str, Any]:
+    """Claim a remote device over the live WebRTC usb channel + probe."""
+    return _usb_descriptor_probe(
+        _usb_remote_client(), vendor_id, product_id, serial,
+    )
+
+
 def _ac_web_run(action: Optional[Dict[str, Any]] = None,
                 **action_kwargs: Any) -> Any:
     """Bridge one WR_* action into the WebRunner executor (Phase 7.7).
@@ -1858,6 +1988,20 @@ def __init__(self):
             "AC_usb_watch_stop": _usb_watch_stop,
             "AC_usb_recent_events": _usb_recent_events,
 
+            # USB passthrough (Phase 2) — flag, ACL, local + remote use
+            "AC_usb_passthrough_enable": _usb_passthrough_enable,
+            "AC_usb_passthrough_status": _usb_passthrough_status,
+            "AC_usb_acl_list": _usb_acl_list,
+            "AC_usb_acl_add": _usb_acl_add,
+            "AC_usb_acl_remove": _usb_acl_remove,
+            "AC_usb_acl_set_default": _usb_acl_set_default,
+            "AC_usb_acl_export": _usb_acl_export,
+            "AC_usb_acl_import": _usb_acl_import,
+            "AC_usb_loopback_list": _usb_loopback_list,
+            "AC_usb_loopback_open": _usb_loopback_open,
+            "AC_usb_remote_list": _usb_remote_list,
+            "AC_usb_remote_open": _usb_remote_open,
+
             # System diagnostics
             "AC_diagnose": _diagnose,
 

test/unit_test/headless/test_usb_passthrough_executor.py

@@ -0,0 +1,95 @@
+"""Tests for the AC_usb_* passthrough executor commands."""
+import pytest
+
+from je_auto_control.utils.executor import action_executor as ax
+from je_auto_control.utils.executor.action_executor import executor
+from je_auto_control.utils.usb.passthrough.backend import (
+    BackendDevice, FakeUsbBackend,
+)
+
+_SAMPLE = BackendDevice(vendor_id="1050", product_id="0407", serial="ABC123")
+
+_NEW_COMMANDS = [
+    "AC_usb_passthrough_enable", "AC_usb_passthrough_status",
+    "AC_usb_acl_list", "AC_usb_acl_add", "AC_usb_acl_remove",
+    "AC_usb_acl_set_default", "AC_usb_acl_export", "AC_usb_acl_import",
+    "AC_usb_loopback_list", "AC_usb_loopback_open",
+    "AC_usb_remote_list", "AC_usb_remote_open",
+]
+
+
+@pytest.fixture()
+def temp_acl(monkeypatch, tmp_path):
+    """Point UsbAcl at a temp path so the user's real ACL is untouched."""
+    monkeypatch.setattr(
+        "je_auto_control.utils.usb.passthrough.acl.default_acl_path",
+        lambda: tmp_path / "usb_acl.json",
+    )
+    monkeypatch.setattr(
+        "je_auto_control.utils.usb.passthrough.loopback.default_passthrough_backend",
+        lambda: FakeUsbBackend(devices=[_SAMPLE]),
+    )
+    return tmp_path
+
+
+def test_all_commands_registered():
+    known = executor.known_commands()
+    for name in _NEW_COMMANDS:
+        assert name in known, name
+
+
+def test_flag_enable_and_status():
+    try:
+        assert ax._usb_passthrough_enable(True)["enabled"] is True
+        assert ax._usb_passthrough_status()["enabled"] is True
+        assert ax._usb_passthrough_enable(False)["enabled"] is False
+    finally:
+        ax._usb_passthrough_enable(False)
+
+
+def test_acl_add_list_remove(temp_acl):
+    assert ax._usb_acl_add("1050", "0407", allow=True)["added"] is True
+    listed = ax._usb_acl_list()
+    assert listed["default"] == "deny"
+    assert len(listed["rules"]) == 1
+    assert listed["rules"][0]["vendor_id"] == "1050"
+    assert ax._usb_acl_remove("1050", "0407")["removed"] is True
+    assert ax._usb_acl_list()["rules"] == []
+
+
+def test_acl_set_default(temp_acl):
+    assert ax._usb_acl_set_default("allow")["default"] == "allow"
+    assert ax._usb_acl_list()["default"] == "allow"
+
+
+def test_acl_export_import(temp_acl):
+    ax._usb_acl_add("1050", "0407", allow=True)
+    out = temp_acl / "exp.json"
+    assert ax._usb_acl_export(str(out))["exported"] is True
+    ax._usb_acl_remove("1050", "0407")
+    assert ax._usb_acl_import(str(out))["imported"] == 1
+    assert len(ax._usb_acl_list()["rules"]) == 1
+
+
+def test_loopback_list_and_open(temp_acl):
+    ax._usb_acl_add("1050", "0407", allow=True)
+    devices = ax._usb_loopback_list()["devices"]
+    assert [d["vendor_id"] for d in devices] == ["1050"]
+    opened = ax._usb_loopback_open("1050", "0407", serial="ABC123")
+    assert opened["ok"] is True
+    assert "descriptor" in opened
+    assert isinstance(opened["descriptor_hex"], str)
+
+
+def test_loopback_open_denied_without_rule(temp_acl):
+    from je_auto_control.utils.usb.passthrough import UsbClientError
+    # No allow rule → default deny → open fails closed.
+    with pytest.raises(UsbClientError):
+        ax._usb_loopback_open("1050", "0407")
+
+
+def test_remote_list_without_session_raises():
+    from je_auto_control.utils.remote_desktop.registry import registry
+    registry._webrtc_viewer = None  # noqa: SLF001  test setup
+    with pytest.raises(RuntimeError):
+        ax._usb_remote_list()