feat(computer_use): align cua-driver 0.10 permission modes

This commit is contained in:
Francesco Bonacci 2026-07-20 17:06:01 -05:00 committed by Teknium
parent 847e401b74
commit c268397752
14 changed files with 620 additions and 74 deletions

View File

@ -599,7 +599,10 @@ def computer_use_guidance(platform_name: Optional[str] = None) -> str:
"downgrade, never an automatic retry. Use native capture/input for "
"browser chrome, OS permission prompts, native dialogs, and unsupported "
"targets. Browser setup is a separately approved action; attaching an "
"existing profile requires cua-driver's own interactive grant.\n\n"
"existing profile is enforced by cua-driver's immutable permission "
"mode: standard requires a certified protected host and fails closed "
"when Hermes has none; explicit Hermes YOLO uses a private unrestricted "
"daemon after the user's launch/session risk acceptance.\n\n"
"## Background mode rules\n"
"- Do NOT use `raise_window=true` on `focus_app` unless the user "
"explicitly asked you to bring a window to front. Input routing to "

View File

@ -3976,10 +3976,14 @@ class AIAgent:
except Exception:
pass
# 4. Release the session-owned computer-use backend. The lazy import
# keeps sessions that never enabled computer use on the narrow path.
# 4. Release the session-owned computer-use backend. This ends the
# exact cua-driver session, drops typed-browser refs/grants, and stops
# a private embedded daemon when Hermes YOLO selected unrestricted
# mode. The import is lazy so sessions without computer_use retain
# the narrow core footprint.
try:
from tools.computer_use import release_computer_use_session
release_computer_use_session(task_id)
except Exception:
pass

View File

@ -182,9 +182,12 @@ browser tools. The contract is capability-based:
`cua_browser_prepare` is a separate approved setup action. Driver-owned
`isolated_new`/`isolated_named` profiles require explicit `allow_launch=true`.
An `existing_profile` requires cua-driver's own exact, interactive grant;
ordinary Hermes approval is not a substitute and no grant token may be
invented, stored, logged, or reused.
An `existing_profile` is decided by cua-driver's immutable permission mode.
Normal Hermes sessions use `standard`, which requires a certified protected
host and fails closed when Hermes has none. Explicit Hermes YOLO (`--yolo`,
`/yolo`, or `approvals.mode: off`) launches a private embedded cua-driver in
`unrestricted` after that risk acceptance, so there are no runtime Cua
approval prompts. Never invent, store, log, or reuse a grant token.
Use the native capture/AX/pixel/foreground ladder for browser chrome, browser
permission UI, OS prompts, native dialogs, extension surfaces, unsupported

View File

@ -0,0 +1,184 @@
"""Behavior contracts for cua-driver 0.10 permission-mode integration."""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import Mock, patch
import pytest
@pytest.fixture(autouse=True)
def _reset_computer_use_state():
from tools.computer_use.tool import reset_backend_for_tests
reset_backend_for_tests()
yield
reset_backend_for_tests()
def test_normal_hermes_session_maps_to_standard_mode():
from tools.computer_use import tool as computer_use
with patch(
"tools.approval.is_approval_bypass_active_for_session",
return_value=False,
):
assert computer_use._cua_permission_mode("session-a") == "standard"
def test_any_explicit_hermes_bypass_maps_to_unrestricted_mode():
from tools.computer_use import tool as computer_use
with patch(
"tools.approval.is_approval_bypass_active_for_session",
return_value=True,
):
assert computer_use._cua_permission_mode("session-a") == "unrestricted"
def test_mode_change_replaces_only_that_sessions_backend():
from tools.computer_use import tool as computer_use
created = []
class _Backend:
def __init__(self, permission_mode="standard"):
self.permission_mode = permission_mode
self.stopped = False
created.append(self)
def start(self):
pass
def stop(self):
self.stopped = True
yolo = False
with patch(
"tools.approval.is_approval_bypass_active_for_session",
side_effect=lambda sid: yolo,
), patch(
"tools.computer_use.cua_backend.CuaDriverBackend", _Backend
):
standard = computer_use._get_backend("session-a")
other = computer_use._get_backend("session-b")
yolo = True
unrestricted = computer_use._get_backend("session-a")
assert getattr(standard, "permission_mode") == "standard"
assert getattr(standard, "stopped") is True
assert getattr(unrestricted, "permission_mode") == "unrestricted"
assert unrestricted is not standard
assert getattr(other, "permission_mode") == "standard"
assert getattr(other, "stopped") is False
def test_mode_change_is_rechecked_after_stale_backend_stops():
from tools.computer_use import tool as computer_use
yolo = False
created = []
class _Backend:
def __init__(self, permission_mode="standard"):
self.permission_mode = permission_mode
created.append(self)
def start(self):
pass
def stop(self):
nonlocal yolo
yolo = False
with patch(
"tools.approval.is_approval_bypass_active_for_session",
side_effect=lambda sid: yolo,
), patch("tools.computer_use.cua_backend.CuaDriverBackend", _Backend):
original = computer_use._get_backend("session-a")
yolo = True
replacement = computer_use._get_backend("session-a")
assert getattr(original, "permission_mode") == "standard"
assert getattr(replacement, "permission_mode") == "standard"
assert replacement is not original
assert [backend.permission_mode for backend in created] == [
"standard",
"standard",
]
def test_release_seam_stops_backend_and_clears_session_state():
from tools.computer_use import tool as computer_use
backend = Mock()
computer_use._backends["session-a"] = backend
computer_use._backend_call_locks["session-a"] = computer_use.threading.RLock()
computer_use._backend_permission_modes["session-a"] = "unrestricted"
computer_use._session_auto_approve["session-a"] = True
computer_use._always_allow["session-a"] = {("click", "background")}
assert computer_use.release_computer_use_session("session-a") is True
assert computer_use.release_computer_use_session("session-a") is False
backend.stop.assert_called_once_with()
assert "session-a" not in computer_use._backend_permission_modes
assert "session-a" not in computer_use._session_auto_approve
assert "session-a" not in computer_use._always_allow
def test_yolo_toggle_immediately_releases_mode_dependent_backend():
from tools import approval
with patch("tools.computer_use.release_computer_use_session") as release:
approval.enable_session_yolo("session-a")
approval.disable_session_yolo("session-a")
assert release.call_args_list == [
(('session-a',), {}),
(('session-a',), {}),
]
def test_unrestricted_embedded_daemon_uses_private_socket_and_two_part_ack():
from tools.computer_use import cua_backend
process = Mock()
process.poll.return_value = None
process.stderr = []
process.wait.return_value = 0
status = SimpleNamespace(returncode=0, stdout="running", stderr="")
stopped = SimpleNamespace(returncode=0, stdout="", stderr="")
daemon = cua_backend._EmbeddedCuaDaemon("cua-driver", "unrestricted")
with patch.object(
cua_backend,
"_resolve_mcp_invocation",
return_value=("/opt/cua-driver", ["mcp"]),
), patch.object(cua_backend.subprocess, "Popen", return_value=process) as popen, patch.object(
cua_backend.subprocess, "run", side_effect=[status, stopped]
):
daemon.start()
command = popen.call_args.args[0]
env = popen.call_args.kwargs["env"]
proxy_command, proxy_args = daemon.proxy_invocation()
daemon.stop()
assert command[:2] == ["/opt/cua-driver", "serve"]
assert "--embedded" in command
assert command[command.index("--permission-mode") + 1] == "unrestricted"
assert "--dangerously-bypass-approvals" in command
assert env["CUA_DRIVER_PERMISSION_MODE"] == "unrestricted"
assert env["CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS"] == "1"
assert proxy_command == "/opt/cua-driver"
assert proxy_args == ["mcp", "--embedded", "--socket", daemon.socket_path]
def test_standard_backend_does_not_spawn_an_embedded_daemon():
from tools.computer_use.cua_backend import CuaDriverBackend
standard = CuaDriverBackend(permission_mode="standard")
unrestricted = CuaDriverBackend(permission_mode="unrestricted")
assert standard._embedded_daemon is None
assert unrestricted._embedded_daemon is not None

View File

@ -226,7 +226,8 @@ def test_backends_are_isolated_by_hermes_session_and_reused_within_it():
created = []
class _Backend:
def __init__(self):
def __init__(self, permission_mode="standard"):
self.permission_mode = permission_mode
created.append(self)
def start(self):
@ -324,7 +325,8 @@ def test_concurrent_hermes_sessions_do_not_share_backend_state():
created = []
class _Backend:
def __init__(self):
def __init__(self, permission_mode="standard"):
self.permission_mode = permission_mode
self.marker = len(created)
created.append(self)
@ -700,8 +702,12 @@ def test_missing_typed_browser_tool_returns_native_fallback_refusal():
call.assert_not_called()
def test_existing_profile_prepare_requires_interactive_driver_grant():
def test_existing_profile_prepare_delegates_to_driver_permission_mode():
driver = _BrowserDriver()
driver.responses["browser_prepare"] = {
"status": "refused",
"code": "browser_consent_required",
}
route = _browser_route(driver)
result = route.prepare(
@ -712,8 +718,17 @@ def test_existing_profile_prepare_requires_interactive_driver_grant():
)
assert result["code"] == "browser_consent_required"
assert result["interactive_grant_required"] is True
assert driver.calls == []
assert driver.calls == [
(
"browser_prepare",
{
"pid": 101,
"window_id": 202,
"strategy": {"kind": "existing_profile"},
"session": "hermes-a",
},
)
]
def test_namespaced_state_and_prepare_actions_use_typed_backend_wrappers():

View File

@ -12,6 +12,7 @@ from tools.approval import (
detect_dangerous_command,
disable_session_yolo,
enable_session_yolo,
is_approval_bypass_active_for_session,
is_session_yolo_enabled,
reset_current_session_key,
set_current_session_key,
@ -183,6 +184,16 @@ class TestYoloMode:
disable_session_yolo("session-a")
assert is_session_yolo_enabled("session-a") is False
def test_bypass_query_uses_the_requested_session(self, monkeypatch):
"""Backend mode selection must not leak YOLO across sessions."""
monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False)
monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual")
enable_session_yolo("session-a")
assert is_approval_bypass_active_for_session("session-a") is True
assert is_approval_bypass_active_for_session("session-b") is False
def test_session_scoped_yolo_bypasses_combined_guard_only_for_current_session(self, monkeypatch):
"""Combined guard should honor session-scoped YOLO without affecting others."""
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)

View File

@ -2249,12 +2249,33 @@ def approve_session(session_key: str, pattern_key: str):
_session_approved.setdefault(session_key, set()).add(pattern_key)
def _release_permission_mode_dependents(session_key: str) -> None:
"""Drop resources whose immutable mode is derived from Hermes YOLO.
The import stays lazy so approval-only sessions do not load computer-use.
Releasing on both edges makes enabling YOLO replace an existing standard
backend and makes disabling YOLO revoke a private unrestricted daemon
immediately, even when no later computer-use call occurs.
"""
try:
from tools.computer_use import release_computer_use_session
release_computer_use_session(session_key)
except Exception:
logger.debug(
"Failed to release permission-mode dependent resources for %s",
session_key,
exc_info=True,
)
def enable_session_yolo(session_key: str) -> None:
"""Enable YOLO bypass for a single session key."""
if not session_key:
return
with _lock:
_session_yolo.add(session_key)
_release_permission_mode_dependents(session_key)
def disable_session_yolo(session_key: str) -> None:
@ -2263,6 +2284,7 @@ def disable_session_yolo(session_key: str) -> None:
return
with _lock:
_session_yolo.discard(session_key)
_release_permission_mode_dependents(session_key)
def clear_session(session_key: str) -> None:
@ -2279,6 +2301,7 @@ def clear_session(session_key: str) -> None:
# immediately so the old run can unwind instead of idling until timeout.
entry.result = "deny"
entry.event.set()
_release_permission_mode_dependents(session_key)
def is_session_yolo_enabled(session_key: str) -> bool:
@ -2594,8 +2617,8 @@ def _get_approval_mode() -> str:
return _normalize_approval_mode(mode)
def is_approval_bypass_active() -> bool:
"""Return True when the user has opted out of Hermes approval prompts.
def is_approval_bypass_active_for_session(session_key: str) -> bool:
"""Return whether one exact session bypasses Hermes approval prompts.
Collapses the canonical three-source bypass check used across the codebase
into one place:
@ -2610,11 +2633,18 @@ def is_approval_bypass_active() -> bool:
"""
return (
_YOLO_MODE_FROZEN
or is_current_session_yolo_enabled()
or is_session_yolo_enabled(session_key)
or _get_approval_mode() == "off"
)
def is_approval_bypass_active() -> bool:
"""Return whether the current approval context has bypass enabled."""
return is_approval_bypass_active_for_session(
get_current_session_key(default="")
)
def _get_approval_timeout() -> int:
"""Read the approval timeout from config. Defaults to 300 seconds.

View File

@ -37,6 +37,7 @@ from __future__ import annotations
# Re-export the public surface so `from tools.computer_use import ...` works.
from tools.computer_use.tool import ( # noqa: F401
handle_computer_use,
release_computer_use_session,
set_approval_callback,
check_computer_use_requirements,
get_computer_use_schema,

View File

@ -13,9 +13,9 @@ The adapter is deliberately stricter than the transport:
* every mutation invalidates refs and requires a fresh state read; and
* changing from trusted input to ``dom_event`` is always explicit.
Browser preparation remains a separate approved action. Existing-profile
attachment is not performed here because it needs cua-driver's documented
interactive grant, not ordinary tool approval.
Browser preparation remains a separate approved action. Existing-profile
attachment is delegated to cua-driver's daemon authorization coordinator;
ordinary Hermes tool approval never substitutes for protected consent.
"""
from __future__ import annotations
@ -344,7 +344,7 @@ class CuaTypedBrowserRoute:
profile_name: Optional[str] = None,
allow_launch: bool = False,
) -> Dict[str, Any]:
"""Run explicit isolated setup; refuse existing-profile attachment."""
"""Run explicit setup through the driver's authoritative mode gate."""
missing = self._require_tool("browser_prepare")
if missing is not None:
return missing
@ -354,10 +354,23 @@ class CuaTypedBrowserRoute:
"browser_pid_required", "browser_prepare requires a positive pid."
)
if profile_mode == "existing_profile":
return _refusal(
"browser_consent_required",
"Existing-profile attachment requires cua-driver's interactive browser-approve grant bound to the exact pid, window, and session; ordinary tool approval is insufficient.",
interactive_grant_required=True,
exact_window = _positive_int(window_id)
if exact_window is None:
return _refusal(
"browser_exact_target_required",
"Existing-profile attachment requires an exact positive pid and window_id pair.",
)
# The driver owns the immutable standard/bounded/unrestricted
# decision. Standard fails closed without a certified host;
# explicit Hermes YOLO owns a private unrestricted daemon.
self.state.clear()
return self._call(
"browser_prepare",
{
"pid": exact_pid,
"window_id": exact_window,
"strategy": {"kind": "existing_profile"},
},
)
if profile_mode not in {"isolated_new", "isolated_named"}:
return _refusal(

View File

@ -37,6 +37,7 @@ from __future__ import annotations
import asyncio
import base64
from collections import deque
import concurrent.futures
import functools
import json
@ -46,7 +47,9 @@ import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from pathlib import PureWindowsPath
from typing import Any, Dict, List, Optional, Tuple
@ -377,6 +380,161 @@ def _wsl_windows_path_to_posix(path: str) -> str:
return os.path.join("/mnt", drive, *(str(part) for part in win.parts[1:]))
class _EmbeddedCuaDaemon:
"""Private host-owned daemon used for an explicit unrestricted session.
Cua Driver permission mode is immutable after daemon startup. Reusing the
machine-wide daemon would therefore let one Hermes session's YOLO choice
affect another session. A private embedded daemon gives the requesting
session its own socket, process, and launch-time risk acknowledgement.
"""
_START_TIMEOUT_SECONDS = 15.0
def __init__(self, driver_cmd: str, permission_mode: str) -> None:
if permission_mode != "unrestricted":
raise ValueError("embedded permission override supports unrestricted only")
self.permission_mode = permission_mode
self._driver_cmd = driver_cmd
self._command = driver_cmd
self._mcp_args: List[str] = list(_CUA_DRIVER_ARGS)
self._process: Any = None
self._stderr_tail: deque[str] = deque(maxlen=20)
self._stderr_thread: Optional[threading.Thread] = None
token = uuid.uuid4().hex[:12]
if sys.platform == "win32":
self.socket_path = rf"\\.\pipe\hermes-cua-{token}"
else:
self.socket_path = os.path.join(
tempfile.gettempdir(), f"hc-{token}.sock"
)
def child_env(self) -> Dict[str, str]:
env = cua_driver_child_env()
env["CUA_DRIVER_PERMISSION_MODE"] = "unrestricted"
env["CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS"] = "1"
return env
def _drain_stderr(self, process: Any) -> None:
stream = getattr(process, "stderr", None)
if stream is None:
return
try:
for line in stream:
text = str(line).strip()
if text:
self._stderr_tail.append(text)
logger.debug("embedded cua-driver: %s", text)
except Exception:
pass
def start(self) -> None:
if self._process is not None and self._process.poll() is None:
return
from tools.environments.local import _sanitize_subprocess_env
if not self._driver_cmd:
self._driver_cmd = resolve_cua_driver_cmd() or ""
if not self._driver_cmd:
raise RuntimeError(cua_driver_install_hint())
self._command, self._mcp_args = _resolve_mcp_invocation(self._driver_cmd)
env = _sanitize_subprocess_env(self.child_env())
command = [
self._command,
"serve",
"--embedded",
"--socket",
self.socket_path,
"--no-permissions-gate",
"--permission-mode",
"unrestricted",
"--dangerously-bypass-approvals",
]
self._process = subprocess.Popen(
command,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
env=env,
)
self._stderr_thread = threading.Thread(
target=self._drain_stderr,
args=(self._process,),
name="hermes-cua-daemon-stderr",
daemon=True,
)
self._stderr_thread.start()
deadline = time.monotonic() + self._START_TIMEOUT_SECONDS
while time.monotonic() < deadline:
if self._process.poll() is not None:
detail = "; ".join(self._stderr_tail) or "no diagnostic output"
raise RuntimeError(
f"embedded cua-driver exited during startup: {detail}"
)
try:
probe = subprocess.run(
[self._command, "status", "--socket", self.socket_path],
stdin=subprocess.DEVNULL,
capture_output=True,
text=True,
timeout=2.0,
env=env,
)
except (OSError, subprocess.SubprocessError):
probe = None
if probe is not None and probe.returncode == 0:
return
time.sleep(0.1)
self.stop()
detail = "; ".join(self._stderr_tail) or "daemon did not become ready"
raise RuntimeError(f"embedded cua-driver startup timed out: {detail}")
def proxy_invocation(self) -> Tuple[str, List[str]]:
if self._process is None or self._process.poll() is not None:
raise RuntimeError("embedded cua-driver daemon is not running")
return self._command, [
*self._mcp_args,
"--embedded",
"--socket",
self.socket_path,
]
def stop(self) -> None:
process = self._process
self._process = None
if process is not None and process.poll() is None:
from tools.environments.local import _sanitize_subprocess_env
try:
subprocess.run(
[self._command, "stop", "--socket", self.socket_path],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=3.0,
env=_sanitize_subprocess_env(self.child_env()),
)
except (OSError, subprocess.SubprocessError):
pass
try:
process.wait(timeout=5.0)
except subprocess.TimeoutExpired:
process.terminate()
try:
process.wait(timeout=2.0)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=2.0)
if sys.platform != "win32" and os.path.exists(self.socket_path):
try:
os.remove(self.socket_path)
except OSError:
pass
def _resolve_mcp_invocation(
driver_cmd: str,
*,
@ -925,8 +1083,13 @@ class _CuaDriverSession:
session object, never the surrounding contexts.
"""
def __init__(self, bridge: _AsyncBridge) -> None:
def __init__(
self,
bridge: _AsyncBridge,
embedded_daemon: Optional[_EmbeddedCuaDaemon] = None,
) -> None:
self._bridge = bridge
self._embedded_daemon = embedded_daemon
self._session = None
self._lock = threading.Lock()
self._started = False
@ -989,14 +1152,19 @@ class _CuaDriverSession:
# the MCP server, instead of hardcoding ["mcp"]. Falls back
# transparently for older drivers / any discovery failure.
self._startup_phase = "manifest-discovery"
command, args = _resolve_mcp_invocation(driver_cmd)
if self._embedded_daemon is not None:
command, args = self._embedded_daemon.proxy_invocation()
child_env = self._embedded_daemon.child_env()
else:
command, args = _resolve_mcp_invocation(driver_cmd)
child_env = cua_driver_child_env()
_t_manifest = _time.monotonic()
params = StdioServerParameters(
command=command,
args=args,
# Apply the telemetry policy first (default: disabled), then
# sanitize Hermes-managed secrets out of the child env.
env=_sanitize_subprocess_env(cua_driver_child_env()),
env=_sanitize_subprocess_env(child_env),
)
async with stdio_client(params) as (read, write):
@ -1379,10 +1547,23 @@ class _CuaDriverSession:
os.close(fd)
call_args["screenshot_out_file"] = shot_file
driver_cmd = resolve_cua_driver_cmd()
if not driver_cmd:
driver_command = resolve_cua_driver_cmd()
if not driver_command:
raise RuntimeError(cua_driver_install_hint())
cmd = [driver_cmd, "call", name, json.dumps(call_args)]
child_env = cua_driver_child_env()
socket_args: List[str] = []
embedded_daemon = getattr(self, "_embedded_daemon", None)
if embedded_daemon is not None:
driver_command = embedded_daemon.proxy_invocation()[0]
child_env = embedded_daemon.child_env()
socket_args = ["--socket", embedded_daemon.socket_path]
cmd = [
driver_command,
"call",
name,
json.dumps(call_args),
*socket_args,
]
attempts = 4
backoff = 0.5
parsed: Any = None
@ -1393,7 +1574,7 @@ class _CuaDriverSession:
proc = _subprocess.run(
cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=max(15.0, timeout),
creationflags=windows_hide_flags(),
env=_sanitize_subprocess_env(cua_driver_child_env()),
env=_sanitize_subprocess_env(child_env),
)
except Exception as e: # pragma: no cover - subprocess spawn failure
raise RuntimeError(f"cua-driver CLI fallback for {name} failed to spawn: {e}") from e
@ -1726,9 +1907,17 @@ def _apps_from_windows(windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
class CuaDriverBackend(ComputerUseBackend):
"""Default computer-use backend. Cross-platform via cua-driver MCP."""
def __init__(self) -> None:
def __init__(self, permission_mode: str = "standard") -> None:
if permission_mode not in {"standard", "unrestricted"}:
raise ValueError(f"unsupported cua-driver permission mode: {permission_mode}")
self.permission_mode = permission_mode
self._embedded_daemon = (
_EmbeddedCuaDaemon(resolve_cua_driver_cmd() or "", permission_mode)
if permission_mode == "unrestricted"
else None
)
self._bridge = _AsyncBridge()
self._session = _CuaDriverSession(self._bridge)
self._session = _CuaDriverSession(self._bridge, self._embedded_daemon)
# Sticky context — updated by capture(), used by action tools.
self._active_pid: Optional[int] = None
self._active_window_id: Optional[int] = None
@ -1797,7 +1986,14 @@ class CuaDriverBackend(ComputerUseBackend):
# machinery's caches are refreshed within this process.
import importlib
importlib.invalidate_caches()
self._session.start()
try:
if self._embedded_daemon is not None:
self._embedded_daemon.start()
self._session.start()
except Exception:
if self._embedded_daemon is not None:
self._embedded_daemon.stop()
raise
# Declare the run's session identity to cua-driver. From the
# cua-driver server instructions: "start_session(session) once
@ -1848,7 +2044,11 @@ class CuaDriverBackend(ComputerUseBackend):
try:
self._session.stop()
finally:
self._bridge.stop()
try:
self._bridge.stop()
finally:
if self._embedded_daemon is not None:
self._embedded_daemon.stop()
def is_available(self) -> bool:
# cua-driver runs on macOS, Windows, and Linux. The Linux path is

View File

@ -290,8 +290,10 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = {
"type": "string",
"enum": ["isolated_new", "isolated_named", "existing_profile"],
"description": (
"Browser preparation mode. existing_profile always requires "
"the driver's separate interactive grant."
"Browser preparation mode. existing_profile is decided by "
"cua-driver's immutable permission mode: standard requires a "
"certified protected host; explicit Hermes YOLO uses a private "
"unrestricted daemon."
),
},
"profile_name": {"type": "string", "description": "Name for isolated_named setup."},

View File

@ -155,6 +155,7 @@ _AUX_VISION_ROUTE_CACHE: Dict[Tuple[str, str], bool] = {}
_backend: Optional[ComputerUseBackend] = None
_backends: Dict[str, ComputerUseBackend] = {}
_backend_call_locks: Dict[str, threading.RLock] = {}
_backend_permission_modes: Dict[str, str] = {}
# Approval state, scoped per conversation/run (keyed by session_id) so a
# gateway serving concurrent sessions can't leak one run's "always approve"
# unlock into another. Falls back to a shared "" bucket for callers that
@ -167,45 +168,97 @@ _session_auto_approve: Dict[str, bool] = {}
_always_allow: Dict[str, set] = {}
def _cua_permission_mode(session_id: str) -> str:
"""Map Hermes's explicit approval bypass onto Cua's immutable mode."""
try:
from tools.approval import (
is_approval_bypass_active_for_session,
)
if is_approval_bypass_active_for_session(session_id):
return "unrestricted"
except Exception:
# Approval state must fail closed if it cannot be resolved.
pass
return "standard"
def _get_backend(session_id: str = "") -> ComputerUseBackend:
global _backend
sid = str(session_id or "")
with _backend_lock:
if sid == "" and _backend is not None:
return _backend
cached = _backends.get(sid)
if cached is not None:
return cached
backend_name = os.environ.get("HERMES_COMPUTER_USE_BACKEND", "cua").lower()
if backend_name in {"cua", "cua-driver", ""}:
from tools.computer_use.cua_backend import CuaDriverBackend
while True:
stale_backend: Optional[ComputerUseBackend] = None
stale_lock: Optional[threading.RLock] = None
with _backend_lock:
# Resolve the mode while holding the cache lock. Session YOLO
# mutation never holds the approval lock while releasing this
# cache, so the lock order cannot cycle.
permission_mode = _cua_permission_mode(sid)
if sid == "" and _backend is not None and sid not in _backends:
# Preserve the long-standing empty-session injection hook used
# by integrations and tests while normalizing it into the
# session-owned cache/lifecycle path.
_backends[sid] = _backend
_backend_call_locks[sid] = threading.RLock()
_backend_permission_modes[sid] = permission_mode
cached = _backends.get(sid)
if cached is not None:
if _backend_permission_modes.get(sid, "standard") == permission_mode:
return cached
# Cua's permission mode cannot change after daemon startup. A
# /yolo toggle replaces only this session's backend.
stale_backend = _backends.pop(sid)
stale_lock = _backend_call_locks.pop(sid, None)
_backend_permission_modes.pop(sid, None)
if sid == "":
_backend = None
else:
backend_name = os.environ.get(
"HERMES_COMPUTER_USE_BACKEND", "cua"
).lower()
if backend_name in {"cua", "cua-driver", ""}:
from tools.computer_use.cua_backend import CuaDriverBackend
backend = CuaDriverBackend()
elif backend_name == "noop": # pragma: no cover
backend = _NoopBackend()
else:
raise RuntimeError(f"Unknown HERMES_COMPUTER_USE_BACKEND={backend_name!r}")
backend = CuaDriverBackend(permission_mode=permission_mode)
elif backend_name == "noop": # pragma: no cover
backend = _NoopBackend()
else:
raise RuntimeError(
f"Unknown HERMES_COMPUTER_USE_BACKEND={backend_name!r}"
)
# Starting under the cache lock preserves the existing
# one-backend-per-session invariant. A concurrent mode toggle
# releases this backend before returning to its caller.
backend.start()
_backends[sid] = backend
_backend_call_locks[sid] = threading.RLock()
_backend_permission_modes[sid] = permission_mode
if sid == "":
_backend = backend
return backend
# Stop a mismatched backend outside the global cache lock. Another
# session can continue creating or releasing its own backend, and the
# loop re-reads the authoritative mode before installing a replacement.
try:
backend.start()
if stale_lock is not None:
with stale_lock:
stale_backend.stop()
elif stale_backend is not None:
stale_backend.stop()
except Exception:
# Don't cache a backend whose start() failed (e.g. a lazy
# dependency install was declined / failed). The next call
# retries cleanly instead of returning a half-initialised backend.
raise
_backends[sid] = backend
_backend_call_locks[sid] = threading.RLock()
if sid == "":
_backend = backend
return backend
pass
def release_computer_use_session(session_id: str) -> bool:
"""Release one session-owned computer-use backend.
This is the production lifecycle seam for hosts and policy plugins. It
removes the exact session backend and its call lock before stopping the
backend, so new lookups cannot retain the stale target/ref namespace.
Approval state is cleared even when no backend was started.
removes the exact session backend, its call lock, and its recorded
permission mode before stopping the backend, so new lookups cannot retain
the stale target/ref namespace and stops a private embedded daemon when
Hermes YOLO selected unrestricted mode. Approval state is cleared even
when no backend was started.
Returns ``True`` when a backend was found and released, ``False`` when the
session was already absent. Safe to call repeatedly.
@ -215,6 +268,7 @@ def release_computer_use_session(session_id: str) -> bool:
with _backend_lock:
backend = _backends.pop(sid, None)
call_lock = _backend_call_locks.pop(sid, None)
_backend_permission_modes.pop(sid, None)
# Preserve the backward-compatible empty-session injection hook:
# older callers/tests may populate only `_backend`.
if sid == "" and backend is None:
@ -276,6 +330,7 @@ def _shutdown_backend_atexit() -> None:
_backend = None
_backends.clear()
_backend_call_locks.clear()
_backend_permission_modes.clear()
with _approval_lock:
_session_auto_approve.clear()
@ -299,9 +354,6 @@ def reset_backend_for_tests() -> None: # pragma: no cover
"""Test helper — tear down the cached backend and per-session state."""
_shutdown_backend_atexit()
_AUX_VISION_ROUTE_CACHE.clear()
with _approval_lock:
_session_auto_approve.clear()
_always_allow.clear()
class _NoopBackend(ComputerUseBackend): # pragma: no cover
@ -379,7 +431,8 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
action = (args.get("action") or "").strip().lower()
if not action:
return json.dumps({"error": "missing `action`"})
# Per-run key for approval-state isolation across concurrent sessions.
# Per-run key for approval-state and daemon-mode isolation across
# concurrent sessions.
session_id = str(kwargs.get("session_id") or "")
# Safety: validate actions before approval prompt.

View File

@ -35,6 +35,7 @@ registry.register(
__all__ = [
"handle_computer_use",
"release_computer_use_session",
"set_approval_callback",
"check_computer_use_requirements",
"release_computer_use_session",

View File

@ -75,6 +75,34 @@ hermes -t computer_use chat
or add `computer_use` to your enabled toolsets in `~/.hermes/config.yaml`.
## Permission modes and logged-in browser profiles
Hermes maps its existing approval UX onto cua-driver 0.10's immutable daemon
modes. There is no second permission toggle to keep in sync:
| Hermes session | cua-driver mode | Human intervention | `existing_profile` |
|---|---|---|---|
| Manual or smart approvals (default) | `standard` | Normal Hermes approvals; Cua stops at its protected boundary | Refuses unless a certified protected host is available; Hermes does not claim one today |
| `--yolo`, `/yolo`, or `approvals.mode: off` | private `unrestricted` daemon | One explicit Hermes risk acceptance; no runtime Cua prompts | Allowed within Cua's built-in, managed, and user policy ceilings |
The unrestricted daemon is private to that Hermes session. Turning `/yolo`
off, resetting/closing the session, cancellation cleanup, or process exit ends
the Cua session and stops that daemon. It never changes the machine-wide
daemon's mode or grants another Hermes conversation the same authority.
`smart` approval remains `standard`: an LLM classification is not protected
human consent. Cua's `bounded` manifest mode is also not inferred from smart
approval or a normal tool confirmation; it needs a separately trusted host
that reviews and launches the exact manifest.
<div class="alert alert--warning">
YOLO/unrestricted mode does not protect against prompt injection or unintended
input. Use it only in a disposable VM or with accounts and data whose full
compromise you accept.
</div>
## `hermes computer-use doctor` — your first triage stop
`hermes computer-use doctor` runs cua-driver's structured
@ -390,14 +418,12 @@ HERMES_CUA_DRIVER_CMD=/path/to/cua/libs/cua-driver/rust/target/debug/cua-driver
### Notes & gotchas
- **Hermes spawns its own `cua-driver mcp` child over stdio** — it does
*not* attach to the long-running `cua-driver serve` autostart daemon
or its named pipe. So the scheduled task / LaunchAgent is unnecessary
for testing (`-NoAutoStart` is fine). The autostart daemon and the
Windows UIAccess worker (`cua-driver-uia.exe`) only matter for
foreground-safe input on some apps (e.g. WPF); the standard tool
surface works through the stdio child. On Windows SSH sessions, the
autostart pattern IS needed — see the Limitations section.
- **Hermes spawns a `cua-driver mcp` stdio proxy.** In a normal session the
proxy connects to (and may start) the standard machine daemon. In explicit
Hermes YOLO, Hermes instead owns a private `cua-driver serve --embedded`
child and points the proxy at its private socket or named pipe. The Windows
autostart/UIAccess pattern still matters for interactive Session 1+ input
from SSH — see the Limitations section.
- **Locked binary on Windows.** A running `cua-driver-serve` daemon can
hold `cua-driver.exe` and block an overwrite on rebuild.
`install-local.ps1` renames the locked binary out of the way