diff --git a/hermes_cli/subcommands/computer_use.py b/hermes_cli/subcommands/computer_use.py index 009a1b3696..ac0d52f4b8 100644 --- a/hermes_cli/subcommands/computer_use.py +++ b/hermes_cli/subcommands/computer_use.py @@ -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() diff --git a/hermes_cli/tools_config_cua.py b/hermes_cli/tools_config_cua.py index 5509f498ae..1d3246fdf4 100644 --- a/hermes_cli/tools_config_cua.py +++ b/hermes_cli/tools_config_cua.py @@ -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 diff --git a/tests/computer_use/test_cua_cli_fallback_env.py b/tests/computer_use/test_cua_cli_fallback_env.py index b006682391..b068b151a5 100644 --- a/tests/computer_use/test_cua_cli_fallback_env.py +++ b/tests/computer_use/test_cua_cli_fallback_env.py @@ -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", ) diff --git a/tests/computer_use/test_cua_no_overlay.py b/tests/computer_use/test_cua_no_overlay.py index 12582a34a2..e063298c74 100644 --- a/tests/computer_use/test_cua_no_overlay.py +++ b/tests/computer_use/test_cua_no_overlay.py @@ -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( diff --git a/tests/computer_use/test_cua_spawn_env_sanitization.py b/tests/computer_use/test_cua_spawn_env_sanitization.py index 2dfd12376c..a975fdf7e0 100644 --- a/tests/computer_use/test_cua_spawn_env_sanitization.py +++ b/tests/computer_use/test_cua_spawn_env_sanitization.py @@ -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`` (`` 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, diff --git a/tests/computer_use/test_cua_wsl_manifest_path.py b/tests/computer_use/test_cua_wsl_manifest_path.py index c44d44babc..2b321d4640 100644 --- a/tests/computer_use/test_cua_wsl_manifest_path.py +++ b/tests/computer_use/test_cua_wsl_manifest_path.py @@ -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"] diff --git a/tests/hermes_cli/test_computer_use_cli.py b/tests/hermes_cli/test_computer_use_cli.py index dc6752d8a8..4766152077 100644 --- a/tests/hermes_cli/test_computer_use_cli.py +++ b/tests/hermes_cli/test_computer_use_cli.py @@ -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", diff --git a/tests/hermes_cli/test_install_cua_driver.py b/tests/hermes_cli/test_install_cua_driver.py index e6764abda4..3683f7e0be 100644 --- a/tests/hermes_cli/test_install_cua_driver.py +++ b/tests/hermes_cli/test_install_cua_driver.py @@ -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, \ diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index eb4df4175c..50d20c91ad 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -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() diff --git a/tests/tools/test_computer_use_cua_0_10_permissions.py b/tests/tools/test_computer_use_cua_0_10_permissions.py index dd3e6c11a0..7b968d1474 100644 --- a/tests/tools/test_computer_use_cua_0_10_permissions.py +++ b/tests/tools/test_computer_use_cua_0_10_permissions.py @@ -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 diff --git a/tests/tools/test_computer_use_cua_0_9.py b/tests/tools/test_computer_use_cua_0_9.py index a0bf9ca419..76fb3cdfbc 100644 --- a/tests/tools/test_computer_use_cua_0_9.py +++ b/tests/tools/test_computer_use_cua_0_9.py @@ -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 = [] diff --git a/tests/tools/test_computer_use_cua_backend_linux.py b/tests/tools/test_computer_use_cua_backend_linux.py index f7868b9127..b13b1bfadd 100644 --- a/tests/tools/test_computer_use_cua_backend_linux.py +++ b/tests/tools/test_computer_use_cua_backend_linux.py @@ -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] diff --git a/tests/tools/test_computer_use_cua_macos_identity.py b/tests/tools/test_computer_use_cua_macos_identity.py index f623e998c1..5aeb68201b 100644 --- a/tests/tools/test_computer_use_cua_macos_identity.py +++ b/tests/tools/test_computer_use_cua_macos_identity.py @@ -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", diff --git a/tests/tools/test_computer_use_delivery_ladder.py b/tests/tools/test_computer_use_delivery_ladder.py index d7656b883a..1d20acdff2 100644 --- a/tests/tools/test_computer_use_delivery_ladder.py +++ b/tests/tools/test_computer_use_delivery_ladder.py @@ -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 diff --git a/tests/tools/test_computer_use_null_pid_windows.py b/tests/tools/test_computer_use_null_pid_windows.py index 464fdb3a0c..666cf7cf14 100644 --- a/tests/tools/test_computer_use_null_pid_windows.py +++ b/tests/tools/test_computer_use_null_pid_windows.py @@ -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([ { diff --git a/tests/tools/test_computer_use_placeholder_ids.py b/tests/tools/test_computer_use_placeholder_ids.py index 320089c290..3719269b35 100644 --- a/tests/tools/test_computer_use_placeholder_ids.py +++ b/tests/tools/test_computer_use_placeholder_ids.py @@ -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 diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index 356d8284fe..bcf285bb04 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -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). diff --git a/tools/computer_use/cua_backend_daemon.py b/tools/computer_use/cua_backend_daemon.py index c1f0a545d7..3f9bab5dee 100644 --- a/tools/computer_use/cua_backend_daemon.py +++ b/tools/computer_use/cua_backend_daemon.py @@ -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, diff --git a/tools/computer_use/cua_backend_driver.py b/tools/computer_use/cua_backend_driver.py index 2215a0ad7b..2beb28422f 100644 --- a/tools/computer_use/cua_backend_driver.py +++ b/tools/computer_use/cua_backend_driver.py @@ -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 " diff --git a/tools/computer_use/cua_backend_session.py b/tools/computer_use/cua_backend_session.py index 43a64b8158..94e7ee52be 100644 --- a/tools/computer_use/cua_backend_session.py +++ b/tools/computer_use/cua_backend_session.py @@ -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: diff --git a/tools/computer_use/doctor.py b/tools/computer_use/doctor.py index f30e58f716..87470dd15a 100644 --- a/tools/computer_use/doctor.py +++ b/tools/computer_use/doctor.py @@ -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") diff --git a/tools/computer_use/permissions.py b/tools/computer_use/permissions.py index 60b99363d8..b1d2e5434d 100644 --- a/tools/computer_use/permissions.py +++ b/tools/computer_use/permissions.py @@ -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") diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index 518f497f14..04401721c8 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -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]: