From f780cb36d883bbe4180c023fefb49fe3337e52bb Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 26 Aug 2026 09:14:17 -0700 Subject: [PATCH] =?UTF-8?q?refactor(computer=5Fuse):=20drop=20the=20cua=5F?= =?UTF-8?q?browser=5F*=20route=20=E2=80=94=20browser=20work=20goes=20throu?= =?UTF-8?q?gh=20browser=5Fexec?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real-profile browsing routes all in-page browser work through the Browser Use CLI (browser_exec), which obsoletes the cua-driver typed-browser surface baked into computer_use. Remove it so computer_use is a pure DESKTOP-control tool (screenshots / mouse / keyboard / window management) and every call's schema drops ~24 browser-only params + 9 actions. - schema.py: 9 cua_browser_* actions and the typed-browser param block removed; 14 desktop actions + shared params kept; description drops the browser rung. - tool.py: cua_browser entries out of _SAFE/_DESTRUCTIVE_ACTIONS; the whole cua_browser dispatch block deleted; {"type","cua_browser_type"} → "type" (desktop typing untouched); _config_preauthorized (browser-prepare-only, a no-op for every desktop action) and the browser-page escalation hint removed. - browser_route.py deleted (no importers outside the package); cua_backend.py drops the import + typed_browser_* methods; backend.py drops the non-abstract defaults. - tests: browser-route/contract suites removed; browser assertions trimmed. Desktop control unchanged. 233 computer_use tests pass; the 1 remaining failure (test_gateway_session_key_yolo_maps_to_unrestricted_mode) is a pre-existing cross-test state leak — fails identically on origin/main, passes in isolation. --- tests/tools/test_computer_use.py | 53 -- ...test_computer_use_browser_authorization.py | 729 ------------------ .../test_computer_use_browser_contract_020.py | 155 ---- .../test_computer_use_cua_0_10_permissions.py | 8 +- tests/tools/test_computer_use_cua_0_9.py | 578 +------------- tools/computer_use/backend.py | 29 - tools/computer_use/browser_route.py | 644 ---------------- tools/computer_use/cua_backend.py | 54 +- tools/computer_use/schema.py | 107 +-- tools/computer_use/tool.py | 216 +----- 10 files changed, 18 insertions(+), 2555 deletions(-) delete mode 100644 tests/tools/test_computer_use_browser_authorization.py delete mode 100644 tests/tools/test_computer_use_browser_contract_020.py delete mode 100644 tools/computer_use/browser_route.py diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index d76af053e4..0ac3fed17e 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -2517,59 +2517,6 @@ class TestBoundsSpaceNote: assert _bounds_space_note(zero, 0, 0) is None -class TestEscalationEnrichment: - """Browser-class background_unavailable refusals gain a typed-page hint.""" - - def _refusal(self, **overrides): - from tools.computer_use.backend import ActionResult - - kw = dict( - ok=False, action="type_text", message="refused", - code="background_unavailable", - escalation={"recommended": "foreground", "reason": "dropped"}, - meta={"event_kind": "text_input", - "target_class": "Chrome_WidgetWin_1"}, - ) - kw.update(overrides) - return ActionResult(**kw) - - def test_browser_text_refusal_gains_page_alternative(self): - from tools.computer_use.tool import _enrich_escalation - - enriched = _enrich_escalation(self._refusal()) - # Driver's recommendation is never overridden — only augmented. - assert enriched["recommended"] == "foreground" - assert enriched["alternative"] == "page" - assert "cua_browser_type" in enriched["alternative_hint"] - - def test_non_browser_target_untouched(self): - from tools.computer_use.tool import _enrich_escalation - - res = self._refusal(meta={"event_kind": "text_input", - "target_class": "Notepad"}) - assert "alternative" not in _enrich_escalation(res) - - def test_non_foreground_recommendation_untouched(self): - from tools.computer_use.tool import _enrich_escalation - - res = self._refusal(escalation={"recommended": "px"}) - assert "alternative" not in _enrich_escalation(res) - - def test_missing_escalation_passthrough(self): - from tools.computer_use.backend import ActionResult - from tools.computer_use.tool import _enrich_escalation - - assert _enrich_escalation( - ActionResult(ok=True, action="click", message="ok")) is None - - def test_enrichment_survives_action_payload(self): - from tools.computer_use.tool import _action_payload - - payload = _action_payload(self._refusal()) - assert payload["escalation"]["alternative"] == "page" - assert payload["verdict"]["decision"] == "escalate" - - class TestElementSpillFile: """Detail dropped from the in-context capture must be recoverable on disk.""" diff --git a/tests/tools/test_computer_use_browser_authorization.py b/tests/tools/test_computer_use_browser_authorization.py deleted file mode 100644 index 216743bd52..0000000000 --- a/tests/tools/test_computer_use_browser_authorization.py +++ /dev/null @@ -1,729 +0,0 @@ -"""Authorization plumbing for the cua-driver typed browser route. - -Covers the authorization modes that let ``existing_profile`` attachment (and -bounded automation generally) work from Hermes: - -* ``bounded`` permission mode — a private embedded daemon launched with a - user-reviewed capability manifest (``--capability-manifest`` + - ``--approve-capability-manifest``), failing loudly when the manifest is - missing. -* mode resolution — config supplies standard/bounded only; explicit session - YOLO still (and exclusively) selects unrestricted. -""" - -from typing import Any, Dict - -import pytest - -from tools.computer_use import cua_backend as cb -from tools.computer_use.browser_route import CuaTypedBrowserRoute -from tools.computer_use.cua_backend import _EmbeddedCuaDaemon - - -def _driver_result(payload: Dict[str, Any]) -> Dict[str, Any]: - return {"structuredContent": dict(payload)} - - -class _PrepareDriver: - def __init__(self) -> None: - self.calls: list[tuple[str, Dict[str, Any]]] = [] - - def has_tool(self, name: str) -> bool: - return True - - def call(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]: - self.calls.append((name, dict(args))) - return _driver_result({"status": "ok"}) - - -def _route(driver: _PrepareDriver) -> CuaTypedBrowserRoute: - return CuaTypedBrowserRoute( - session_id="hermes-a", - call_tool=driver.call, - has_tool=driver.has_tool, - ) - - -# ── existing-profile authorization ownership ─────────────────────────── - - -def test_existing_profile_prepare_delegates_authorization_to_driver(): - driver = _PrepareDriver() - result = _route(driver).prepare( - pid=101, - window_id=202, - profile_mode="existing_profile", - grant_existing_profile=True, - ) - - assert result["status"] == "ok" - assert driver.calls == [ - ( - "browser_prepare", - { - "pid": 101, - "window_id": 202, - "strategy": {"kind": "existing_profile"}, - "session": "hermes-a", - }, - ) - ] - - -def test_existing_profile_prepare_refused_without_config_grant(): - driver = _PrepareDriver() - result = _route(driver).prepare( - pid=101, - window_id=202, - profile_mode="existing_profile", - ) - - assert result["status"] == "refused" - assert result["code"] == "browser_existing_profile_not_granted" - assert "computer_use.grant_existing_profile" in result["message"] - # Never reached the driver: the host refuses before the transport. - assert driver.calls == [] - - -def test_existing_profile_prepare_refused_in_unrestricted_without_grant(): - """An approval bypass must not stand in for the config grant. - - ``--yolo`` / ``-z`` give the session a private unrestricted daemon that - answers every prepare, so without this host-side floor the documented - ``grant_existing_profile: false`` default silently stopped protecting the - live profile's pages, cookies, and storage. - """ - driver = _PrepareDriver() - result = _route(driver).prepare( - pid=101, - window_id=202, - profile_mode="existing_profile", - grant_existing_profile=False, - permission_mode="unrestricted", - ) - - assert result["code"] == "browser_existing_profile_not_granted" - assert driver.calls == [] - - -def test_existing_profile_prepare_bounded_mode_exempt_from_grant(): - """bounded's reviewed capability manifest is the authorization boundary.""" - driver = _PrepareDriver() - result = _route(driver).prepare( - pid=101, - window_id=202, - profile_mode="existing_profile", - grant_existing_profile=False, - permission_mode="bounded", - ) - - assert result["status"] == "ok" - assert [name for name, _ in driver.calls] == ["browser_prepare"] - - -def test_isolated_prepare_unaffected_by_the_grant(): - """The floor is scoped to existing_profile; isolated launches still work.""" - driver = _PrepareDriver() - result = _route(driver).prepare( - pid=101, - profile_mode="isolated_new", - allow_launch=True, - grant_existing_profile=False, - ) - - assert result["status"] == "ok" - assert [name for name, _ in driver.calls] == ["browser_prepare"] - - -def test_backend_resolves_authorization_and_ignores_model_supplied_values(monkeypatch): - """pid/window_id come from the model; the grant never does.""" - captured: Dict[str, Any] = {} - - class _Route: - def prepare(self, **kwargs: Any) -> Dict[str, Any]: - captured.update(kwargs) - return {"status": "ok"} - - backend = cb.CuaDriverBackend.__new__(cb.CuaDriverBackend) - backend.permission_mode = "unrestricted" - monkeypatch.setattr(cb, "_cua_grant_existing_profile", lambda: False) - monkeypatch.setattr( - cb.CuaDriverBackend, "_browser_route", lambda self: _Route() - ) - - backend.typed_browser_prepare( - pid=101, - window_id=202, - profile_mode="existing_profile", - # A model that tries to grant itself access must be ignored. - grant_existing_profile=True, - permission_mode="bounded", - ) - - assert captured["grant_existing_profile"] is False - assert captured["permission_mode"] == "unrestricted" - - -# ── config grant stands in for the approval prompt ────────────────────── - - -def _preauth(**cfg: Any): - from tools.computer_use import tool as cu_tool - - return cu_tool._config_preauthorized - - -def test_config_grant_preauthorizes_existing_profile_prepare(monkeypatch): - """The durable opt-in is the authorization; re-prompting is redundant. - - It also made the documented opt-in unusable on non-interactive runs, - where the prompt has nobody to answer it and the call dies on approval - timeout rather than attaching. - """ - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"grant_existing_profile": True} - ) - assert _preauth()( - "cua_browser_prepare", {"profile_mode": "existing_profile"} - ) is True - - -def test_no_preauthorization_without_the_grant(monkeypatch): - monkeypatch.setattr(cb, "_computer_use_cfg", dict) - assert _preauth()( - "cua_browser_prepare", {"profile_mode": "existing_profile"} - ) is False - - -def test_preauthorization_scoped_to_existing_profile(monkeypatch): - """Isolated launches keep prompting even when the grant is on.""" - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"grant_existing_profile": True} - ) - assert _preauth()( - "cua_browser_prepare", {"profile_mode": "isolated_new"} - ) is False - assert _preauth()("click", {"profile_mode": "existing_profile"}) is False - - -def test_preauthorization_fails_closed_on_config_error(monkeypatch): - def _boom(): - raise RuntimeError("config unreadable") - - monkeypatch.setattr(cb, "_computer_use_cfg", _boom) - assert _preauth()( - "cua_browser_prepare", {"profile_mode": "existing_profile"} - ) is False - - -def test_dispatch_does_not_forward_removed_approval_token(): - from unittest.mock import Mock - - from tools.computer_use.tool import _dispatch - - backend = Mock() - backend.typed_browser_prepare.return_value = {"status": "ok"} - - _dispatch( - backend, - "cua_browser_prepare", - { - "pid": 101, - "window_id": 202, - "profile_mode": "existing_profile", - }, - ) - - kwargs = backend.typed_browser_prepare.call_args.kwargs - assert "approval_token" not in kwargs - assert kwargs["profile_mode"] == "existing_profile" - - -def test_schema_does_not_expose_approval_token(): - from tools.computer_use.schema import COMPUTER_USE_SCHEMA - - assert "approval_token" not in COMPUTER_USE_SCHEMA["parameters"]["properties"] - - -# ── bounded embedded daemon ───────────────────────────────────────────── - - -def test_macos_embedded_daemon_launches_through_cuadriver_app(monkeypatch): - validated = [] - monkeypatch.setattr(cb, "_validate_cua_driver_app_signature", lambda app: validated.append(app)) - command = cb._embedded_daemon_spawn_command( - "/tmp/cua-driver", - ["serve", "--embedded", "--socket", "/tmp/private.sock"], - platform="darwin", - app_path="/Applications/CuaDriver.app", - ) - - # Signature validation is mandatory before any launch command is built. - assert validated == ["/Applications/CuaDriver.app"] - assert command == [ - "/usr/bin/open", - "-n", - "-g", - "-a", - "/Applications/CuaDriver.app", - "--args", - "serve", - "--embedded", - "--socket", - "/tmp/private.sock", - ] - - -def _codesign_proc(returncode=0, stderr=""): - import subprocess as _sp - - return _sp.CompletedProcess(["codesign"], returncode, stdout="", stderr=stderr) - - -def _patch_codesign(monkeypatch, proc): - monkeypatch.setattr(cb.shutil, "which", lambda name: "/usr/bin/codesign") - monkeypatch.setattr(cb.subprocess, "run", lambda *a, **kw: proc) - - -def test_driver_signature_valid_official_identity(monkeypatch): - _patch_codesign( - monkeypatch, - _codesign_proc(stderr="Identifier=com.trycua.driver\nTeamIdentifier=4YEC26S9KF\n"), - ) - cb._validate_cua_driver_app_signature("/Applications/CuaDriver.app") # no raise - - -def test_driver_signature_rejects_suffixed_identifier(monkeypatch): - import pytest - - _patch_codesign( - monkeypatch, - _codesign_proc(stderr="Identifier=com.trycua.driver.evil\nTeamIdentifier=4YEC26S9KF\n"), - ) - with pytest.raises(RuntimeError, match="identifier"): - cb._validate_cua_driver_app_signature("/Applications/CuaDriver.app") - - -def test_driver_signature_rejects_wrong_team(monkeypatch): - import pytest - - _patch_codesign( - monkeypatch, - _codesign_proc(stderr="Identifier=com.trycua.driver\nTeamIdentifier=EVIL000000\n"), - ) - with pytest.raises(RuntimeError, match="team"): - cb._validate_cua_driver_app_signature("/Applications/CuaDriver.app") - - -def test_driver_signature_unsigned_rejected_by_default(monkeypatch): - import pytest - - _patch_codesign( - monkeypatch, - _codesign_proc(stderr="Identifier=com.trycua.driver\nTeamIdentifier=not set\n"), - ) - monkeypatch.setattr(cb, "_computer_use_cfg", lambda: {}) - with pytest.raises(RuntimeError, match="team"): - cb._validate_cua_driver_app_signature("/Applications/CuaDriver.app") - - -def test_driver_signature_unsigned_allowed_by_config_opt_in(monkeypatch): - _patch_codesign( - monkeypatch, - _codesign_proc(stderr="Identifier=com.trycua.driver\nTeamIdentifier=not set\n"), - ) - monkeypatch.setattr(cb, "_computer_use_cfg", lambda: {"allow_unsigned_driver": True}) - cb._validate_cua_driver_app_signature("/Applications/CuaDriver.app") # no raise - - -def test_driver_signature_rejects_unsigned_bundle(monkeypatch): - import pytest - - _patch_codesign(monkeypatch, _codesign_proc(returncode=1, stderr="code object is not signed at all")) - with pytest.raises(RuntimeError, match="not code-signed"): - cb._validate_cua_driver_app_signature("/Applications/CuaDriver.app") - - -def test_resolve_app_path_has_no_applications_fallback(tmp_path): - # A driver binary OUTSIDE any .app bundle must resolve to None — the old - # /Applications fallback could launch a DIFFERENT install than the - # resolved driver. - assert cb._resolve_cua_driver_app_path(str(tmp_path / "cua-driver")) is None - - -def test_non_macos_embedded_daemon_keeps_direct_binary_launch(): - command = cb._embedded_daemon_spawn_command( - "/tmp/cua-driver", - ["serve", "--embedded", "--socket", "/tmp/private.sock"], - platform="linux", - ) - - assert command == [ - "/tmp/cua-driver", - "serve", - "--embedded", - "--socket", - "/tmp/private.sock", - ] - - -def test_bounded_daemon_requires_a_manifest(): - with pytest.raises(ValueError, match="capability_manifest"): - _EmbeddedCuaDaemon("cua-driver", "bounded") - - -def test_bounded_daemon_requires_manifest_file_to_exist(tmp_path): - with pytest.raises(ValueError, match="not found"): - _EmbeddedCuaDaemon( - "cua-driver", "bounded", - capability_manifest=str(tmp_path / "missing.yaml"), - ) - - -def test_bounded_daemon_env_does_not_bypass_approvals(tmp_path): - manifest = tmp_path / "manifest.yaml" - manifest.write_text("version: 3\n", encoding="utf-8") - daemon = _EmbeddedCuaDaemon( - "cua-driver", "bounded", capability_manifest=str(manifest) - ) - - env = daemon.child_env() - assert env["CUA_DRIVER_PERMISSION_MODE"] == "bounded" - assert "CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS" not in env - - -def test_unrestricted_daemon_env_keeps_explicit_bypass(): - daemon = _EmbeddedCuaDaemon("cua-driver", "unrestricted") - env = daemon.child_env() - assert env["CUA_DRIVER_PERMISSION_MODE"] == "unrestricted" - assert env["CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS"] == "1" - - -def test_bounded_daemon_serves_with_approved_manifest(tmp_path, monkeypatch): - """The spawn command carries the manifest + launch-time approval flags.""" - manifest = tmp_path / "manifest.yaml" - manifest.write_text("version: 3\n", encoding="utf-8") - daemon = _EmbeddedCuaDaemon( - "cua-driver", "bounded", capability_manifest=str(manifest) - ) - - captured: Dict[str, Any] = {} - - class _FakeProc: - stderr = None - - def poll(self): - return None - - def _fake_popen(command, **kwargs): - captured["command"] = list(command) - return _FakeProc() - - def _fake_run(command, **kwargs): - class _Probe: - returncode = 0 - return _Probe() - - monkeypatch.setattr(cb.subprocess, "Popen", _fake_popen) - monkeypatch.setattr(cb.subprocess, "run", _fake_run) - monkeypatch.setattr( - cb, "_resolve_mcp_invocation", lambda cmd: (cmd, ["mcp"]) - ) - - daemon.start() - - command = captured["command"] - assert "--permission-mode" in command - assert command[command.index("--permission-mode") + 1] == "bounded" - assert "--capability-manifest" in command - assert ( - command[command.index("--capability-manifest") + 1] - == str(manifest) - ) - assert "--approve-capability-manifest" in command - assert "--dangerously-bypass-approvals" not in command - - -def test_unrestricted_daemon_serve_command_unchanged(monkeypatch): - daemon = _EmbeddedCuaDaemon("cua-driver", "unrestricted") - - captured: Dict[str, Any] = {} - - class _FakeProc: - stderr = None - - def poll(self): - return None - - monkeypatch.setattr( - cb.subprocess, "Popen", - lambda command, **kw: captured.update(command=list(command)) or _FakeProc(), - ) - - def _fake_run(command, **kwargs): - class _Probe: - returncode = 0 - return _Probe() - - monkeypatch.setattr(cb.subprocess, "run", _fake_run) - monkeypatch.setattr( - cb, "_resolve_mcp_invocation", lambda cmd: (cmd, ["mcp"]) - ) - - daemon.start() - - command = captured["command"] - assert "--dangerously-bypass-approvals" in command - assert "--capability-manifest" not in command - - -# ── standard-mode --grant existing-profile ────────────────────────────── - - -def test_grant_existing_profile_defaults_off(monkeypatch): - monkeypatch.setattr(cb, "_computer_use_cfg", dict) - assert cb._cua_grant_existing_profile() is False - - -def test_grant_existing_profile_reads_config(monkeypatch): - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"grant_existing_profile": True} - ) - assert cb._cua_grant_existing_profile() is True - - -# ── permission-mode resolution ────────────────────────────────────────── - - -def test_configured_mode_defaults_to_standard(monkeypatch): - monkeypatch.setattr(cb, "_computer_use_cfg", dict) - assert cb._cua_configured_permission_mode() == "standard" - - -def test_configured_mode_honors_bounded(monkeypatch): - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"permission_mode": "Bounded"} - ) - assert cb._cua_configured_permission_mode() == "bounded" - - -@pytest.mark.parametrize("value", ["unrestricted", "yolo", "off", 3, None]) -def test_configured_mode_never_yields_unrestricted(monkeypatch, value): - """A config line must never silently bypass approvals.""" - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"permission_mode": value} - ) - assert cb._cua_configured_permission_mode() == "standard" - - -def test_capability_manifest_reads_config(monkeypatch): - monkeypatch.setattr( - cb, "_computer_use_cfg", - lambda: {"capability_manifest": " ~/manifests/cua.yaml "}, - ) - assert cb._cua_capability_manifest() == "~/manifests/cua.yaml" - monkeypatch.setattr(cb, "_computer_use_cfg", dict) - assert cb._cua_capability_manifest() is None - - -def test_session_yolo_overrides_configured_bounded(monkeypatch): - import tools.computer_use.tool as cu_tool - - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"permission_mode": "bounded"} - ) - import tools.approval as approval - - monkeypatch.setattr( - approval, "is_approval_bypass_active_for_session", lambda sid: True - ) - assert cu_tool._cua_permission_mode("sess-1") == "unrestricted" - - -def test_no_yolo_uses_configured_bounded(monkeypatch): - import tools.computer_use.tool as cu_tool - - monkeypatch.setattr( - cb, "_computer_use_cfg", lambda: {"permission_mode": "bounded"} - ) - import tools.approval as approval - - monkeypatch.setattr( - approval, "is_approval_bypass_active_for_session", lambda sid: False - ) - monkeypatch.setattr( - approval, "get_current_session_key", lambda default="": "" - ) - assert cu_tool._cua_permission_mode("sess-1") == "bounded" - - -def test_no_yolo_no_config_stays_standard(monkeypatch): - import tools.computer_use.tool as cu_tool - - monkeypatch.setattr(cb, "_computer_use_cfg", dict) - import tools.approval as approval - - monkeypatch.setattr( - approval, "is_approval_bypass_active_for_session", lambda sid: False - ) - monkeypatch.setattr( - approval, "get_current_session_key", lambda default="": "" - ) - assert cu_tool._cua_permission_mode("sess-1") == "standard" - - -def test_backend_accepts_bounded_with_manifest(tmp_path, monkeypatch): - manifest = tmp_path / "manifest.yaml" - manifest.write_text("version: 3\n", encoding="utf-8") - monkeypatch.setattr( - cb, "_cua_capability_manifest", lambda: str(manifest) - ) - monkeypatch.setattr(cb, "resolve_cua_driver_cmd", lambda override=None: "cua-driver") - - backend = cb.CuaDriverBackend(permission_mode="bounded") - assert backend.permission_mode == "bounded" - assert backend._embedded_daemon is not None - assert backend._embedded_daemon.capability_manifest == str(manifest) - - -def test_backend_bounded_without_manifest_fails_loudly(monkeypatch): - monkeypatch.setattr(cb, "_cua_capability_manifest", lambda: None) - monkeypatch.setattr(cb, "resolve_cua_driver_cmd", lambda override=None: "cua-driver") - - with pytest.raises(ValueError, match="capability_manifest"): - cb.CuaDriverBackend(permission_mode="bounded") - - -def test_backend_rejects_unknown_mode(): - with pytest.raises(ValueError, match="unsupported"): - cb.CuaDriverBackend(permission_mode="wide-open") - - -# ── manifest is a ceiling, not a mode ─────────────────────────────────── - - -def _captured_serve_command(monkeypatch, daemon): - captured: Dict[str, Any] = {} - - class _FakeProc: - stderr = None - - def poll(self): - return None - - def _fake_popen(command, **kwargs): - captured["command"] = list(command) - return _FakeProc() - - def _fake_run(command, **kwargs): - class _Probe: - returncode = 0 - - return _Probe() - - monkeypatch.setattr(cb.subprocess, "Popen", _fake_popen) - monkeypatch.setattr(cb.subprocess, "run", _fake_run) - monkeypatch.setattr(cb, "_resolve_mcp_invocation", lambda cmd: (cmd, ["mcp"])) - daemon.start() - return captured["command"] - - -def test_unrestricted_daemon_carries_a_v3_manifest(monkeypatch, tmp_path): - """An approval bypass must not silently discard a v3 ceiling. - - cua-driver accepts a v3 manifest alongside any permission mode and it can - only narrow a profile, never widen it. Pairing it with unrestricted is - what bounds an approval-bypassed run to declared scope; dropping it meant - the most carefully configured run became the least constrained one. - """ - manifest = tmp_path / "cua-capabilities.yaml" - manifest.write_text("version: 3\nallow:\n tools:\n - list_windows\n", encoding="utf-8") - daemon = _EmbeddedCuaDaemon( - "cua-driver", "unrestricted", capability_manifest=str(manifest) - ) - - command = _captured_serve_command(monkeypatch, daemon) - - assert "--dangerously-bypass-approvals" in command - assert "--capability-manifest" in command - assert command[command.index("--capability-manifest") + 1] == str(manifest) - assert "--approve-capability-manifest" in command - assert command[command.index("--permission-mode") + 1] == "unrestricted" - - -def test_unrestricted_daemon_does_not_forward_a_legacy_manifest(monkeypatch, tmp_path): - """v1/v2 manifests must declare mode: bounded, so they cannot ride along. - - Forwarding one anyway aborts driver startup with "legacy capability - manifest mode must be bounded" — turning a working session into a hard - failure. Verified against cua-driver 0.20.0. - """ - manifest = tmp_path / "legacy.yaml" - manifest.write_text( - "version: 1\nmode: bounded\nexpires_after: 1h\nidle_timeout: 5m\n", - encoding="utf-8", - ) - daemon = _EmbeddedCuaDaemon( - "cua-driver", "unrestricted", capability_manifest=str(manifest) - ) - - command = _captured_serve_command(monkeypatch, daemon) - - assert "--dangerously-bypass-approvals" in command - assert "--capability-manifest" not in command - - -def test_bounded_forwards_a_legacy_manifest_unchanged(monkeypatch, tmp_path): - """bounded is exactly where legacy manifests belong; nothing changes.""" - manifest = tmp_path / "legacy.yaml" - manifest.write_text( - "version: 1\nmode: bounded\nexpires_after: 1h\nidle_timeout: 5m\n", - encoding="utf-8", - ) - daemon = _EmbeddedCuaDaemon( - "cua-driver", "bounded", capability_manifest=str(manifest) - ) - - command = _captured_serve_command(monkeypatch, daemon) - - assert command[command.index("--permission-mode") + 1] == "bounded" - assert command[command.index("--capability-manifest") + 1] == str(manifest) - assert "--approve-capability-manifest" in command - - -def test_unparseable_manifest_is_not_forwarded_to_unrestricted(monkeypatch, tmp_path): - """Fail safe: never turn a working bypassed run into a startup abort.""" - manifest = tmp_path / "broken.yaml" - manifest.write_text("{{{ not yaml", encoding="utf-8") - daemon = _EmbeddedCuaDaemon( - "cua-driver", "unrestricted", capability_manifest=str(manifest) - ) - - command = _captured_serve_command(monkeypatch, daemon) - - assert "--capability-manifest" not in command - - -def test_unrestricted_daemon_without_a_manifest_is_unchanged(monkeypatch): - """No manifest configured -> nothing new on the command line.""" - daemon = _EmbeddedCuaDaemon("cua-driver", "unrestricted") - - command = _captured_serve_command(monkeypatch, daemon) - - assert "--dangerously-bypass-approvals" in command - assert "--capability-manifest" not in command - - -def test_unrestricted_daemon_rejects_a_missing_manifest_path(tmp_path): - """A declared-but-absent ceiling fails loudly rather than silently opening up.""" - with pytest.raises(ValueError, match="capability manifest not found"): - _EmbeddedCuaDaemon( - "cua-driver", - "unrestricted", - capability_manifest=str(tmp_path / "does-not-exist.yaml"), - ) - - -def test_bounded_still_requires_a_manifest(): - with pytest.raises(ValueError, match="requires computer_use.capability_manifest"): - _EmbeddedCuaDaemon("cua-driver", "bounded") diff --git a/tests/tools/test_computer_use_browser_contract_020.py b/tests/tools/test_computer_use_browser_contract_020.py deleted file mode 100644 index 9562e5cedb..0000000000 --- a/tests/tools/test_computer_use_browser_contract_020.py +++ /dev/null @@ -1,155 +0,0 @@ -"""Behavior coverage for the cua-driver 0.20 public browser contract.""" - -import json -from typing import Any, Dict -from unittest.mock import Mock - -from tools.computer_use.browser_route import CuaTypedBrowserRoute -from tools.computer_use.schema import COMPUTER_USE_SCHEMA -from tools.computer_use.tool import _dispatch - - -class _Driver: - def __init__(self, responses: list[Dict[str, Any]]) -> None: - self.responses = list(responses) - self.calls: list[tuple[str, Dict[str, Any]]] = [] - - def has_tool(self, _name: str) -> bool: - return True - - def call(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]: - self.calls.append((name, dict(args))) - return self.responses.pop(0) - - -def _route(driver: _Driver) -> CuaTypedBrowserRoute: - return CuaTypedBrowserRoute( - session_id="hermes-browser-contract", - call_tool=driver.call, - has_tool=driver.has_tool, - ) - - -def test_public_schema_exposes_020_state_and_type_options(): - properties = COMPUTER_USE_SCHEMA["parameters"]["properties"] - - assert properties["include_screenshot"]["type"] == "boolean" - assert properties["replace"]["type"] == "boolean" - assert properties["browser_type_mode"]["enum"] == ["insert_text", "keystrokes"] - assert "approval_token" not in properties - - -def test_browser_state_forwards_screenshot_request_and_preserves_mcp_image(): - driver = _Driver([ - { - "structuredContent": { - "status": "ok", - "target_id": "target-a", - "binding_quality": "exact", - "mutation_allowed": True, - "tabs": [{"tab_id": "tab-a"}], - }, - "images": ["/9j/browser-shot"], - "image_mime_types": ["image/jpeg"], - } - ]) - - result = _route(driver).observe( - pid=101, - window_id=202, - include_screenshot=True, - ) - - assert driver.calls == [ - ( - "get_browser_state", - { - "pid": 101, - "window_id": 202, - "include_screenshot": True, - "session": "hermes-browser-contract", - }, - ) - ] - assert result["_mcp_images"] == [ - {"data": "/9j/browser-shot", "mime_type": "image/jpeg"} - ] - assert "screenshot_deferred" not in result - - -def test_browser_bind_reports_screenshot_deferred_only_when_no_image_returned(): - driver = _Driver([ - { - "structuredContent": { - "status": "ok", - "target_id": "target-a", - "binding_quality": "exact", - "mutation_allowed": True, - "tabs": [{"tab_id": "tab-a"}], - }, - } - ]) - - result = _route(driver).observe( - pid=101, - window_id=202, - include_screenshot=True, - ) - - assert result["screenshot_deferred"] is True - assert "_mcp_images" not in result - - -def test_browser_state_dispatch_returns_mcp_image_as_multimodal_content(): - backend = Mock() - backend.typed_browser_state.return_value = { - "status": "ok", - "url": "https://example.test/", - "_mcp_images": [{"data": "iVBORbrowser-shot", "mime_type": "image/png"}], - } - - result = _dispatch( - backend, - "cua_browser_state", - {"tab_id": "tab-a", "include_screenshot": True}, - ) - - backend.typed_browser_state.assert_called_once_with( - tab_id="tab-a", include_screenshot=True - ) - assert result["_multimodal"] is True - assert json.loads(result["content"][0]["text"])["url"] == "https://example.test/" - assert result["content"][1] == { - "type": "image_url", - "image_url": {"url": "data:image/png;base64,iVBORbrowser-shot"}, - } - assert "iVBORbrowser-shot" not in result["text_summary"] - - -def test_browser_type_replace_reaches_typed_browser_backend(): - backend = Mock() - backend.typed_browser_action.return_value = {"status": "ok"} - - result = _dispatch( - backend, - "cua_browser_type", - { - "tab_id": "tab-a", - "ref": "field-a", - "text": "replacement", - "browser_type_mode": "keystrokes", - "replace": True, - }, - ) - - assert json.loads(result)["status"] == "ok" - backend.typed_browser_action.assert_called_once_with( - "browser_type", - tab_id="tab-a", - args={ - "ref": "field-a", - "text": "replacement", - "replace": True, - "mode": "keystrokes", - }, - ) 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 9601a5e383..a6e8f05b94 100644 --- a/tests/tools/test_computer_use_cua_0_10_permissions.py +++ b/tests/tools/test_computer_use_cua_0_10_permissions.py @@ -244,25 +244,19 @@ def test_standard_existing_profile_grant_stays_in_process_off_macos(): assert socket_path is None -def test_transport_reset_invalidates_native_and_browser_capabilities(): +def test_transport_reset_invalidates_native_capabilities(): from tools.computer_use.cua_backend import CuaDriverBackend backend = CuaDriverBackend(permission_mode="standard") backend._active_pid = 10 backend._active_window_id = 20 backend._snapshot_tokens = {1: "old-token"} - backend._typed_browser.state.pid = 10 - backend._typed_browser.state.window_id = 20 - backend._typed_browser.state.target_id = "old-target" - backend._typed_browser.state.refs = {"old-ref": {"click"}} backend._handle_transport_reset() assert backend._active_pid is None assert backend._active_window_id is None assert backend._snapshot_tokens == {} - assert backend._typed_browser.state.target_id is None - assert backend._typed_browser.state.refs == {} # ── the escalation is at least audible ────────────────────────────────── diff --git a/tests/tools/test_computer_use_cua_0_9.py b/tests/tools/test_computer_use_cua_0_9.py index 48c8817743..a0bf9ca419 100644 --- a/tests/tools/test_computer_use_cua_0_9.py +++ b/tests/tools/test_computer_use_cua_0_9.py @@ -336,16 +336,16 @@ def test_concurrent_hermes_sessions_do_not_share_backend_state(): def stop(self): pass - def typed_browser_state(self, **kwargs): - return {"marker": self.marker, "pid": kwargs.get("pid")} + def list_apps(self): + return [{"marker": self.marker}] def invoke(session_id): return json.loads( computer_use.handle_computer_use( - {"action": "cua_browser_state", "pid": 101, "window_id": 202}, + {"action": "list_apps"}, session_id=session_id, ) - )["marker"] + )["apps"][0]["marker"] with patch("tools.computer_use.cua_backend.CuaDriverBackend", _Backend): with ThreadPoolExecutor(max_workers=4) as executor: @@ -388,573 +388,3 @@ def test_persistent_focus_has_a_separate_approval_scope(): assert result["error"] == "denied by user" assert result["action"] == "bring_to_front" - -# --------------------------------------------------------------------------- -# Session-scoped typed browser routing -# --------------------------------------------------------------------------- - - -class _BrowserDriver: - def __init__(self, *, mutation_allowed: bool = True) -> None: - self.calls: list[tuple[str, Dict[str, Any]]] = [] - self.mutation_allowed = mutation_allowed - self.snapshot = 0 - self.responses: Dict[str, Dict[str, Any]] = {} - - def has_tool(self, name: str) -> bool: - return name in { - "get_browser_state", - "browser_prepare", - "browser_navigate", - "browser_click", - "browser_type", - "browser_pointer", - "browser_dialog", - "browser_set_input_files", - "browser_download", - } - - def call(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]: - self.calls.append((name, dict(args))) - if name in self.responses: - return _driver_result(self.responses[name]) - if name == "get_browser_state" and "pid" in args: - return _driver_result({ - "status": "ok", - "binding_quality": "exact", - "mutation_allowed": self.mutation_allowed, - "target_id": "opaque-target", - "tabs": [{"tab_id": "opaque-tab"}], - }) - if name == "get_browser_state": - self.snapshot += 1 - return _driver_result({ - "status": "ok", - "refs": { - f"p{self.snapshot}:1": { - "actions": ["click", "type", "pointer", "scroll"] - } - }, - "continuation": f"continuation-{self.snapshot}", - }) - return _driver_result({"status": "ok", "effect": "confirmed"}) - - -def _browser_route(driver: _BrowserDriver, session_id: str = "hermes-a"): - from tools.computer_use.browser_route import CuaTypedBrowserRoute - - return CuaTypedBrowserRoute( - session_id=session_id, - call_tool=driver.call, - has_tool=driver.has_tool, - ) - - -def _bind_and_snapshot(route) -> str: - bound = route.observe(pid=101, window_id=202) - assert bound["exact_binding"] is True - snapshot = route.observe(tab_id="opaque-tab") - assert snapshot["fresh_state"] is True - return next(iter(route.state.refs)) - - -def test_exact_browser_binding_injects_hermes_session_capability(): - driver = _BrowserDriver() - route = _browser_route(driver, session_id="hermes-owned-session") - - payload = route.observe(pid=101, window_id=202) - - assert payload["exact_binding"] is True - assert payload["mutation_allowed"] is True - assert driver.calls == [ - ( - "get_browser_state", - {"pid": 101, "window_id": 202, "session": "hermes-owned-session"}, - ) - ] - - -def test_browser_mutation_requires_driver_granted_mutation_capability(): - driver = _BrowserDriver(mutation_allowed=False) - route = _browser_route(driver) - route.observe(pid=101, window_id=202) - - result = route.mutate( - "browser_navigate", - tab_id="opaque-tab", - args={"url": "about:blank"}, - ) - - assert result["code"] == "browser_mutation_unproven" - assert result["native_fallback_required"] is True - assert [name for name, _ in driver.calls] == ["get_browser_state"] - - -def test_browser_bind_requires_fresh_tab_state_before_first_mutation(): - driver = _BrowserDriver() - route = _browser_route(driver) - route.observe(pid=101, window_id=202) - - result = route.mutate( - "browser_navigate", - tab_id="opaque-tab", - args={"url": "about:blank"}, - ) - - assert result["code"] == "browser_verification_required" - assert [name for name, _ in driver.calls] == ["get_browser_state"] - - -def test_browser_mutation_enforces_current_ref_and_fresh_verification(): - driver = _BrowserDriver() - route = _browser_route(driver) - current_ref = _bind_and_snapshot(route) - - stale = route.mutate( - "browser_click", - tab_id="opaque-tab", - args={"ref": "p0:stale"}, - ) - assert stale["code"] == "browser_ref_stale" - - first = route.mutate( - "browser_click", - tab_id="opaque-tab", - args={"ref": current_ref}, - ) - assert first["next_step"] == "fresh_browser_state" - assert first["verification_required"] is True - - chained = route.mutate( - "browser_navigate", - tab_id="opaque-tab", - args={"url": "about:blank"}, - ) - assert chained["code"] == "browser_verification_required" - - fresh_ref = next(iter(route.observe(tab_id="opaque-tab")["refs"])) - second = route.mutate( - "browser_type", - tab_id="opaque-tab", - args={"ref": fresh_ref, "text": "hello"}, - ) - assert second["verification_required"] is True - - -def test_live_semantic_v2_content_refs_are_the_action_capabilities(): - from tools.computer_use.browser_route import _ref_map - - refs = _ref_map({ - "status": "ok", - "refs": [], - "content_refs": [ - { - "ref": "p7:3", - "role": "button", - "actions": ["click", "pointer"], - } - ], - }) - - assert refs == {"p7:3": {"click", "pointer"}} - - -def test_split_refs_and_content_refs_merge_without_clobbering(): - """cua-driver >= 0.17 splits the semantic_v2 payload: action-bearing refs - live in ``refs`` while ``content_refs`` re-lists every node with EMPTY - action lists. The old exclusive preference (content_refs first) dropped - all actions, making every click refuse with browser_ref_stale — caught - live on 0.19.3 against example.org (PR #79515 by weisiwu). - """ - from tools.computer_use.browser_route import _ref_map - - refs = _ref_map({ - "status": "ok", - "refs": [ - {"ref": "p1:1", "role": "link", "actions": ["click", "pointer"]}, - ], - "content_refs": [ - {"ref": "p1:0", "role": "rootwebarea", "actions": []}, - # same ref re-listed with no actions — must NOT clobber - {"ref": "p1:1", "role": "link", "actions": []}, - {"ref": "p1:2", "role": "heading", "actions": []}, - ], - }) - - assert refs["p1:1"] == {"click", "pointer"} - assert refs["p1:0"] == set() - assert refs["p1:2"] == set() - - -def test_dom_event_is_forwarded_only_when_explicitly_requested(): - driver = _BrowserDriver() - route = _browser_route(driver) - current_ref = _bind_and_snapshot(route) - - result = route.mutate( - "browser_pointer", - tab_id="opaque-tab", - args={ - "action": "right_click", - "ref": current_ref, - "input_route": "dom_event", - }, - ) - - name, sent = driver.calls[-1] - assert name == "browser_pointer" - assert sent["input_route"] == "dom_event" - assert result["input_trust"] == "dom_event" - assert result["trust_downgrade_explicit"] is True - - -def test_trust_route_is_rejected_for_tools_without_a_live_route_property(): - driver = _BrowserDriver() - route = _browser_route(driver) - current_ref = _bind_and_snapshot(route) - - result = route.mutate( - "browser_type", - tab_id="opaque-tab", - args={"ref": current_ref, "text": "hello", "input_route": "dom_event"}, - ) - - assert result["code"] == "browser_input_route_unsupported" - assert [name for name, _ in driver.calls].count("browser_type") == 0 - - -def test_scope_ref_must_come_from_this_routes_latest_snapshot(): - driver = _BrowserDriver() - route = _browser_route(driver) - _bind_and_snapshot(route) - - result = route.observe(tab_id="opaque-tab", scope_ref="other-session:1") - - assert result["code"] == "browser_ref_stale" - assert len(driver.calls) == 2 - - -def test_typed_browser_refs_do_not_cross_route_sessions(): - driver = _BrowserDriver() - first = _browser_route(driver, session_id="hermes-a") - second = _browser_route(driver, session_id="hermes-b") - first_ref = _bind_and_snapshot(first) - _bind_and_snapshot(second) - - result = second.mutate( - "browser_click", - tab_id="opaque-tab", - args={"ref": first_ref}, - ) - - assert result["code"] == "browser_ref_stale" - - -def test_trusted_browser_refusal_does_not_silently_change_route(): - driver = _BrowserDriver() - driver.responses["browser_click"] = { - "status": "refused", - "code": "browser_input_trust_unavailable", - } - route = _browser_route(driver) - current_ref = _bind_and_snapshot(route) - - result = route.mutate( - "browser_click", - tab_id="opaque-tab", - args={"ref": current_ref}, - ) - - browser_click_calls = [ - args for name, args in driver.calls if name == "browser_click" - ] - assert len(browser_click_calls) == 1 - assert browser_click_calls[0].get("input_route") is None - assert result["trust_change_requires_explicit_choice"] is True - assert result["native_fallback_available"] is True - assert route.state.refs == {} - assert route.state.verification_required is True - - -def test_typed_mutation_disarms_refs_before_transport_failure(): - driver = _BrowserDriver() - route = _browser_route(driver) - current_ref = _bind_and_snapshot(route) - - def fail_transport(name, args): - raise RuntimeError("connection lost after dispatch") - - route._call_tool = fail_transport - with pytest.raises(RuntimeError, match="connection lost"): - route.mutate( - "browser_click", - tab_id="opaque-tab", - args={"ref": current_ref}, - ) - - assert route.state.refs == {} - assert route.state.verification_required is True - - -def test_read_only_dialog_inspection_does_not_invalidate_page_state(): - driver = _BrowserDriver() - route = _browser_route(driver) - current_ref = _bind_and_snapshot(route) - - inspected = route.mutate( - "browser_dialog", - tab_id="opaque-tab", - args={"action": "inspect"}, - ) - - assert inspected["fresh_dialog_state"] is True - assert current_ref in route.state.refs - assert route.state.verification_required is False - - -def test_missing_typed_browser_tool_returns_native_fallback_refusal(): - from tools.computer_use.browser_route import CuaTypedBrowserRoute - - call = Mock() - route = CuaTypedBrowserRoute( - session_id="hermes-a", - call_tool=call, - has_tool=lambda name: False, - ) - - result = route.observe(pid=101, window_id=202) - - assert result["code"] == "typed_browser_unavailable" - assert result["native_fallback_required"] is True - call.assert_not_called() - - -def test_existing_profile_prepare_delegates_to_driver_permission_mode(): - """Past the host-side grant floor, the driver still owns the decision.""" - driver = _BrowserDriver() - driver.responses["browser_prepare"] = { - "status": "refused", - "code": "browser_consent_required", - } - route = _browser_route(driver) - - result = route.prepare( - pid=101, - window_id=202, - profile_mode="existing_profile", - allow_launch=True, - grant_existing_profile=True, - ) - - assert result["code"] == "browser_consent_required" - assert driver.calls == [ - ( - "browser_prepare", - { - "pid": 101, - "window_id": 202, - "strategy": {"kind": "existing_profile"}, - "session": "hermes-a", - }, - ) - ] - - -def test_namespaced_state_and_prepare_actions_use_typed_backend_wrappers(): - from tools.computer_use.tool import _dispatch - - backend = Mock() - backend.typed_browser_state.return_value = {"status": "ok"} - backend.typed_browser_prepare.return_value = {"status": "ok"} - - _dispatch( - backend, - "cua_browser_state", - {"pid": 101, "window_id": 202}, - ) - _dispatch( - backend, - "cua_browser_prepare", - { - "pid": 101, - "window_id": 202, - "profile_mode": "isolated_new", - "allow_launch": True, - }, - ) - - backend.typed_browser_state.assert_called_once_with(pid=101, window_id=202) - backend.typed_browser_prepare.assert_called_once_with( - pid=101, - window_id=202, - profile_mode="isolated_new", - profile_name=None, - allow_launch=True, - ) - - -def test_public_schema_exposes_only_namespaced_typed_browser_actions(): - from tools.computer_use.schema import COMPUTER_USE_SCHEMA - - action_enum = COMPUTER_USE_SCHEMA["parameters"]["properties"]["action"]["enum"] - assert "cua_browser_state" in action_enum - assert "cua_browser_click" in action_enum - assert "get_browser_state" not in action_enum - assert "browser_click" not in action_enum - assert "browser_type_mode" in COMPUTER_USE_SCHEMA["parameters"]["properties"] - - -@pytest.mark.parametrize( - ("outer_action", "driver_tool", "args"), - [ - ("cua_browser_navigate", "browser_navigate", {"url": "about:blank"}), - ("cua_browser_click", "browser_click", {"ref": "p1:1"}), - ("cua_browser_type", "browser_type", {"ref": "p1:1", "text": "hello"}), - ( - "cua_browser_pointer", - "browser_pointer", - {"action": "hover", "ref": "p1:1"}, - ), - ], -) -def test_namespaced_outer_browser_actions_map_to_exact_driver_tools( - outer_action, driver_tool, args -): - from tools.computer_use.tool import _dispatch - - backend = Mock() - backend.typed_browser_action.return_value = {"status": "ok"} - - _dispatch( - backend, - outer_action, - {"tab_id": "opaque-tab", **args}, - ) - - backend.typed_browser_action.assert_called_once_with( - driver_tool, - tab_id="opaque-tab", - args=args, - ) - - -# --------------------------------------------------------------------------- -# Existing additive result and reconnect contracts -# --------------------------------------------------------------------------- - - -def test_driver_verdict_fields_are_preserved_and_surfaced_additively(): - from tools.computer_use.backend import ActionResult - from tools.computer_use.tool import _text_response - - result = ActionResult( - ok=True, - action="click", - effect="suspected_noop", - escalation={"recommended": "foreground"}, - code="background_unavailable", - path="ax", - verified=False, - ) - payload = json.loads(_text_response(result)) - assert payload["effect"] == "suspected_noop" - assert payload["escalation"] == {"recommended": "foreground"} - assert payload["code"] == "background_unavailable" - assert payload["verified"] is False - - bare = json.loads(_text_response(ActionResult(ok=True, action="click"))) - assert bare["ok"] is True - assert bare["action"] == "click" - # Verdict routes to fresh verification; a human hint may accompany the - # decision (contract is the decision, not the exact dict shape). - assert bare["verdict"]["decision"] == "verify_fresh_state" - for k in ("effect", "escalation", "code", "verified", "path", "degraded", "delivery_mode"): - assert k not in bare - - -def test_call_tool_restarts_a_dead_session(): - from tools.computer_use.cua_backend import _CuaDriverSession - - session = _CuaDriverSession.__new__(_CuaDriverSession) - session._started = False - starts = [] - - def start(): - starts.append(True) - session._started = True - session._session = object() - - session.start = start - session._require_started = lambda: None - session._is_transient_daemon_error = lambda exc: False - session._is_closed_session_error = lambda exc: False - - class _Bridge: - def run(self, coro, timeout=None): - coro.close() - return _driver_result({}) - - async def call(name, args): - return {} - - session._bridge = _Bridge() - session._call_tool_async = call - session.call_tool("click", {"pid": 1}) - assert starts == [True] - - -def test_bind_response_directs_the_caller_to_drop_pid_and_window_id(): - """A bind looks like a successful read, but carries no page content. - - Any call passing pid/window_id lands in the binding branch, which clears - state and mints new tab_ids. A caller that keeps re-sending them re-binds - forever: the tab_id it just received is already unbound on the next call, - and every mutation stays refused. The response has to say so. - """ - driver = _BrowserDriver() - route = _browser_route(driver) - - payload = route.observe(pid=101, window_id=202) - - assert payload["snapshot_required"] is True - assert payload["next_step"] == "fresh_browser_state" - assert "WITHOUT pid or window_id" in payload["hint"] - assert "screenshot_deferred" not in payload - - -def test_bind_reports_include_screenshot_as_deferred(): - """The flag had no page content to attach to; don't drop it silently.""" - driver = _BrowserDriver() - route = _browser_route(driver) - - payload = route.observe(pid=101, window_id=202, include_screenshot=True) - - assert payload["screenshot_deferred"] is True - assert payload["snapshot_required"] is True - - -def test_snapshot_after_bind_clears_the_requirement(): - driver = _BrowserDriver() - route = _browser_route(driver) - route.observe(pid=101, window_id=202) - - snapshot = route.observe(tab_id="opaque-tab") - - assert snapshot["fresh_state"] is True - assert "snapshot_required" not in snapshot - assert "hint" not in snapshot - - -def test_verification_refusal_names_the_exact_next_call(): - driver = _BrowserDriver() - route = _browser_route(driver) - route.observe(pid=101, window_id=202) - - result = route.mutate( - "browser_navigate", - tab_id="opaque-tab", - args={"url": "about:blank"}, - ) - - assert result["code"] == "browser_verification_required" - assert "WITHOUT pid or window_id" in result["message"] diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index df5d02ffc4..aa4507e362 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -216,35 +216,6 @@ class ComputerUseBackend(ABC): `element` is the 1-based SOM index returned by a prior capture call. """ - # ── Optional typed-browser adapter ────────────────────────────── - @staticmethod - def _typed_browser_unavailable() -> Dict[str, Any]: - return { - "ok": False, - "status": "refused", - "code": "typed_browser_unavailable", - "message": "This computer-use backend has no typed browser route; use native capture/input.", - "native_fallback_required": True, - } - - def typed_browser_state(self, **kwargs: Any) -> Dict[str, Any]: - """Optional exact-bind/read hook; native-only backends fail closed.""" - return self._typed_browser_unavailable() - - def typed_browser_prepare(self, **kwargs: Any) -> Dict[str, Any]: - """Optional setup hook; native-only backends fail closed.""" - return self._typed_browser_unavailable() - - def typed_browser_action( - self, - driver_tool: str, - *, - tab_id: Optional[str] = None, - args: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - """Optional mutation hook; native-only backends fail closed.""" - return self._typed_browser_unavailable() - # ── Timing ────────────────────────────────────────────────────── def wait(self, seconds: float) -> ActionResult: """Default implementation: time.sleep.""" diff --git a/tools/computer_use/browser_route.py b/tools/computer_use/browser_route.py deleted file mode 100644 index ef45cec297..0000000000 --- a/tools/computer_use/browser_route.py +++ /dev/null @@ -1,644 +0,0 @@ -"""Session-scoped typed-browser routing for cua-driver. - -The public model surface remains the single ``computer_use`` tool. This -module owns the stateful adapter between its namespaced ``cua_browser_*`` -actions and cua-driver's raw ``get_browser_state`` / ``browser_*`` tools. - -The adapter is deliberately stricter than the transport: - -* native binding must be exact before mutation; -* the driver session id is injected by the adapter, never accepted from the - model; -* refs are usable only from the latest snapshot in this Hermes session; -* every mutation invalidates refs and requires a fresh state read; and -* changing from trusted input to ``dom_event`` is always explicit. - -Browser preparation remains a separate approved action. Existing-profile -attachment is delegated to cua-driver's daemon authorization coordinator; -ordinary Hermes tool approval never substitutes for protected consent. -""" - -from __future__ import annotations - -from dataclasses import dataclass, field -from typing import Any, Callable, Dict, Iterable, Optional, Set - - -ToolCaller = Callable[[str, Dict[str, Any]], Dict[str, Any]] -ToolProbe = Callable[[str], bool] - - -def _positive_int(value: Any) -> Optional[int]: - if isinstance(value, bool): - return None - try: - parsed = int(value) - except (TypeError, ValueError): - return None - return parsed if parsed > 0 else None - - -def _tool_payload(out: Dict[str, Any]) -> Dict[str, Any]: - """Return structured data without discarding refusals or MCP images.""" - structured = out.get("structuredContent") - data = out.get("data") - payload: Dict[str, Any] = {} - if isinstance(data, dict): - payload.update(data) - elif isinstance(data, str) and data: - payload["message"] = data - if isinstance(structured, dict): - payload.update(structured) - images = out.get("images") - mime_types = out.get("image_mime_types") - if isinstance(images, list): - preserved_images = [] - for index, image in enumerate(images): - if not isinstance(image, str) or not image: - continue - mime_type = "" - if ( - isinstance(mime_types, list) - and index < len(mime_types) - and isinstance(mime_types[index], str) - ): - mime_type = mime_types[index] - preserved_images.append({"data": image, "mime_type": mime_type}) - if preserved_images: - payload["_mcp_images"] = preserved_images - if out.get("isError") is True: - payload.setdefault("isError", True) - return payload - - -def _ref_map(payload: Dict[str, Any]) -> Dict[str, Set[str]]: - """Normalize semantic-v2 action refs to ``ref -> actions``. - - cua-driver has emitted both mapping and list representations while the - semantic snapshot contract evolved. Accept both without weakening the - capability rule: a ref with no declared action remains readable only. - - cua-driver >= 0.17 splits the semantic_v2 payload: action-bearing refs - live in the ``refs`` array while ``content_refs`` carries every node - with empty action lists. Merge both sources (plus the legacy - snapshot.refs forms) so click/pointer/type refs stay usable. - """ - normalized: Dict[str, Set[str]] = {} - - def absorb(raw: Any) -> None: - if isinstance(raw, dict): - entries: Iterable[tuple[Optional[str], Any]] = raw.items() - elif isinstance(raw, list): - entries = ((None, item) for item in raw) - else: - return - for key, value in entries: - if isinstance(value, dict): - ref = value.get("ref") or key - actions = value.get("actions") - else: - ref = key - actions = None - if not isinstance(ref, str) or not ref: - continue - action_set = { - action for action in (actions or []) if isinstance(action, str) - } - # Merge, never drop: content_refs entries carry empty action - # lists and must not clobber the same ref declared in ``refs``. - normalized[ref] = normalized.get(ref, set()) | action_set - - # 0.17 authoritative action refs; older builds emit only ``refs``; some - # transitional builds emit a mapping. Absorb every shape. - absorb(payload.get("refs")) - absorb(payload.get("content_refs")) - snapshot = payload.get("snapshot") - if isinstance(snapshot, dict): - absorb(snapshot.get("refs")) - return normalized - - -def _continuation(payload: Dict[str, Any]) -> Optional[str]: - direct = payload.get("continuation") - if isinstance(direct, str) and direct: - return direct - snapshot = payload.get("snapshot") - if isinstance(snapshot, dict): - nested = snapshot.get("continuation") - if isinstance(nested, str) and nested: - return nested - return None - - -def _tab_ids(payload: Dict[str, Any]) -> Set[str]: - result: Set[str] = set() - for tab in payload.get("tabs") or []: - if not isinstance(tab, dict): - continue - tab_id = tab.get("tab_id") or tab.get("id") - if isinstance(tab_id, str) and tab_id: - result.add(tab_id) - return result - - -def _refusal_code(payload: Dict[str, Any]) -> Optional[str]: - code = payload.get("code") - if isinstance(code, str): - return code - refusal = payload.get("refusal") - if isinstance(refusal, dict) and isinstance(refusal.get("code"), str): - return refusal["code"] - return None - - -def _refusal( - code: str, - message: str, - *, - native_fallback: bool = False, - **extra: Any, -) -> Dict[str, Any]: - payload: Dict[str, Any] = { - "ok": False, - "status": "refused", - "code": code, - "message": message, - } - if native_fallback: - payload["native_fallback_required"] = True - payload.update(extra) - return payload - - -@dataclass -class BrowserRouteState: - """Capabilities minted for one explicit cua-driver session.""" - - pid: Optional[int] = None - window_id: Optional[int] = None - target_id: Optional[str] = None - tab_ids: Set[str] = field(default_factory=set) - tab_id: Optional[str] = None - binding_quality: Optional[str] = None - mutation_allowed: bool = False - refs: Dict[str, Set[str]] = field(default_factory=dict) - continuation: Optional[str] = None - verification_required: bool = False - - def clear_refs(self) -> None: - self.refs.clear() - self.continuation = None - - def clear(self) -> None: - self.pid = None - self.window_id = None - self.target_id = None - self.tab_ids.clear() - self.tab_id = None - self.binding_quality = None - self.mutation_allowed = False - self.clear_refs() - self.verification_required = False - - -class CuaTypedBrowserRoute: - """Exact-bind typed-browser adapter for a single driver session.""" - - def __init__( - self, - *, - session_id: str, - call_tool: ToolCaller, - has_tool: ToolProbe, - ) -> None: - self._session_id = session_id - self._call_tool = call_tool - self._has_tool = has_tool - self.state = BrowserRouteState() - - def _call(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]: - payload = dict(args) - # The wrapper owns the session capability. Never let a model-provided - # id replace it or address another run's target/ref namespace. - payload["session"] = self._session_id - return _tool_payload(self._call_tool(name, payload)) - - def _require_tool(self, name: str) -> Optional[Dict[str, Any]]: - if self._has_tool(name): - return None - return _refusal( - "typed_browser_unavailable", - f"The connected cua-driver does not advertise {name}; use the native AX/PX/foreground ladder.", - native_fallback=True, - ) - - def observe( - self, - *, - pid: Any = None, - window_id: Any = None, - tab_id: Optional[str] = None, - snapshot_format: str = "semantic_v2", - query: Optional[str] = None, - scope_ref: Optional[str] = None, - continuation: Optional[str] = None, - include_screenshot: bool = False, - ) -> Dict[str, Any]: - """Bind an exact native window or snapshot a bound tab.""" - missing = self._require_tool("get_browser_state") - if missing is not None: - return missing - - binding_request = pid is not None or window_id is not None - if binding_request: - exact_pid = _positive_int(pid) - exact_window = _positive_int(window_id) - self.state.clear() - if exact_pid is None or exact_window is None: - return _refusal( - "browser_exact_target_required", - "Typed browser binding requires an exact positive pid and window_id pair.", - native_fallback=True, - ) - bind_args: Dict[str, Any] = { - "pid": exact_pid, - "window_id": exact_window, - } - if include_screenshot: - bind_args["include_screenshot"] = True - payload = self._call("get_browser_state", bind_args) - if payload.get("status") != "ok": - code = _refusal_code(payload) - payload.setdefault("ok", False) - payload["native_fallback_available"] = True - if code == "browser_requires_setup": - payload["setup_required"] = True - return payload - - target_id = payload.get("target_id") - quality = payload.get("binding_quality") - mutation_allowed = payload.get("mutation_allowed") is True - if not isinstance(target_id, str) or not target_id: - return _refusal( - "browser_binding_unproven", - "Browser bind returned no opaque target capability; use native control.", - native_fallback=True, - ) - - self.state.pid = exact_pid - self.state.window_id = exact_window - self.state.target_id = target_id - self.state.tab_ids = _tab_ids(payload) - self.state.binding_quality = quality if isinstance(quality, str) else None - self.state.mutation_allowed = mutation_allowed - # Binding mints the target/tab capabilities but is not a page - # snapshot. Require one fresh tab read before any mutation. - self.state.verification_required = True - # ...and say so in the payload. Any call carrying pid/window_id - # lands here, so a caller that keeps re-sending them re-binds - # forever: every bind clears state and mints new tab_ids, so the - # tab_id it just received is already unbound on the next call and - # every mutation stays refused. The way out is to drop - # pid/window_id, which is not otherwise discoverable from a bind - # response that looks like a successful read. - payload["snapshot_required"] = True - payload["next_step"] = "fresh_browser_state" - payload["hint"] = ( - "Binding only, no page content. Call cua_browser_state again " - "WITHOUT pid or window_id (optionally with tab_id, query, " - "snapshot_format, include_screenshot) to take the snapshot " - "this binding requires before any mutation." - ) - if include_screenshot and not payload.get("_mcp_images"): - # A bind normally carries no page content. Report deferral - # only when the driver did not attach the requested image; - # some driver versions do return a native screenshot here. - payload["screenshot_deferred"] = True - payload["exact_binding"] = quality == "exact" - if quality != "exact" or not mutation_allowed: - payload["native_fallback_required"] = True - return payload - - target_id = self.state.target_id - if not target_id or self.state.binding_quality != "exact": - return _refusal( - "browser_exact_binding_required", - "Bind the exact native pid/window_id before reading a browser tab.", - native_fallback=True, - ) - selected_tab = tab_id or self.state.tab_id - if not isinstance(selected_tab, str) or not selected_tab: - return _refusal( - "browser_tab_required", - "Choose an opaque tab_id returned by the exact bind.", - ) - if selected_tab not in self.state.tab_ids: - return _refusal( - "browser_tab_unbound", - "The requested tab_id was not minted by this session's exact bind.", - ) - if continuation is not None and continuation != self.state.continuation: - return _refusal( - "browser_continuation_stale", - "The continuation is not current for this session/tab; take a fresh snapshot.", - ) - if scope_ref is not None and scope_ref not in self.state.refs: - return _refusal( - "browser_ref_stale", - "scope_ref must come from this session's latest browser snapshot.", - ) - - args: Dict[str, Any] = { - "target_id": target_id, - "tab_id": selected_tab, - "snapshot_format": snapshot_format, - } - if query: - args["query"] = query - if scope_ref: - args["scope_ref"] = scope_ref - if continuation: - args["continuation"] = continuation - if include_screenshot: - args["include_screenshot"] = True - - continuing = continuation is not None - if not continuing: - # A new snapshot supersedes every prior ref before the transport - # call. Failure therefore cannot leave a stale ref usable. - self.state.clear_refs() - payload = self._call("get_browser_state", args) - if payload.get("status") not in (None, "ok") or payload.get("isError") is True: - self.state.clear_refs() - self.state.verification_required = True - payload.setdefault("ok", False) - return payload - - discovered = _ref_map(payload) - if continuing: - self.state.refs.update(discovered) - else: - self.state.refs = discovered - self.state.continuation = _continuation(payload) - self.state.tab_id = selected_tab - self.state.verification_required = False - payload["fresh_state"] = True - payload["refs_current"] = len(self.state.refs) - return payload - - def prepare( - self, - *, - pid: Any, - window_id: Any = None, - profile_mode: str, - profile_name: Optional[str] = None, - allow_launch: bool = False, - grant_existing_profile: bool = False, - permission_mode: str = "standard", - ) -> Dict[str, Any]: - """Run explicit setup through the driver's authoritative mode gate.""" - missing = self._require_tool("browser_prepare") - if missing is not None: - return missing - exact_pid = _positive_int(pid) - if exact_pid is None: - return _refusal( - "browser_pid_required", "browser_prepare requires a positive pid." - ) - if profile_mode == "existing_profile": - exact_window = _positive_int(window_id) - if exact_window is None: - return _refusal( - "browser_exact_target_required", - "Existing-profile attachment requires an exact positive pid and window_id pair.", - ) - # Host-side floor for the config grant. The driver owns the - # immutable standard/bounded/unrestricted decision, but an - # unrestricted daemon answers every prepare — so relying on the - # driver alone let an approval bypass (`--yolo`, `-z`) silently - # nullify `computer_use.grant_existing_profile: false` and expose - # the live profile's pages, cookies, and storage over CDP. - # Approval bypass is consent to skip *prompts*, not consent to - # read an existing browser profile, so this key is enforced here - # regardless of permission mode. bounded is exempt: its reviewed - # capability manifest is the authorization boundary. - if permission_mode != "bounded" and not grant_existing_profile: - return _refusal( - "browser_existing_profile_not_granted", - "Attaching to an existing browser profile requires the " - "one-time opt-in `computer_use.grant_existing_profile: " - "true` in config.yaml. Hermes cannot grant this at " - "runtime, and an approval bypass does not substitute for " - "it. Use profile_mode=isolated_new to browse without it.", - ) - self.state.clear() - args: Dict[str, Any] = { - "pid": exact_pid, - "window_id": exact_window, - "strategy": {"kind": "existing_profile"}, - } - return self._call("browser_prepare", args) - if profile_mode not in {"isolated_new", "isolated_named"}: - return _refusal( - "browser_profile_mode_invalid", - "Use isolated_new, isolated_named, or existing_profile.", - ) - if not allow_launch: - return _refusal( - "browser_launch_not_approved", - "Driver-owned isolated setup requires explicit allow_launch=true.", - ) - profile: Dict[str, Any] = {"mode": profile_mode} - if profile_mode == "isolated_named": - if not isinstance(profile_name, str) or not profile_name: - return _refusal( - "browser_profile_name_required", - "isolated_named requires a non-empty profile name.", - ) - profile["name"] = profile_name - args: Dict[str, Any] = { - "pid": exact_pid, - "allow_launch": True, - "profile": profile, - } - exact_window = _positive_int(window_id) - if exact_window is not None: - args["window_id"] = exact_window - # Preparation/reconnect may have side effects even if its transport - # fails. Invalidate old capabilities before crossing that boundary. - self.state.clear() - return self._call("browser_prepare", args) - - def _require_mutation( - self, - *, - tool: str, - tab_id: Optional[str], - allow_without_snapshot: bool = False, - ) -> tuple[Optional[str], Optional[Dict[str, Any]]]: - missing = self._require_tool(tool) - if missing is not None: - return None, missing - if ( - not self.state.target_id - or self.state.binding_quality != "exact" - or not self.state.mutation_allowed - ): - return None, _refusal( - "browser_mutation_unproven", - "Typed browser mutation requires status=ok, binding_quality=exact, and mutation_allowed=true; use native control otherwise.", - native_fallback=True, - ) - selected_tab = tab_id or self.state.tab_id - if not isinstance(selected_tab, str) or not selected_tab: - return None, _refusal( - "browser_tab_required", "Choose a bound tab_id first." - ) - if selected_tab not in self.state.tab_ids: - return None, _refusal( - "browser_tab_unbound", - "The requested tab_id was not minted by this session's exact bind.", - ) - if self.state.verification_required and not allow_without_snapshot: - return None, _refusal( - "browser_verification_required", - "Take a fresh cua_browser_state snapshot before another " - "browser mutation: call cua_browser_state WITHOUT pid or " - "window_id. Re-sending pid/window_id re-binds instead of " - "snapshotting, which mints new tab_ids and leaves this " - "mutation refused.", - ) - return selected_tab, None - - def _require_ref( - self, - ref: Any, - *, - actions: Set[str], - ) -> Optional[Dict[str, Any]]: - if not isinstance(ref, str) or ref not in self.state.refs: - return _refusal( - "browser_ref_stale", - "Use a current ref from the latest cua_browser_state snapshot.", - ) - declared = self.state.refs[ref] - if actions and not declared.intersection(actions): - return _refusal( - "browser_action_unavailable", - "The current ref does not declare the requested browser action.", - ) - return None - - def mutate( - self, - tool: str, - *, - tab_id: Optional[str] = None, - args: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - """Invoke one typed browser tool against current capabilities.""" - call_args = dict(args or {}) - dialog_inspect = ( - tool == "browser_dialog" and call_args.get("action") == "inspect" - ) - selected_tab, refusal = self._require_mutation( - tool=tool, - tab_id=tab_id, - allow_without_snapshot=dialog_inspect, - ) - if refusal is not None: - return refusal - assert selected_tab is not None and self.state.target_id is not None - - ref = call_args.get("ref") - supports_trust_choice = tool in {"browser_click", "browser_pointer"} - requested_route = call_args.get("input_route") - if requested_route is not None and not supports_trust_choice: - return _refusal( - "browser_input_route_unsupported", - f"{tool} does not expose a trust-route choice in the live 0.9 schema.", - ) - route = requested_route or "trusted" - if route not in {"trusted", "dom_event"}: - return _refusal( - "browser_input_route_invalid", - "Use input_route=trusted or explicitly request dom_event.", - ) - if route == "dom_event" and not ref: - return _refusal( - "browser_dom_event_ref_required", - "The dom_event trust class requires a current semantic ref.", - ) - - required_actions: Set[str] = set() - if tool == "browser_click" and ref: - required_actions = {"click", "pointer"} - elif tool == "browser_type": - required_actions = {"type", "edit", "input"} - elif tool == "browser_pointer" and ref: - pointer_action = call_args.get("action") - required_actions = ( - {"scroll", "pointer"} if pointer_action == "scroll" else {"pointer"} - ) - elif tool == "browser_set_input_files": - required_actions = {"set_input_files", "upload", "files"} - elif tool == "browser_download": - required_actions = {"download", "click"} - - if required_actions: - invalid_ref = self._require_ref(ref, actions=required_actions) - if invalid_ref is not None: - return invalid_ref - destination_ref = call_args.get("destination_ref") - if destination_ref is not None: - invalid_destination = self._require_ref( - destination_ref, actions={"pointer", "drag", "drop"} - ) - if invalid_destination is not None: - return invalid_destination - - call_args["target_id"] = self.state.target_id - call_args["tab_id"] = selected_tab - if not dialog_inspect: - # A lost/refused response does not prove the action was a no-op. - # Disarm refs before transport so callers must observe fresh state - # before any retry, trust downgrade, or different mutation. - self.state.tab_id = selected_tab - self.state.clear_refs() - self.state.verification_required = True - payload = self._call(tool, call_args) - code = _refusal_code(payload) - refused = ( - payload.get("isError") is True - or payload.get("status") not in (None, "ok") - or code is not None - ) - if supports_trust_choice: - payload["input_trust"] = route - if route == "dom_event": - payload["trust_downgrade_explicit"] = True - - if refused: - payload["native_fallback_available"] = True - if dialog_inspect and code in { - "browser_ref_stale", - "browser_binding_ambiguous", - }: - self.state.clear_refs() - self.state.verification_required = True - if code == "browser_input_trust_unavailable": - payload["trust_change_requires_explicit_choice"] = True - payload["native_fallback_available"] = True - return payload - - if dialog_inspect: - payload["fresh_dialog_state"] = True - return payload - - # Never chain mutations from remembered state. Navigation and a fresh - # snapshot both invalidate refs in the driver; applying the same rule to - # all mutations guarantees fresh-state verification before another act. - payload["verification_required"] = True - payload["next_step"] = "fresh_browser_state" - return payload diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index b34e1376f9..3e70607c67 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -61,7 +61,6 @@ from tools.computer_use.backend import ( ComputerUseBackend, UIElement, ) -from tools.computer_use.browser_route import CuaTypedBrowserRoute logger = logging.getLogger(__name__) @@ -324,8 +323,8 @@ def _cua_grant_existing_profile() -> bool: It DOES apply to unrestricted mode. An approval bypass (``--yolo``, ``-z``) is consent to skip prompts, not consent to read an existing browser profile's live pages, cookies, and storage, so the host-side - floor in ``CuaTypedBrowserRoute.prepare`` enforces this key even when the - private unrestricted daemon would answer the prepare. + grant floor enforces this key even when the private unrestricted daemon + would answer the launch. """ return bool(_computer_use_cfg().get("grant_existing_profile", False)) @@ -2791,31 +2790,11 @@ class CuaDriverBackend(ComputerUseBackend): # part of the required Cua Driver 0.20 runtime contract checked at # backend startup. self._session_id: str = f"hermes-{uuid.uuid4().hex[:12]}" - self._typed_browser = CuaTypedBrowserRoute( - session_id=self._session_id, - call_tool=self._session.call_tool, - has_tool=self._session._has_tool, - ) self._session.set_transport_reset_callback(self._handle_transport_reset) def _handle_transport_reset(self) -> None: """Invalidate every capability minted by the replaced transport.""" self._clear_active_target() - route = getattr(self, "_typed_browser", None) - if route is not None: - route.state.clear() - - def _browser_route(self) -> CuaTypedBrowserRoute: - """Return the per-backend typed route, including test-constructed instances.""" - route = getattr(self, "_typed_browser", None) - if route is None: - route = CuaTypedBrowserRoute( - session_id=self._session_id, - call_tool=self._session.call_tool, - has_tool=self._session._has_tool, - ) - self._typed_browser = route - return route # ── Lifecycle ────────────────────────────────────────────────── def start(self) -> None: @@ -3968,35 +3947,6 @@ class CuaDriverBackend(ComputerUseBackend): # session-scoped input action. return self._action("bring_to_front", args, inject_session=False) - # ── Typed browser (cua-driver 0.9 contract) ─────────────────── - def typed_browser_state(self, **kwargs: Any) -> Dict[str, Any]: - """Exact-bind a native browser window or read fresh semantic state.""" - return self._browser_route().observe(**kwargs) - - def typed_browser_prepare(self, **kwargs: Any) -> Dict[str, Any]: - """Prepare an explicitly approved driver-owned browser profile. - - The authorization inputs are resolved here, from config and this - backend's immutable mode — never from model-supplied kwargs. - """ - kwargs.pop("grant_existing_profile", None) - kwargs.pop("permission_mode", None) - return self._browser_route().prepare( - grant_existing_profile=_cua_grant_existing_profile(), - permission_mode=self.permission_mode, - **kwargs, - ) - - def typed_browser_action( - self, - driver_tool: str, - *, - tab_id: Optional[str] = None, - args: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - """Run one namespaced typed-browser mutation in this exact route.""" - return self._browser_route().mutate(driver_tool, tab_id=tab_id, args=args) - # ── Pointer + display introspection ───────────────────────────── def move_cursor(self, x: int, y: int) -> ActionResult: diff --git a/tools/computer_use/schema.py b/tools/computer_use/schema.py index 9ad8ffdd57..0fb5d2f46a 100644 --- a/tools/computer_use/schema.py +++ b/tools/computer_use/schema.py @@ -21,8 +21,7 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "background-FIRST, not background-only: the default delivery routes " "to the target window without stealing the user's cursor or focus " "(works even on hidden/minimized windows), and when a result's " - "`verdict` says to escalate you climb — pixel coordinates, the typed " - "browser route (cua_browser_* actions for page content), or " + "`verdict` says to escalate you climb — pixel coordinates, or " "delivery_mode='foreground' (briefly fronts the window; separate " "approval). Each result carries a `verdict` with the next step; " "follow it — never repeat confirmed input, and re-capture to verify " @@ -59,15 +58,6 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "list_apps", "list_windows", "focus_app", - "cua_browser_state", - "cua_browser_prepare", - "cua_browser_navigate", - "cua_browser_click", - "cua_browser_type", - "cua_browser_pointer", - "cua_browser_dialog", - "cua_browser_set_input_files", - "cua_browser_download", ], "description": ( "Which action to perform. `capture` is free (no side " @@ -239,101 +229,6 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "approval scope. Default false." ), }, - # ── cua-driver typed browser route ───────────────────── - "tab_id": { - "type": "string", - "description": "Opaque tab capability returned by cua_browser_state.", - }, - "ref": { - "type": "string", - "description": "Current semantic ref from the latest cua_browser_state snapshot.", - }, - "destination_ref": { - "type": "string", - "description": "Current destination ref for a typed pointer action.", - }, - "url": {"type": "string", "description": "URL for cua_browser_navigate."}, - "input_route": { - "type": "string", - "enum": ["trusted", "dom_event"], - "description": ( - "Typed-browser trust class. Defaults to trusted. dom_event " - "is an explicit downgrade and is never selected silently." - ), - }, - "snapshot_format": { - "type": "string", - "enum": ["semantic_v2", "dom_refs_v1"], - "description": "Typed-browser snapshot format; semantic_v2 is the default.", - }, - "include_screenshot": { - "type": "boolean", - "description": ( - "For cua_browser_state, include the current browser screenshot " - "as image content in the tool result. Defaults to false. " - "Applies to snapshot calls only: passing pid/window_id makes " - "the call a binding, which carries no page content and " - "reports screenshot_deferred instead." - ), - }, - "query": {"type": "string", "description": "Optional browser-state query."}, - "scope_ref": {"type": "string", "description": "Optional current ref to scope a snapshot."}, - "continuation": {"type": "string", "description": "Continuation minted by the current snapshot."}, - "profile_mode": { - "type": "string", - "enum": ["isolated_new", "isolated_named", "existing_profile"], - "description": ( - "Browser preparation mode. isolated_new/isolated_named use " - "a driver-owned profile; existing_profile reuses the user's " - "real profile and is consent-gated — if refused, the refusal " - "names the exact config key to enable (you cannot grant it)." - ), - }, - "profile_name": {"type": "string", "description": "Name for isolated_named setup."}, - "allow_launch": { - "type": "boolean", - "description": "Explicitly allow launch of a driver-owned isolated browser.", - }, - "browser_pointer_action": { - "type": "string", - "enum": ["hover", "right_click", "double_click", "scroll", "drag"], - "description": "Operation for cua_browser_pointer.", - }, - "browser_dialog_action": { - "type": "string", - "enum": ["inspect", "accept", "dismiss"], - "description": "Page JavaScript dialog action; native prompts stay on the native ladder.", - }, - "browser_type_mode": { - "type": "string", - "enum": ["insert_text", "keystrokes"], - "description": "Delivery form for cua_browser_type; defaults to insert_text.", - }, - "replace": { - "type": "boolean", - "description": ( - "For cua_browser_type, select the target's complete value " - "before typing so the supplied text replaces it. Defaults " - "to false; true with empty text clears the field." - ), - }, - "dialog_id": {"type": "string", "description": "Opaque page-dialog capability."}, - "prompt_text": {"type": "string", "description": "Optional text for a page prompt dialog."}, - "files": { - "type": "array", - "items": {"type": "string"}, - "description": "Explicit paths for cua_browser_set_input_files.", - }, - "destination_root": { - "type": "string", - "description": "Approved destination root for cua_browser_download.", - }, - "delta_x": {"type": "number", "description": "Typed pointer horizontal delta."}, - "delta_y": {"type": "number", "description": "Typed pointer vertical delta."}, - "x": {"type": "number", "description": "Typed browser viewport x coordinate."}, - "y": {"type": "number", "description": "Typed browser viewport y coordinate."}, - "to_x": {"type": "number", "description": "Typed browser drag destination x."}, - "to_y": {"type": "number", "description": "Typed browser drag destination y."}, # ── return shape ─────────────────────────────────────── "capture_after": { "type": "boolean", diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index 825404b231..0b3801b515 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -79,16 +79,13 @@ def set_approval_callback(cb) -> None: # Actions that read, not mutate. Always allowed. _SAFE_ACTIONS = frozenset({ - "capture", "wait", "list_apps", "list_windows", "cua_browser_state", + "capture", "wait", "list_apps", "list_windows", }) # Actions that mutate user-visible state. Go through approval. _DESTRUCTIVE_ACTIONS = frozenset({ "click", "double_click", "right_click", "middle_click", "drag", "scroll", "type", "key", "set_value", "focus_app", - "cua_browser_prepare", "cua_browser_navigate", "cua_browser_click", - "cua_browser_type", "cua_browser_pointer", "cua_browser_dialog", - "cua_browser_set_input_files", "cua_browser_download", }) # Hard-blocked key combinations. Mirrored from #4562 — these are destructive @@ -277,32 +274,6 @@ def _cua_permission_mode(session_id: str) -> str: return "standard" -def _config_preauthorized(action: str, args: Dict[str, Any]) -> bool: - """True when config already carries the authorization for this action. - - ``computer_use.grant_existing_profile`` is a durable, file-backed opt-in - that the model can never set. When it is on, an extra runtime prompt for - the existing-profile prepare asks the user to re-authorize what they - already authorized — and it makes the documented opt-in unusable on any - non-interactive run, where the prompt has nobody to answer it and the - call dies on approval timeout instead of attaching. - - Scope is deliberately narrow: only the existing-profile prepare, only - when the grant is present. Isolated-profile launches still prompt, and - any resolution failure falls closed to prompting. - """ - if action != "cua_browser_prepare": - return False - if args.get("profile_mode") != "existing_profile": - return False - try: - from tools.computer_use.cua_backend import _cua_grant_existing_profile - - return _cua_grant_existing_profile() is True - except Exception: - return False - - def _get_backend(session_id: str = "") -> ComputerUseBackend: global _backend sid = str(session_id or "") @@ -557,7 +528,7 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: session_id = str(kwargs.get("session_id") or "") # Safety: validate actions before approval prompt. - if action in {"type", "cua_browser_type"}: + if action == "type": text = args.get("text", "") pat = _is_blocked_type(text) if pat: @@ -582,9 +553,8 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: "code": "bring_to_front_requires_foreground", }) - # Approval gate (destructive actions only). A durable config grant is - # already the user's authorization, so it stands in for the prompt. - if action in _DESTRUCTIVE_ACTIONS and not _config_preauthorized(action, args): + # Approval gate (destructive actions only). + if action in _DESTRUCTIVE_ACTIONS: err = _request_approval(action, args, session_id) if err is not None: return err @@ -730,94 +700,6 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> res = backend.focus_app(app, raise_window=bool(args.get("raise_window"))) return _maybe_follow_capture(backend, res, capture_after) - # cua-driver's typed browser surface is namespaced inside the existing - # computer_use tool so it cannot collide with native browser/MCP tools. - # The backend owns the opaque driver session, target, tab and ref state; - # none of those capabilities can be supplied across Hermes sessions. - if action == "cua_browser_state": - state_args: Dict[str, Any] = {} - for public, internal in ( - ("pid", "pid"), - ("window_id", "window_id"), - ("tab_id", "tab_id"), - ("snapshot_format", "snapshot_format"), - ("query", "query"), - ("scope_ref", "scope_ref"), - ("continuation", "continuation"), - ("include_screenshot", "include_screenshot"), - ): - if args.get(public) is not None: - state_args[internal] = args[public] - return _browser_state_response(backend.typed_browser_state(**state_args)) - - if action == "cua_browser_prepare": - return json.dumps(backend.typed_browser_prepare( - pid=args.get("pid"), - window_id=args.get("window_id"), - profile_mode=args.get("profile_mode", "isolated_new"), - profile_name=args.get("profile_name"), - allow_launch=bool(args.get("allow_launch")), - )) - - browser_tools = { - "cua_browser_navigate": "browser_navigate", - "cua_browser_click": "browser_click", - "cua_browser_type": "browser_type", - "cua_browser_pointer": "browser_pointer", - "cua_browser_dialog": "browser_dialog", - "cua_browser_set_input_files": "browser_set_input_files", - "cua_browser_download": "browser_download", - } - driver_tool = browser_tools.get(action) - if driver_tool is not None: - call_args: Dict[str, Any] = {} - allowed_fields = { - "browser_navigate": ("url",), - "browser_click": ("ref", "input_route", "x", "y"), - "browser_type": ("ref", "text", "replace"), - "browser_pointer": ( - "ref", "destination_ref", "input_route", "x", "y", - "to_x", "to_y", "delta_x", "delta_y", - ), - "browser_dialog": ( - "dialog_id", "prompt_text", "delivery_mode", - ), - "browser_set_input_files": ("ref", "files"), - "browser_download": ("ref", "destination_root"), - } - for field in allowed_fields[driver_tool]: - if args.get(field) is not None: - call_args[field] = args[field] - if ( - driver_tool in {"browser_click", "browser_pointer"} - and args.get("coordinate") is not None - ): - coordinate = args["coordinate"] - if isinstance(coordinate, (list, tuple)) and len(coordinate) == 2: - call_args["x"], call_args["y"] = coordinate - pointer_action = args.get("browser_pointer_action") - dialog_action = args.get("browser_dialog_action") - # Direct adapter callers may omit the public discriminator from args; - # retain this narrow compatibility path without making it usable to - # override the namespaced action selected by handle_computer_use. - nested_action = args.get("action") - if nested_action not in browser_tools: - if driver_tool == "browser_pointer" and pointer_action is None: - pointer_action = nested_action - if driver_tool == "browser_dialog" and dialog_action is None: - dialog_action = nested_action - if pointer_action is not None: - call_args["action"] = pointer_action - if dialog_action is not None: - call_args["action"] = dialog_action - if args.get("browser_type_mode") is not None: - call_args["mode"] = args["browser_type_mode"] - return json.dumps(backend.typed_browser_action( - driver_tool, - tab_id=args.get("tab_id"), - args=call_args, - )) - # delivery_mode / bring_to_front thread through every input action so the # model can escalate background → foreground per cua-driver's ladder. delivery_mode = args.get("delivery_mode") @@ -943,41 +825,6 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> # Response shaping # --------------------------------------------------------------------------- -def _browser_state_response(payload: Dict[str, Any]) -> Any: - """Return browser state as JSON, preserving requested MCP image parts.""" - state = dict(payload) - raw_images = state.pop("_mcp_images", None) - if not isinstance(raw_images, list) or not raw_images: - return json.dumps(state) - - text_summary = json.dumps(state) - content: List[Dict[str, Any]] = [ - {"type": "text", "text": text_summary}, - ] - image_count = 0 - for image in raw_images: - if not isinstance(image, dict): - continue - data = image.get("data") - if not isinstance(data, str) or not data: - continue - mime_type = image.get("mime_type") - if not isinstance(mime_type, str) or not mime_type.startswith("image/"): - mime_type = "image/jpeg" if data.startswith("/9j/") else "image/png" - content.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime_type};base64,{data}"}, - }) - image_count += 1 - if image_count == 0: - return text_summary - return { - "_multimodal": True, - "content": content, - "text_summary": text_summary, - "meta": {"action": "cua_browser_state", "images": image_count}, - } - def _classify_action_result(res: ActionResult) -> Dict[str, Any]: """Choose the next ladder step from semantic evidence, in precedence order. @@ -1002,11 +849,10 @@ def _classify_action_result(res: ActionResult) -> Dict[str, Any]: decision["recommended"] = res.escalation.get("recommended") decision["hint"] = ( "The input likely did not land. Climb one rung following " - "`recommended`: 'px' → re-issue by coordinate; 'page' → the typed " - "cua_browser_* route; 'foreground' (or a failed pixel click) → " - "re-issue with delivery_mode='foreground' (separate approval). Do " - "not predict the rung from the app being Electron/Chromium — react " - "to this signal." + "`recommended`: 'px' → re-issue by coordinate; 'foreground' (or a " + "failed pixel click) → re-issue with delivery_mode='foreground' " + "(separate approval). Do not predict the rung from the app being " + "Electron/Chromium — react to this signal." ) return decision # Transport success without semantic proof is not proof of effect. @@ -1052,51 +898,9 @@ def _text_response(res: ActionResult) -> str: return json.dumps(_action_payload(res)) -# Window classes of browsers whose page content the typed cua_browser_* route -# can drive with trusted input and ZERO focus steal. When background text -# delivery is refused for one of these surfaces, the driver's only hint is -# "foreground" (it doesn't know Hermes has a typed page route), so the model -# flashes the user's window to front for every keystroke batch. The hint below -# offers the no-flash rung first; foreground remains valid for browser chrome, -# native dialogs, and anything the typed route can't bind exactly. -_TYPED_BROWSER_WINDOW_CLASSES = { - "chrome_widgetwin_1", # Chrome, Edge, Brave, Electron-embedded Chromium - "mozillawindowclass", # Firefox -} - - def _enrich_escalation(res: ActionResult) -> Optional[Dict[str, Any]]: - """Return the driver's escalation dict, adding a typed-page alternative. - - Purely additive: never changes the driver's `recommended` rung, only - appends `alternative`/`alternative_hint` when the refused target is a - known browser window class and the refused event is page-directed input - (typing/keys into page content). The model can then try the - `cua_browser_*` route — trusted input, no window flash — before a - foreground escalation, per the documented ladder ordering. - """ - escalation = res.escalation - if not isinstance(escalation, dict): - return escalation - if escalation.get("recommended") != "foreground": - return escalation - meta = res.meta or {} - target_class = str(meta.get("target_class") or "").lower() - if target_class not in _TYPED_BROWSER_WINDOW_CLASSES: - return escalation - if meta.get("event_kind") not in {"text_input", "key_press"}: - return escalation - enriched = dict(escalation) - enriched["alternative"] = "page" - enriched["alternative_hint"] = ( - "target is a browser window: if the input goes into PAGE content " - "(not browser chrome or a native dialog), the typed cua_browser_* " - "route can deliver it without any window flash — bind with " - "cua_browser_state (exact pid/window_id), then cua_browser_type. " - "Use foreground only for chrome/native surfaces or if typed binding " - "is unavailable." - ) - return enriched + """Return the driver's escalation dict unchanged.""" + return res.escalation # Fixed cap for the AX `elements` array surfaced in a capture response. Dense