ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

simplify(compat): tools/computer_use/cua_backend — drop 24 re-exports + shutil re-import, repoint 5 callers + 14 test files

Browse Source
This commit is contained in:
Teknium
2026-09-03 13:22:51 -07:00
parent 2dcd97d648
commit 5c3db6750f
23 changed files with 161 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/subcommands/computer_use.py
+1 -1
View File
@@ -18,7 +18,7 @@ def _cu_status(args) -> int:
import os as _os
import subprocess
from hermes_cli.tools_config import _cua_driver_contract_status
from tools.computer_use.cua_backend import cua_driver_update_check, resolve_cua_driver_cmd
from tools.computer_use.cua_backend_driver import cua_driver_update_check, resolve_cua_driver_cmd
# Must match the runtime resolver: Desktop/TUI processes can omit
# ~/.local/bin even though the official installer put the driver there.
path = resolve_cua_driver_cmd()
hermes_cli/tools_config_cua.py
+3 -3
View File
@@ -98,7 +98,7 @@ def _cua_version_summary(raw: str, *, limit: int = 120) -> str:
def _resolved_cua_driver_cmd() -> Optional[str]:
"""Resolve cua-driver exactly as the runtime and Desktop status do."""
from tools.computer_use.cua_backend import resolve_cua_driver_cmd
from tools.computer_use.cua_backend_driver import resolve_cua_driver_cmd
return resolve_cua_driver_cmd()
@@ -118,7 +118,7 @@ _CUA_DRIVER_CONTRACT_CACHE: dict = {}
def _cua_driver_contract_status(binary: Optional[str] = None) -> dict:
"""Inspect whether an installed driver supports Hermes' runtime contract (30s cache keyed on the
binary's path/mtime/size fingerprint)."""
from tools.computer_use.cua_backend import cua_driver_runtime_contract_status
from tools.computer_use.cua_backend_driver import cua_driver_runtime_contract_status
resolved = binary or _resolved_cua_driver_cmd()
if not resolved:
return cua_driver_runtime_contract_status(None)
@@ -214,7 +214,7 @@ def _confirmed_update_check(driver_cmd: str, require_confirmed_update: bool) ->
cost a multi-minute silent reinstall on every update — while explicit `install --upgrade`
falls through."""
try:
from tools.computer_use.cua_backend import cua_driver_update_check
from tools.computer_use.cua_backend_driver import cua_driver_update_check
_state = cua_driver_update_check()
except Exception:
_state = None
tests/computer_use/test_cua_cli_fallback_env.py
+2 -2
View File
@@ -12,7 +12,7 @@ secrets that ``_lifecycle_coro``'s primary MCP spawn already strips via
import json
from unittest.mock import MagicMock
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
def _make_session() -> _CuaDriverSession:
@@ -34,7 +34,7 @@ def test_cli_fallback_strips_provider_secret_from_subprocess_env(monkeypatch):
monkeypatch.setenv("ANTHROPIC_API_KEY", "«redacted:sk-…»")
monkeypatch.setenv("PATH", "/usr/bin:/bin")
monkeypatch.setattr(
"tools.computer_use.cua_backend.resolve_cua_driver_cmd",
"tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
lambda: "/resolved/cua-driver",
)
tests/computer_use/test_cua_no_overlay.py
+18 -17
View File
@@ -16,6 +16,7 @@ from unittest.mock import MagicMock, mock_open, patch
import pytest
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
class TestNoOverlayFlag:
@@ -108,7 +109,7 @@ class TestDriverSupportsNoOverlay:
with patch("subprocess.run") as mock_run:
mock_run.return_value.stdout = fake_help
mock_run.return_value.stderr = ""
assert cua_backend._cua_driver_supports_no_overlay("cua-driver") is True
assert cua_backend_driver._cua_driver_supports_no_overlay("cua-driver") is True
@@ -120,8 +121,8 @@ class TestDriverSupportsNoOverlay:
from unittest.mock import MagicMock
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(stdout="--no-overlay in help", stderr="")
cua_backend._cua_driver_supports_no_overlay.cache_clear()
cua_backend._cua_driver_supports_no_overlay("cua-driver")
cua_backend_driver._cua_driver_supports_no_overlay.cache_clear()
cua_backend_driver._cua_driver_supports_no_overlay("cua-driver")
kwargs = mock_run.call_args.kwargs
assert "env" in kwargs, (
"subprocess.run was called without env= — cua-driver is a "
@@ -160,7 +161,7 @@ class TestMcpInvocationUsesResolvedCommand:
``driver_cmd`` parameter.
"""
from unittest.mock import patch
from tools.computer_use.cua_backend import _resolve_mcp_invocation
from tools.computer_use.cua_backend_driver import _resolve_mcp_invocation
manifest = (
'{"mcp_invocation":'
@@ -169,10 +170,10 @@ class TestMcpInvocationUsesResolvedCommand:
with patch("subprocess.run", new=self._fake_run(stdout=manifest)), \
patch.object(cua_backend, "_cua_no_overlay", return_value=True), \
patch.object(
cua_backend, "_cua_driver_supports_no_overlay",
cua_backend_driver, "_cua_driver_supports_no_overlay",
return_value=True,
) as mock_probe:
cua_backend._cua_driver_supports_no_overlay.cache_clear()
cua_backend_driver._cua_driver_supports_no_overlay.cache_clear()
cmd, args = _resolve_mcp_invocation("/usr/bin/cua-driver")
assert cmd == "/opt/relocated/cua-driver"
# The support probe must be called with the manifest-resolved
@@ -188,15 +189,15 @@ class TestMcpInvocationUsesResolvedCommand:
"""
with patch.object(cua_backend, "_cua_no_overlay", return_value=True), \
patch.object(
cua_backend, "_cua_driver_supports_no_overlay",
cua_backend_driver, "_cua_driver_supports_no_overlay",
side_effect=lambda cmd: cmd == "/opt/relocated/cua-driver",
):
# System binary does NOT support, manifest binary DOES.
args = cua_backend._mcp_args_with_overlay_flag(
args = cua_backend_driver._mcp_args_with_overlay_flag(
["mcp"], driver_cmd="/usr/bin/cua-driver",
)
assert "--no-overlay" not in args
args = cua_backend._mcp_args_with_overlay_flag(
args = cua_backend_driver._mcp_args_with_overlay_flag(
["mcp"], driver_cmd="/opt/relocated/cua-driver",
)
assert "--no-overlay" in args
@@ -205,22 +206,22 @@ class TestMcpInvocationUsesResolvedCommand:
class TestMcpArgsOverlayFlag:
def test_appended_when_enabled_and_supported(self):
with patch.object(cua_backend, "_cua_no_overlay", return_value=True), \
patch.object(cua_backend, "_cua_driver_supports_no_overlay", return_value=True):
result = cua_backend._mcp_args_with_overlay_flag(["mcp"])
patch.object(cua_backend_driver, "_cua_driver_supports_no_overlay", return_value=True):
result = cua_backend_driver._mcp_args_with_overlay_flag(["mcp"])
assert result == ["mcp", "--no-overlay"]
def test_not_appended_when_disabled(self):
with patch.object(cua_backend, "_cua_no_overlay", return_value=False), \
patch.object(cua_backend, "_cua_driver_supports_no_overlay", return_value=True):
result = cua_backend._mcp_args_with_overlay_flag(["mcp"])
patch.object(cua_backend_driver, "_cua_driver_supports_no_overlay", return_value=True):
result = cua_backend_driver._mcp_args_with_overlay_flag(["mcp"])
assert result == ["mcp"]
def test_does_not_mutate_original_list(self):
original = ["mcp"]
with patch.object(cua_backend, "_cua_no_overlay", return_value=True), \
patch.object(cua_backend, "_cua_driver_supports_no_overlay", return_value=True):
result = cua_backend._mcp_args_with_overlay_flag(original)
patch.object(cua_backend_driver, "_cua_driver_supports_no_overlay", return_value=True):
result = cua_backend_driver._mcp_args_with_overlay_flag(original)
assert "--no-overlay" in result
assert "--no-overlay" not in original
@@ -233,13 +234,13 @@ class TestEmbeddedDaemonOverlayFlag:
status = MagicMock(returncode=0)
with patch.object(
cua_backend,
cua_backend_driver,
"_resolve_mcp_invocation",
return_value=("/usr/bin/cua-driver", ["mcp"]),
), patch.object(
cua_backend, "_cua_no_overlay", return_value=True,
), patch.object(
cua_backend, "_cua_driver_supports_no_overlay", return_value=True,
cua_backend_driver, "_cua_driver_supports_no_overlay", return_value=True,
), patch.object(
cua_backend.subprocess, "Popen", return_value=process,
) as popen, patch.object(
tests/computer_use/test_cua_spawn_env_sanitization.py
+10 -8
View File
@@ -5,9 +5,9 @@ PR #58889 fixed the CLI-fallback transport; review of that fix found four
sibling spawn sites still handing the third-party ``cua-driver`` binary the
full parent environment (provider API keys included):
- ``cua_backend._resolve_mcp_invocation`` (``cua-driver manifest``) — no
- ``cua_backend_driver._resolve_mcp_invocation`` (``cua-driver manifest``) — no
``env=`` at all
- ``cua_backend.cua_driver_update_check`` (``check-update --json``) —
- ``cua_backend_driver.cua_driver_update_check`` (``check-update --json``) —
telemetry env but no secret sanitization
- ``doctor._drive_health_report`` (``<binary> mcp``) — telemetry env only
- ``permissions._run`` (every permission probe) — telemetry env only
@@ -82,6 +82,7 @@ def test_resolve_mcp_invocation_sanitizes_env(monkeypatch):
monkeypatch.delenv("HERMES_CUA_TELEMETRY", raising=False)
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
captured = {}
_patch_windows_hide_flags(monkeypatch, cua_backend)
@@ -90,7 +91,7 @@ def test_resolve_mcp_invocation_sanitizes_env(monkeypatch):
cua_backend.subprocess, "run", _capture_run(captured, stdout=manifest)
)
cmd, args = cua_backend._resolve_mcp_invocation("cua-driver")
cmd, args = cua_backend_driver._resolve_mcp_invocation("cua-driver")
assert cmd == "cua-driver"
_assert_sanitized(captured)
assert captured["creationflags"] == CREATE_NO_WINDOW
@@ -102,6 +103,7 @@ def test_update_check_sanitizes_env(monkeypatch):
monkeypatch.delenv("HERMES_CUA_TELEMETRY", raising=False)
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
captured = {}
_patch_windows_hide_flags(monkeypatch, cua_backend)
@@ -113,13 +115,13 @@ def test_update_check_sanitizes_env(monkeypatch):
# PATH is pinned to /usr/bin:/bin above, so the driver won't resolve;
# pin it so the check reaches the (sanitized) subprocess spawn.
monkeypatch.setattr(
cua_backend, "resolve_cua_driver_cmd", lambda *a, **k: "cua-driver"
cua_backend_driver, "resolve_cua_driver_cmd", lambda *a, **k: "cua-driver"
)
monkeypatch.setattr(
cua_backend.subprocess, "run", _capture_run(captured, stdout=payload)
)
cua_backend.cua_driver_update_check(timeout=1.0)
cua_backend_driver.cua_driver_update_check(timeout=1.0)
_assert_sanitized(captured)
assert captured["creationflags"] == CREATE_NO_WINDOW
@@ -129,14 +131,14 @@ def test_cli_fallback_sanitizes_env_and_hides_console_on_windows(monkeypatch):
monkeypatch.setenv("PATH", "/usr/bin:/bin")
monkeypatch.delenv("HERMES_CUA_TELEMETRY", raising=False)
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend, cua_backend_driver, cua_backend_session
captured = {}
_patch_windows_hide_flags(monkeypatch, cua_backend)
_patch_windows_hide_flags(monkeypatch, cua_backend_session)
# Hermetic CI has no cua-driver binary; pin the resolver so the test
# exercises the spawn-env path instead of the install-hint early exit.
monkeypatch.setattr(
cua_backend, "resolve_cua_driver_cmd", lambda override=None: "cua-driver"
cua_backend_driver, "resolve_cua_driver_cmd", lambda override=None: "cua-driver"
)
monkeypatch.setattr(
cua_backend.subprocess,
tests/computer_use/test_cua_wsl_manifest_path.py
+3 -2
View File
@@ -5,11 +5,12 @@ from types import SimpleNamespace
from unittest.mock import patch
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
def test_wsl_windows_manifest_path_translates_to_drvfs():
with patch("hermes_constants.is_wsl", return_value=True):
assert cua_backend._wsl_windows_path_to_posix(
assert cua_backend_driver._wsl_windows_path_to_posix(
r"C:\Users\Fernando\AppData\Local\cua-driver\cua-driver.exe"
) == "/mnt/c/Users/Fernando/AppData/Local/cua-driver/cua-driver.exe"
@@ -30,7 +31,7 @@ def test_resolve_mcp_invocation_normalizes_windows_manifest_command_in_wsl():
patch.object(cua_backend.subprocess, "run", return_value=proc),
patch("hermes_constants.is_wsl", return_value=True),
):
command, args = cua_backend._resolve_mcp_invocation("cua-driver")
command, args = cua_backend_driver._resolve_mcp_invocation("cua-driver")
assert command == "/mnt/c/Users/Fernando/AppData/Local/cua-driver/cua-driver.exe"
assert args == ["mcp"]
tests/hermes_cli/test_computer_use_cli.py
+7 -9
View File
@@ -9,6 +9,8 @@ from unittest.mock import Mock
import pytest
from tools.computer_use import cua_backend_driver
def _run(*args: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
@@ -51,18 +53,17 @@ def test_computer_use_status_returns_zero_for_compatible_driver(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from hermes_cli import tools_config
from tools.computer_use import cua_backend
driver = r"C:\Users\tester\.local\bin\cua-driver.exe"
monkeypatch.delenv("HERMES_CUA_DRIVER_CMD", raising=False)
monkeypatch.setattr(cua_backend, "resolve_cua_driver_cmd", lambda: driver)
monkeypatch.setattr(cua_backend_driver, "resolve_cua_driver_cmd", lambda: driver)
monkeypatch.setattr(
tools_config,
"_cua_driver_contract_status",
lambda _binary=None: {"ready": True},
)
monkeypatch.setattr(
cua_backend,
cua_backend_driver,
"cua_driver_update_check",
lambda: {"update_available": False},
)
@@ -74,9 +75,8 @@ def test_computer_use_status_returns_nonzero_when_driver_is_missing(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
from tools.computer_use import cua_backend
monkeypatch.setattr(cua_backend, "resolve_cua_driver_cmd", lambda: None)
monkeypatch.setattr(cua_backend_driver, "resolve_cua_driver_cmd", lambda: None)
assert _invoke(monkeypatch, "status") == 1
assert "cua-driver: not installed" in capsys.readouterr().out
@@ -87,11 +87,10 @@ def test_computer_use_status_returns_nonzero_for_incompatible_standard_driver(
capsys: pytest.CaptureFixture[str],
) -> None:
from hermes_cli import tools_config
from tools.computer_use import cua_backend
driver = r"C:\Users\tester\.local\bin\cua-driver.exe"
monkeypatch.delenv("HERMES_CUA_DRIVER_CMD", raising=False)
monkeypatch.setattr(cua_backend, "resolve_cua_driver_cmd", lambda: driver)
monkeypatch.setattr(cua_backend_driver, "resolve_cua_driver_cmd", lambda: driver)
monkeypatch.setattr(
tools_config,
"_cua_driver_contract_status",
@@ -112,11 +111,10 @@ def test_computer_use_status_returns_nonzero_for_incompatible_custom_driver(
capsys: pytest.CaptureFixture[str],
) -> None:
from hermes_cli import tools_config
from tools.computer_use import cua_backend
driver = r"C:\custom\cmd.exe"
monkeypatch.setenv("HERMES_CUA_DRIVER_CMD", driver)
monkeypatch.setattr(cua_backend, "resolve_cua_driver_cmd", lambda: driver)
monkeypatch.setattr(cua_backend_driver, "resolve_cua_driver_cmd", lambda: driver)
monkeypatch.setattr(
tools_config,
"_cua_driver_contract_status",
tests/hermes_cli/test_install_cua_driver.py
+9 -9
View File
@@ -515,7 +515,7 @@ class TestRequireConfirmedUpdate:
"reason": "",
},
), \
patch("tools.computer_use.cua_backend.cua_driver_update_check",
patch("tools.computer_use.cua_backend_driver.cua_driver_update_check",
return_value=check_state), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner, \
@@ -657,7 +657,7 @@ class TestRequireConfirmedUpdate:
side_effect=[incompatible,
{"ready": True, "version": "0.20.0", "reason": ""}],
), \
patch("tools.computer_use.cua_backend.cua_driver_update_check",
patch("tools.computer_use.cua_backend_driver.cua_driver_update_check",
return_value=None) as check, \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner, \
@@ -689,7 +689,7 @@ class TestUpdateCheckTimeoutDefaults:
def _captured_timeout(self):
from unittest.mock import MagicMock
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
captured = {}
@@ -699,11 +699,11 @@ class TestUpdateCheckTimeoutDefaults:
m.stdout = '{"update_available": false, "current_version": "1.0"}'
return m
with patch("tools.computer_use.cua_backend.resolve_cua_driver_cmd",
with patch("tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
return_value="/x/cua-driver"), \
patch("tools.computer_use.cua_backend.subprocess.run",
side_effect=fake_run):
cua_backend.cua_driver_update_check()
cua_backend_driver.cua_driver_update_check()
return captured.get("timeout")
@pytest.mark.windows_only
@@ -721,7 +721,7 @@ class TestUpdateCheckTimeoutDefaults:
def test_explicit_timeout_wins(self):
from unittest.mock import MagicMock
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
captured = {}
@@ -731,11 +731,11 @@ class TestUpdateCheckTimeoutDefaults:
m.stdout = "{}"
return m
with patch("tools.computer_use.cua_backend.resolve_cua_driver_cmd",
with patch("tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
return_value="/x/cua-driver"), \
patch("tools.computer_use.cua_backend.subprocess.run",
side_effect=fake_run):
cua_backend.cua_driver_update_check(timeout=3.0)
cua_backend_driver.cua_driver_update_check(timeout=3.0)
assert captured.get("timeout") == 3.0
@@ -1421,7 +1421,7 @@ class TestConfirmedVersionPinning:
"_cua_driver_contract_status",
return_value={"ready": True, "version": "0.20.0", "reason": ""},
), \
patch("tools.computer_use.cua_backend.cua_driver_update_check",
patch("tools.computer_use.cua_backend_driver.cua_driver_update_check",
return_value=check_state), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner, \
tests/tools/test_computer_use.py
+30 -27
View File
@@ -72,6 +72,7 @@ class TestRegistration:
def test_cua_driver_cmd_env_override_is_resolved_dynamically(self, tmp_path, monkeypatch):
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
driver = tmp_path / "custom-cua-driver"
driver.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
@@ -80,7 +81,7 @@ class TestRegistration:
monkeypatch.setenv("HERMES_CUA_DRIVER_CMD", str(driver))
monkeypatch.setenv("PATH", "/usr/bin:/bin")
assert cua_backend.resolve_cua_driver_cmd() == str(driver)
assert cua_backend_driver.resolve_cua_driver_cmd() == str(driver)
assert cua_backend.cua_driver_binary_available() is True
@@ -391,7 +392,7 @@ class TestCaptureResponse:
class TestCuaCaptureImageDimensions:
def test_png_dimensions_are_sniffed_from_image_bytes(self):
from tools.computer_use.cua_backend import _image_dimensions_from_bytes
from tools.computer_use.cua_backend_parse import _image_dimensions_from_bytes
raw_png = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42m"
@@ -602,7 +603,7 @@ class TestImageAwareTokenEstimator:
class TestRunAgentMultimodalHelpers:
def test_append_subdir_hint_to_multimodal_appends_to_text_part(self):
from run_agent import _append_subdir_hint_to_multimodal
from agent.tool_dispatch_helpers import _append_subdir_hint_to_multimodal
env = {
"_multimodal": True,
"content": [
@@ -667,7 +668,7 @@ class TestElementLabelParsing:
"""
def test_classic_quoted_label_format(self):
from tools.computer_use.cua_backend import _parse_elements_from_tree
from tools.computer_use.cua_backend_parse import _parse_elements_from_tree
tree = (
' - [14] AXButton "One"\n'
' - [15] AXButton "Two"\n'
@@ -683,7 +684,7 @@ class TestElementLabelParsing:
def test_new_id_eq_format(self):
"""cua-driver v0.1.6 format: [N] AXRole (order) id=Label"""
from tools.computer_use.cua_backend import _parse_elements_from_tree
from tools.computer_use.cua_backend_parse import _parse_elements_from_tree
tree = (
"[14] AXButton (1) id=One\n"
"[15] AXButton (2) id=Two\n"
@@ -705,7 +706,7 @@ class TestElementLabelParsing:
the regex only matched the quoted and id= forms. A pure-digit (N) is an
order number, not a label, and must be skipped in favour of id=.
"""
from tools.computer_use.cua_backend import _parse_elements_from_tree
from tools.computer_use.cua_backend_parse import _parse_elements_from_tree
tree = (
'- [77] AXButton (Auto) [help="..." actions=[press]]\n'
'- [78] AXButton (Light) [help="..." actions=[press]]\n'
@@ -741,7 +742,7 @@ class TestUpdateCheck:
# The update check now short-circuits to None when no driver
# resolves; CI has none installed, so pin a resolved path.
with patch(
"tools.computer_use.cua_backend.resolve_cua_driver_cmd",
"tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
return_value="/usr/local/bin/cua-driver",
):
yield
@@ -754,9 +755,10 @@ class TestUpdateCheck:
def test_update_available(self):
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
payload = '{"current_version":"0.3.1","latest_version":"0.3.2","update_available":true}'
with self._run_returning(payload):
st = cua_backend.cua_driver_update_check()
st = cua_backend_driver.cua_driver_update_check()
assert st is not None and st["update_available"] is True
msg = cua_backend.cua_driver_update_nudge()
assert msg is not None
@@ -764,9 +766,10 @@ class TestUpdateCheck:
def test_error_payload_is_indeterminate(self):
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_driver
payload = '{"current_version":"0.3.2","update_available":false,"error":"github 503"}'
with self._run_returning(payload):
assert cua_backend.cua_driver_update_check() is None
assert cua_backend_driver.cua_driver_update_check() is None
assert cua_backend.cua_driver_update_nudge() is None
class TestLazyMcpInstall:
@@ -1154,7 +1157,7 @@ def _make_cua_backend_with_tool_result(result: Dict[str, Any]):
class TestCuaDriverWindowResultShapes:
def test_extracts_windows_from_structured_content(self):
from tools.computer_use.cua_backend import _windows_from_tool_result
from tools.computer_use.cua_backend_parse import _windows_from_tool_result
windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}]
@@ -1195,7 +1198,7 @@ class TestCuaDriverSessionReconnect:
def _make_session(self, bridge):
import threading
from typing import Any, cast
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
session = cast(Any, _CuaDriverSession.__new__(_CuaDriverSession))
session._bridge = bridge
session._session = object()
@@ -1402,10 +1405,10 @@ class TestCuaDriverSessionReconnect:
(screenshot_out_file path) when no inline base64 is present."""
import base64 as _b64
from typing import Any, cast
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
monkeypatch.setattr(
"tools.computer_use.cua_backend.resolve_cua_driver_cmd",
"tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
lambda: "/resolved/cua-driver",
)
@@ -1657,7 +1660,7 @@ class TestCuaEnvironmentScrubbing:
to StdioServerParameters, and asserts the scrub contract.
"""
from unittest.mock import MagicMock, patch, AsyncMock
from tools.computer_use.cua_backend import _CuaDriverSession, _AsyncBridge
from tools.computer_use.cua_backend_session import _CuaDriverSession, _AsyncBridge
import asyncio
bridge = _AsyncBridge()
@@ -1681,9 +1684,9 @@ class TestCuaEnvironmentScrubbing:
return MagicMock()
with patch.dict(os.environ, test_env, clear=True), \
patch("tools.computer_use.cua_backend.resolve_cua_driver_cmd",
patch("tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
return_value="cua-driver"), \
patch("tools.computer_use.cua_backend._resolve_mcp_invocation",
patch("tools.computer_use.cua_backend_driver._resolve_mcp_invocation",
return_value=("cua-driver", ["mcp"])), \
patch("mcp.StdioServerParameters", side_effect=capture_env), \
patch("mcp.client.stdio.stdio_client") as mock_stdio, \
@@ -1749,12 +1752,12 @@ class TestCuaCliFallbackResolution:
``~/.local/bin``. Falling back to the bare ``cua-driver`` command
would reintroduce the original bug at runtime.
"""
from tools.computer_use.cua_backend import _AsyncBridge, _CuaDriverSession
from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession
proc = MagicMock(stdout="{}", stderr="", returncode=0)
session = _CuaDriverSession(_AsyncBridge())
with patch(
"tools.computer_use.cua_backend.resolve_cua_driver_cmd",
"tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
return_value="/Users/example/.local/bin/cua-driver",
), patch("subprocess.run", return_value=proc) as run:
session._call_tool_via_cli("click", {"x": 1, "y": 2}, timeout=0.1)
@@ -1925,7 +1928,7 @@ class TestZIndexSorting:
"""Wayland may return z_index: null. _ingest_windows must coerce
it to 0 (backmost) so it doesn't crash the sort or get selected
over real foreground windows."""
from tools.computer_use.cua_backend import _ingest_windows
from tools.computer_use.cua_backend_parse import _ingest_windows
raw = [
{"app_name": "Desktop", "pid": 300, "window_id": 3,
@@ -1950,7 +1953,7 @@ class TestImageMimeTypePropagation:
def test_extract_tool_result_captures_mime_alongside_image(self):
from unittest.mock import MagicMock
from tools.computer_use.cua_backend import _extract_tool_result
from tools.computer_use.cua_backend_parse import _extract_tool_result
image_part = MagicMock()
image_part.type = "image"
@@ -2019,7 +2022,7 @@ class TestMcpInvocationResolution:
def test_manifest_with_invocation_block_drives_subcommand(self):
from unittest.mock import patch
from tools.computer_use.cua_backend import _resolve_mcp_invocation
from tools.computer_use.cua_backend_driver import _resolve_mcp_invocation
manifest = (
'{"schema_version":"1",'
@@ -2034,7 +2037,7 @@ class TestMcpInvocationResolution:
"""If the manifest knows the args but not the command, keep our
resolved driver path (so HERMES_CUA_DRIVER_CMD still wins)."""
from unittest.mock import patch
from tools.computer_use.cua_backend import _resolve_mcp_invocation
from tools.computer_use.cua_backend_driver import _resolve_mcp_invocation
manifest = '{"mcp_invocation":{"args":["mcp"]}}'
with patch("subprocess.run", new=self._fake_run(stdout=manifest)):
@@ -2047,7 +2050,7 @@ class TestMcpInvocationResolution:
a string instead of a list, etc.), we still fall back rather than
passing junk to subprocess.Popen."""
from unittest.mock import patch
from tools.computer_use.cua_backend import _resolve_mcp_invocation
from tools.computer_use.cua_backend_driver import _resolve_mcp_invocation
manifest = (
'{"mcp_invocation":'
@@ -2068,7 +2071,7 @@ class TestStructuredElementsConsumption:
"""
def test_structured_parser_reads_frames(self):
from tools.computer_use.cua_backend import _parse_elements_from_structured
from tools.computer_use.cua_backend_parse import _parse_elements_from_structured
raw = [
{"element_index": 1, "role": "AXButton", "label": "OK",
@@ -2144,7 +2147,7 @@ class TestCapabilityDiscovery:
"""
def test_supports_capability_global_match_any_tool(self):
from tools.computer_use.cua_backend import _CuaDriverSession, _AsyncBridge
from tools.computer_use.cua_backend_session import _CuaDriverSession, _AsyncBridge
session = _CuaDriverSession(_AsyncBridge())
session._capabilities = {
@@ -2158,7 +2161,7 @@ class TestCapabilityDiscovery:
assert session.supports_capability("never.heard.of.it") is False
def test_supports_capability_scoped_to_specific_tool(self):
from tools.computer_use.cua_backend import _CuaDriverSession, _AsyncBridge
from tools.computer_use.cua_backend_session import _CuaDriverSession, _AsyncBridge
session = _CuaDriverSession(_AsyncBridge())
session._capabilities = {
@@ -2403,7 +2406,7 @@ class TestStartupTimeoutPhaseDetail:
import threading
from typing import Any, cast
from unittest.mock import MagicMock, patch as _patch
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
session = cast(Any, _CuaDriverSession.__new__(_CuaDriverSession))
session._lock = threading.Lock()
tests/tools/test_computer_use_cua_0_10_permissions.py
+11 -8
View File
@@ -8,6 +8,8 @@ from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from tools.computer_use import cua_backend_driver
@pytest.fixture(autouse=True)
def _reset_computer_use_state():
@@ -43,10 +45,11 @@ def test_gateway_session_key_yolo_maps_to_unrestricted_mode():
not the DB session_id the tool path passes. Mode resolution must consult
both namespaces or /yolo is silently dead on messaging platforms."""
from tools import approval
from tools import approval_context
from tools.computer_use import tool as computer_use
gateway_key = "agent:main:telegram:private:12345"
token = approval.set_current_session_key(gateway_key)
token = approval_context.set_current_session_key(gateway_key)
try:
approval.enable_session_yolo(gateway_key)
# Tool dispatch passes the (different) DB session id.
@@ -56,9 +59,9 @@ def test_gateway_session_key_yolo_maps_to_unrestricted_mode():
finally:
approval.disable_session_yolo(gateway_key)
try:
approval.reset_current_session_key(token)
approval_context.reset_current_session_key(token)
except Exception:
approval.set_current_session_key("")
approval_context.set_current_session_key("")
def test_mode_change_replaces_only_that_sessions_backend():
@@ -154,7 +157,7 @@ def test_release_seam_stops_backend_and_clears_session_state():
def test_yolo_toggle_immediately_releases_mode_dependent_backend():
from tools import approval
with patch("tools.computer_use.release_computer_use_session") as release:
with patch("tools.computer_use.tool.release_computer_use_session") as release:
approval.enable_session_yolo("session-a")
approval.disable_session_yolo("session-a")
@@ -176,7 +179,7 @@ def test_unrestricted_embedded_daemon_uses_private_socket_and_two_part_ack():
daemon = cua_backend._EmbeddedCuaDaemon("cua-driver", "unrestricted")
with patch.object(cua_backend.sys, "platform", "linux"), patch.object(
cua_backend,
cua_backend_driver,
"_resolve_mcp_invocation",
return_value=("/opt/cua-driver", ["mcp"]),
), patch.object(
@@ -215,7 +218,7 @@ def test_standard_backend_does_not_spawn_an_embedded_daemon():
def test_retired_browser_grant_cannot_change_standard_runtime(tmp_path, monkeypatch):
from tools.computer_use.cua_backend import _AsyncBridge, _CuaDriverSession
from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession
(tmp_path / "config.yaml").write_text(
"computer_use:\n grant_existing_profile: true\n",
@@ -231,10 +234,10 @@ def test_retired_browser_grant_cannot_change_standard_runtime(tmp_path, monkeypa
return MagicMock()
with patch(
"tools.computer_use.cua_backend.resolve_cua_driver_cmd",
"tools.computer_use.cua_backend_driver.resolve_cua_driver_cmd",
return_value="/opt/cua-driver",
), patch(
"tools.computer_use.cua_backend._resolve_mcp_invocation",
"tools.computer_use.cua_backend_driver._resolve_mcp_invocation",
return_value=("/opt/cua-driver", ["mcp"]),
), patch(
"mcp.StdioServerParameters", side_effect=capture_params
tests/tools/test_computer_use_cua_0_9.py
+1 -1
View File
@@ -119,7 +119,7 @@ def test_normalized_fixture_is_sanitized_and_records_the_selected_contract():
def test_foreground_support_is_discovered_from_tool_input_schema():
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
fixture = json.loads(FIXTURE.read_text(encoding="utf-8"))
listed = []
tests/tools/test_computer_use_cua_backend_linux.py
+5 -5
View File
@@ -73,13 +73,13 @@ LINUX_LIST_WINDOWS = [
def _normalized_windows(raw=ISSUE_58026_WINDOWS):
from tools.computer_use.cua_backend import _ingest_windows
from tools.computer_use.cua_backend_parse import _ingest_windows
return _ingest_windows(raw)
def test_parse_xprop_net_active_window_standard_output():
from tools.computer_use.cua_backend import _parse_xprop_net_active_window
from tools.computer_use.cua_backend_parse import _parse_xprop_net_active_window
raw = "_NET_ACTIVE_WINDOW(WINDOW): window id # 0x503000b\n"
assert _parse_xprop_net_active_window(raw) == 0x503000b
@@ -90,7 +90,7 @@ def test_default_capture_prefers_x11_active_window_when_z_index_tied():
"""The ``_NET_ACTIVE_WINDOW`` tie-break is a Linux/X11-only branch of
``_select_capture_target``; run it where ``sys.platform`` really is
linux instead of patching the branch selector."""
from tools.computer_use.cua_backend import _select_capture_target
from tools.computer_use.cua_backend_capture import _select_capture_target
windows = _normalized_windows()
@@ -110,7 +110,7 @@ def test_default_capture_skips_desktop_helper_when_active_window_unknown():
Linux-only: the helper-skipping pool filter is inside the
``sys.platform == "linux"`` branch."""
from tools.computer_use.cua_backend import _select_capture_target
from tools.computer_use.cua_backend_capture import _select_capture_target
windows = _normalized_windows()
@@ -137,7 +137,7 @@ def test_linux_null_is_on_screen_is_treated_as_unknown_not_offscreen():
def test_explicit_app_capture_preserves_filtered_target_order():
"""When the caller filters first, target selection should not skip the match."""
from tools.computer_use.cua_backend import _select_capture_target
from tools.computer_use.cua_backend_capture import _select_capture_target
chrome = _normalized_windows(LINUX_LIST_WINDOWS)[1]
tests/tools/test_computer_use_cua_macos_identity.py
+16 -15
View File
@@ -8,6 +8,7 @@ import subprocess
import pytest
from tools.computer_use import cua_backend
from tools.computer_use import cua_backend_daemon
def _codesign_proc(
@@ -28,7 +29,7 @@ def _codesign_proc(
def _patch_codesign(monkeypatch, proc):
monkeypatch.setattr(cua_backend.shutil, "which", lambda name: "/usr/bin/codesign")
monkeypatch.setattr(cua_backend_daemon.shutil, "which", lambda name: "/usr/bin/codesign")
monkeypatch.setattr(cua_backend.subprocess, "run", lambda *args, **kwargs: proc)
@@ -45,8 +46,8 @@ def test_resolve_app_path_follows_real_symlink_and_is_idempotent(tmp_path):
except (OSError, NotImplementedError):
pytest.skip("symlinks are unavailable on this host")
assert cua_backend._resolve_cua_driver_app_path(str(shim)) == str(app)
assert cua_backend._resolve_cua_driver_app_path(str(executable)) == str(app)
assert cua_backend_daemon._resolve_cua_driver_app_path(str(shim)) == str(app)
assert cua_backend_daemon._resolve_cua_driver_app_path(str(executable)) == str(app)
def test_resolve_app_path_follows_standard_driver_symlink(monkeypatch):
@@ -57,7 +58,7 @@ def test_resolve_app_path_follows_standard_driver_symlink(monkeypatch):
monkeypatch.setattr(cua_backend.os.path, "isfile", lambda path: True)
monkeypatch.setattr(cua_backend.os, "access", lambda path, mode: True)
assert cua_backend._resolve_cua_driver_app_path(symlink) == "/Applications/CuaDriver.app"
assert cua_backend_daemon._resolve_cua_driver_app_path(symlink) == "/Applications/CuaDriver.app"
def test_resolve_app_path_does_not_fall_back_to_an_unrelated_bundle(monkeypatch):
@@ -67,21 +68,21 @@ def test_resolve_app_path_does_not_fall_back_to_an_unrelated_bundle(monkeypatch)
lambda path: "/usr/local/bin/cua-driver",
)
assert cua_backend._resolve_cua_driver_app_path("cua-driver") is None
assert cua_backend_daemon._resolve_cua_driver_app_path("cua-driver") is None
@pytest.mark.parametrize("team_id", ["4YEC26S9KF", "YCK386LBJ7"])
def test_driver_signature_accepts_official_team_ids(monkeypatch, team_id):
_patch_codesign(monkeypatch, _codesign_proc(team_id=team_id))
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_driver_signature_still_rejects_unrecognised_team(monkeypatch):
_patch_codesign(monkeypatch, _codesign_proc(team_id="EVIL000000"))
with pytest.raises(RuntimeError, match="signed by team"):
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_driver_signature_still_requires_exact_bundle_identifier(monkeypatch):
@@ -94,7 +95,7 @@ def test_driver_signature_still_requires_exact_bundle_identifier(monkeypatch):
)
with pytest.raises(RuntimeError, match="has identifier"):
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_driver_signature_rejects_unsigned_by_default(monkeypatch):
@@ -102,7 +103,7 @@ def test_driver_signature_rejects_unsigned_by_default(monkeypatch):
monkeypatch.setattr(cua_backend, "_computer_use_cfg", lambda: {})
with pytest.raises(RuntimeError, match="signed by team"):
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_driver_signature_allows_unsigned_only_with_opt_in(monkeypatch):
@@ -113,7 +114,7 @@ def test_driver_signature_allows_unsigned_only_with_opt_in(monkeypatch):
lambda: {"allow_unsigned_driver": True},
)
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_unsigned_opt_in_still_requires_exact_bundle_identifier(monkeypatch):
@@ -131,14 +132,14 @@ def test_unsigned_opt_in_still_requires_exact_bundle_identifier(monkeypatch):
)
with pytest.raises(RuntimeError, match="has identifier"):
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_driver_signature_requires_codesign(monkeypatch):
monkeypatch.setattr(cua_backend.shutil, "which", lambda name: None)
monkeypatch.setattr(cua_backend_daemon.shutil, "which", lambda name: None)
with pytest.raises(RuntimeError, match="codesign is required"):
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_driver_signature_rejects_codesign_failure(monkeypatch):
@@ -148,7 +149,7 @@ def test_driver_signature_rejects_codesign_failure(monkeypatch):
)
with pytest.raises(RuntimeError, match="not code-signed"):
cua_backend._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
cua_backend_daemon._validate_cua_driver_app_signature("/Applications/CuaDriver.app")
def test_embedded_spawn_resolves_shim_and_accepts_current_team(monkeypatch):
@@ -158,7 +159,7 @@ def test_embedded_spawn_resolves_shim_and_accepts_current_team(monkeypatch):
monkeypatch.setattr(cua_backend.os, "access", lambda path, mode: True)
_patch_codesign(monkeypatch, _codesign_proc(team_id="YCK386LBJ7"))
command = cua_backend._embedded_daemon_spawn_command(
command = cua_backend_daemon._embedded_daemon_spawn_command(
"/Users/test/.local/bin/cua-driver",
["serve", "--embedded", "--socket", "/tmp/private.sock"],
platform="darwin",
tests/tools/test_computer_use_delivery_ladder.py
+2 -2
View File
@@ -319,7 +319,7 @@ def test_lifecycle_finally_resets_started_for_reentry():
"""After the lifecycle coro exits (MCP drop / crash), _started must be
False so _require_started() no longer passes into a dead/None session.
We drive the finally block directly via the coro's cleanup semantics."""
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
sess = _CuaDriverSession.__new__(_CuaDriverSession)
sess._session = object()
@@ -336,7 +336,7 @@ def test_lifecycle_finally_resets_started_for_reentry():
def test_call_tool_restarts_a_dead_session(monkeypatch):
"""call_tool on a session whose lifecycle died (_started False) must
call start() to rebuild it, not raise 'not started' or hang."""
from tools.computer_use.cua_backend import _CuaDriverSession
from tools.computer_use.cua_backend_session import _CuaDriverSession
sess = _CuaDriverSession.__new__(_CuaDriverSession)
sess._started = False # dead session
tests/tools/test_computer_use_null_pid_windows.py
+2 -2
View File
@@ -37,7 +37,7 @@ _PNG_B64 = (
class TestIngestWindows:
def test_skips_window_with_null_pid(self):
from tools.computer_use.cua_backend import _ingest_windows
from tools.computer_use.cua_backend_parse import _ingest_windows
raw = [
{"app_name": "Desktop", "pid": None, "window_id": 1, "z_index": 0},
@@ -52,7 +52,7 @@ class TestIngestWindows:
def test_preserves_fields_capture_relies_on(self):
from tools.computer_use.cua_backend import _ingest_windows
from tools.computer_use.cua_backend_parse import _ingest_windows
out = _ingest_windows([
{
tests/tools/test_computer_use_placeholder_ids.py
+1 -1
View File
@@ -91,7 +91,7 @@ def test_malformed_ids_are_not_treated_as_placeholders():
def test_placeholder_predicate():
from tools.computer_use.cua_backend import _is_placeholder_id
from tools.computer_use.cua_backend_parse import _is_placeholder_id
assert _is_placeholder_id(0) is True
assert _is_placeholder_id("0") is True
tools/computer_use/cua_backend.py
+8 -14
View File
@@ -4,8 +4,7 @@ background loop (``cua_backend_session``); the same tool surface works on all th
with `hermes computer-use install`. The macOS path uses private SkyLight SPIs that can break on OS updates.
Siblings: ``cua_backend_driver`` (binary/contract/update), ``cua_backend_capture`` + ``cua_backend_input``
(mixins), ``cua_backend_parse``, ``cua_backend_session`` (bridge + session + CLI fallback), ``cua_backend_daemon``
(private daemon + macOS app identity). Moved names are re-imported here so ``patch("tools.computer_use.cua_backend.X")``
keeps working; siblings look policy helpers up lazily through this module."""
(private daemon + macOS app identity). Siblings look this module's config/policy helpers up lazily."""
from __future__ import annotations
@@ -13,7 +12,6 @@ import contextlib
import importlib
import logging
import os
import shutil # noqa: F401 (tests patch cua_backend.shutil / .subprocess / .threading)
import subprocess
import sys
import threading
@@ -22,18 +20,14 @@ from typing import Any, Dict, List, Optional
from hermes_cli._subprocess_compat import windows_hide_flags
from tools.computer_use.backend import ActionResult, ComputerUseBackend
from tools.computer_use.cua_backend_capture import _CaptureMixin, _select_capture_target # noqa: F401
from tools.computer_use.cua_backend_daemon import ( # noqa: F401
_EmbeddedCuaDaemon, _embedded_daemon_spawn_command, _resolve_cua_driver_app_path, _validate_cua_driver_app_signature)
from tools.computer_use.cua_backend_driver import ( # noqa: F401
_CUA_DRIVER_ARGS, _CUA_DRIVER_CMD_ENV, _cua_driver_supports_no_overlay, _mcp_args_with_overlay_flag,
_resolve_mcp_invocation, _wsl_windows_path_to_posix, cua_driver_binary_available, cua_driver_install_hint,
cua_driver_runtime_contract_status, cua_driver_update_check, cua_driver_update_nudge, resolve_cua_driver_cmd)
from tools.computer_use.cua_backend_capture import _CaptureMixin
from tools.computer_use.cua_backend_daemon import _EmbeddedCuaDaemon
from tools.computer_use.cua_backend_driver import (
_CUA_DRIVER_CMD_ENV, cua_driver_binary_available, cua_driver_runtime_contract_status, cua_driver_update_nudge,
resolve_cua_driver_cmd)
from tools.computer_use.cua_backend_input import _InputMixin
from tools.computer_use.cua_backend_parse import ( # noqa: F401
_action_result_from, _extract_tool_result, _image_dimensions_from_bytes, _ingest_windows, _is_placeholder_id,
_parse_elements_from_structured, _parse_elements_from_tree, _parse_xprop_net_active_window, _windows_from_tool_result)
from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession # noqa: F401
from tools.computer_use.cua_backend_parse import _action_result_from
from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession
logger = logging.getLogger(__name__)
# cua-driver's anonymous PostHog telemetry gate ("0" disables; absent => ON upstream).
tools/computer_use/cua_backend_daemon.py
+9 -8
View File
@@ -1,6 +1,5 @@
"""Private embedded cua-driver daemon for non-standard permission modes, plus the macOS CuaDriver.app identity
checks its launch path depends on. Driver resolution / policy helpers are looked up lazily through
``tools.computer_use.cua_backend`` so tests that patch them there keep working."""
checks its launch path depends on. Config/policy helpers are looked up lazily through the facade."""
from __future__ import annotations
@@ -17,6 +16,8 @@ import uuid
from collections import deque
from typing import Any, Dict, List, Optional, Tuple
from tools.computer_use import cua_backend_driver as _driver
logger = logging.getLogger("tools.computer_use.cua_backend")
# The only bundle identity the private daemon may launch through, and the teams that sign official
@@ -26,7 +27,7 @@ _CUA_DRIVER_TEAM_IDS = ("4YEC26S9KF", "YCK386LBJ7")
_QUIET_ERRORS = (OSError, subprocess.SubprocessError)
def _cb():
"""Origin module, looked up lazily so ``patch("tools.computer_use.cua_backend.X")`` applies."""
"""Facade module (config/policy helpers), looked up lazily to avoid the import cycle."""
from tools.computer_use import cua_backend
return cua_backend
@@ -119,7 +120,7 @@ class _EmbeddedCuaDaemon:
"bounded mode — it will NOT bound this %s session. Migrate the manifest to version 3 to keep a "
"ceiling on approval-bypassed runs.", permission_mode)
self.permission_mode, self._driver_cmd, self._command = permission_mode, driver_cmd, driver_cmd
self._mcp_args: List[str] = list(_cb()._CUA_DRIVER_ARGS)
self._mcp_args: List[str] = list(_driver._CUA_DRIVER_ARGS)
self._process: Any = None
self._owns_runtime = self._running = False
self._stderr_tail: deque[str] = deque(maxlen=20)
@@ -153,15 +154,15 @@ class _EmbeddedCuaDaemon:
# The private daemon owns the cursor overlay, so the overlay policy must apply to this long-lived serve
# process, not only its MCP proxy. Appended BEFORE the macOS app-launch wrapping so the flag travels inside
# `open ... --args` with the rest of the serve args.
return _cb()._mcp_args_with_overlay_flag(serve_args, driver_cmd=self._command)
return _driver._mcp_args_with_overlay_flag(serve_args, driver_cmd=self._command)
def start(self) -> None:
if self._running:
return
self._driver_cmd = self._driver_cmd or _cb().resolve_cua_driver_cmd() or ""
self._driver_cmd = self._driver_cmd or _driver.resolve_cua_driver_cmd() or ""
if not self._driver_cmd:
raise RuntimeError(_cb().cua_driver_install_hint())
self._command, self._mcp_args = _cb()._resolve_mcp_invocation(self._driver_cmd)
raise RuntimeError(_driver.cua_driver_install_hint())
self._command, self._mcp_args = _driver._resolve_mcp_invocation(self._driver_cmd)
env = self._sanitized_env()
command = _embedded_daemon_spawn_command(self._command, self._serve_args(), platform=sys.platform)
self._process = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
tools/computer_use/cua_backend_driver.py
+7 -8
View File
@@ -1,6 +1,5 @@
"""cua-driver binary resolution, MCP-invocation discovery, the 0.20 runtime contract gate, and the update check.
Config-derived policy (``_cua_no_overlay``, ``_run_driver`` ...) is looked up lazily through
``tools.computer_use.cua_backend`` so tests that patch it there keep working; logger name parity likewise."""
Config-derived policy (``_cua_no_overlay``, ``_run_driver`` ...) is looked up lazily through the facade."""
from __future__ import annotations
@@ -32,7 +31,7 @@ _SEMVER_RE = re.compile(r"v?(\d+)\.(\d+)\.(\d+)(?:[-+].*)?")
_UPSTREAM_SCRIPTS = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts"
def _cb():
"""Origin module, looked up lazily so ``patch("tools.computer_use.cua_backend.X")`` applies."""
"""Facade module (config/policy helpers), looked up lazily to avoid the import cycle."""
from tools.computer_use import cua_backend
return cua_backend
@@ -100,7 +99,7 @@ def resolve_cua_driver_cmd(override: Optional[str] = None) -> Optional[str]:
def cua_driver_binary_available() -> bool:
"""True if `cua-driver` resolves via env, PATH, or known install paths."""
return _cb().resolve_cua_driver_cmd() is not None
return resolve_cua_driver_cmd() is not None
def cua_driver_install_hint() -> str:
installer = (f" irm {_UPSTREAM_SCRIPTS}/install.ps1 | iex" if sys.platform == "win32"
@@ -111,7 +110,7 @@ def cua_driver_install_hint() -> str:
def _mcp_args_with_overlay_flag(args: List[str], driver_cmd: str = _CUA_DRIVER_DEFAULT_CMD) -> List[str]:
"""Return *args* with ``--no-overlay`` appended when configured and supported."""
on = _cb()._cua_no_overlay() and _cb()._cua_driver_supports_no_overlay(driver_cmd)
on = _cb()._cua_no_overlay() and _cua_driver_supports_no_overlay(driver_cmd)
return [*args, "--no-overlay"] if on else list(args)
@functools.lru_cache(maxsize=1)
@@ -176,7 +175,7 @@ def _manifest_contract_reason(manifest: Optional[Dict[str, Any]]) -> str:
def cua_driver_runtime_contract_status(binary: Optional[str] = None) -> Dict[str, Any]:
"""Report whether a local driver can host Hermes' 0.20 integration."""
resolved = binary or _cb().resolve_cua_driver_cmd()
resolved = binary or resolve_cua_driver_cmd()
version: Optional[str] = None
reason = "cua-driver is not installed"
if resolved:
@@ -203,13 +202,13 @@ def cua_driver_update_check(*, timeout: Optional[float] = None) -> Optional[Dict
See #1734.
"""
timeout = (25.0 if sys.platform == "win32" else 8.0) if timeout is None else timeout
driver_cmd = _cb().resolve_cua_driver_cmd()
driver_cmd = resolve_cua_driver_cmd()
data = _driver_json(driver_cmd, "check-update", "--json", timeout=timeout, require_ok=False) if driver_cmd else None
return None if data is None or data.get("error") else data
def cua_driver_update_nudge() -> Optional[str]:
"""One-line "an update is available" message, or ``None`` when up to date, indeterminate, or driver too old."""
state = _cb().cua_driver_update_check()
state = cua_driver_update_check()
if not state or not state.get("update_available"):
return None
return (f"cua-driver {state.get('latest_version') or '?'} is available "
tools/computer_use/cua_backend_session.py
+10 -10
View File
@@ -1,7 +1,6 @@
"""cua-driver MCP session plumbing: the asyncio bridge thread and the lazily-started, self-healing
``_CuaDriverSession`` (MCP transport with a ``cua-driver call`` CLI fallback). Driver resolution /
policy helpers are looked up lazily through ``tools.computer_use.cua_backend`` so tests that patch
them there keep working."""
``_CuaDriverSession`` (MCP transport with a ``cua-driver call`` CLI fallback). Config/policy
helpers are looked up lazily through the facade."""
from __future__ import annotations
@@ -15,6 +14,8 @@ import os
import threading
from typing import Any, Dict, List, Optional
from hermes_cli._subprocess_compat import windows_hide_flags
from tools.computer_use import cua_backend_driver as _driver
from tools.computer_use.cua_backend_parse import _extract_tool_result, _mcp_field, _tool_envelope
logger = logging.getLogger("tools.computer_use.cua_backend")
@@ -97,13 +98,12 @@ def _cli_run_json(cmd: List[str], env: Dict[str, str], name: str, timeout: float
never start) -> fail fast, no ~3.5s backoff."""
import subprocess as _subprocess
import time as _time
from tools.computer_use import cua_backend as _cb
backoff, last_err = 0.5, ""
for attempt in range(_CLI_ATTEMPTS):
try:
proc = _subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=max(15.0, timeout), creationflags=_cb.windows_hide_flags(), env=env,
timeout=max(15.0, timeout), creationflags=windows_hide_flags(), env=env,
stdin=_subprocess.DEVNULL)
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
@@ -222,14 +222,14 @@ class _CuaDriverSession:
# reports HOW FAR it got instead of an opaque "never reached ready".
self._startup_phase = "binary-check"
try:
driver_cmd = _cb.resolve_cua_driver_cmd()
driver_cmd = _driver.resolve_cua_driver_cmd()
if not driver_cmd:
raise RuntimeError(_cb.cua_driver_install_hint())
raise RuntimeError(_driver.cua_driver_install_hint())
self._startup_phase = "manifest-discovery"
daemon = self._embedded_daemon
(command, args), child_env = (
(daemon.proxy_invocation(), daemon.child_env()) if daemon is not None
else (_cb._resolve_mcp_invocation(driver_cmd), _cb.cua_driver_child_env()))
else (_driver._resolve_mcp_invocation(driver_cmd), _cb.cua_driver_child_env()))
_t_manifest = _time.monotonic()
# Telemetry policy first (default: disabled), then strip Hermes secrets.
params = StdioServerParameters(command=command, args=args, env=_sanitize_subprocess_env(child_env))
@@ -449,9 +449,9 @@ class _CuaDriverSession:
fd, shot_file = _tempfile.mkstemp(prefix="cua_shot_", suffix=".png")
os.close(fd)
call_args["screenshot_out_file"] = shot_file
driver_command = _cb.resolve_cua_driver_cmd()
driver_command = _driver.resolve_cua_driver_cmd()
if not driver_command:
raise RuntimeError(_cb.cua_driver_install_hint())
raise RuntimeError(_driver.cua_driver_install_hint())
child_env, socket_args = _cb.cua_driver_child_env(), []
daemon = getattr(self, "_embedded_daemon", None)
if daemon is not None:
tools/computer_use/doctor.py
+1 -1
View File
@@ -329,7 +329,7 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (),
for stream in (sys.stdout, sys.stderr):
with suppress(AttributeError, OSError):
stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
from tools.computer_use.cua_backend import resolve_cua_driver_cmd
from tools.computer_use.cua_backend_driver import resolve_cua_driver_cmd
binary = resolve_cua_driver_cmd(driver_cmd)
if not binary:
print(f"cua-driver: not installed (looked for {driver_cmd or 'cua-driver (PATH and canonical install paths)'!r}).\n Run: hermes computer-use install")
tools/computer_use/permissions.py
+2 -2
View File
@@ -69,7 +69,7 @@ def computer_use_status(driver_cmd: Optional[str] = None) -> Dict[str, Any]:
"""OS-aware readiness for the desktop card; key order is an API payload contract. ``ready`` is the single signal the
UI keys off: macOS = both TCC grants, elsewhere = driver health (no TCC model); ``None`` = unknown (binary missing /
probe failed). ``can_grant`` is macOS-only."""
from tools.computer_use.cua_backend import resolve_cua_driver_cmd # same resolver as the tool itself
from tools.computer_use.cua_backend_driver import resolve_cua_driver_cmd # same resolver as the tool itself
plat, binary = sys.platform, resolve_cua_driver_cmd(driver_cmd)
out: Dict[str, Any] = {"platform": plat, "platform_supported": plat in _RUNTIME_PLATFORMS,
"installed": bool(binary), "version": None, "ready": None, "can_grant": plat == "darwin",
@@ -95,7 +95,7 @@ def request_permissions_grant(driver_cmd: Optional[str] = None) -> int:
if sys.platform != "darwin":
print("Computer Use permissions are a macOS concept; nothing to grant here.")
return 64
from tools.computer_use.cua_backend import resolve_cua_driver_cmd
from tools.computer_use.cua_backend_driver import resolve_cua_driver_cmd
binary = resolve_cua_driver_cmd(driver_cmd)
if not binary:
print("cua-driver: not installed. Run: hermes computer-use install")
tools/computer_use/tool.py
+3 -2
View File
@@ -102,7 +102,8 @@ def _cua_permission_mode(session_id: str) -> str:
from tools.computer_use.cua_backend import _cua_configured_permission_mode
configured = _cua_configured_permission_mode()
with contextlib.suppress(Exception):
from tools.approval import get_current_session_key, is_approval_bypass_active_for_session
from tools.approval import is_approval_bypass_active_for_session
from tools.approval_context import get_current_session_key
if is_approval_bypass_active_for_session(session_id) or (
bool(key := get_current_session_key(default="")) and is_approval_bypass_active_for_session(key)):
with _approval_lock:
@@ -750,7 +751,7 @@ def check_computer_use_requirements() -> bool:
"""macOS/Windows/Linux + cua-driver binary (or env override). `hermes computer-use doctor` names blocked checks."""
if sys.platform not in ("darwin", "win32", "linux"):
return False
from tools.computer_use.cua_backend import cua_driver_binary_available
from tools.computer_use.cua_backend_driver import cua_driver_binary_available
return cua_driver_binary_available()
def get_computer_use_schema() -> Dict[str, Any]: