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

refactor(computer_use): drop the cua_browser_* route — browser work goes through browser_exec

Browse Source 
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.
This commit is contained in:
Teknium
2026-08-26 09:14:17 -07:00
parent 7fb52c14ba
commit f780cb36d8
10 changed files with 18 additions and 2555 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_computer_use.py
-53
View File
@@ -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."""
tests/tools/test_computer_use_browser_authorization.py
-729
View File
@@ -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")
tests/tools/test_computer_use_browser_contract_020.py
-155
View File
@@ -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",
},
)
tests/tools/test_computer_use_cua_0_10_permissions.py
+1 -7
View File
@@ -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 ──────────────────────────────────
tests/tools/test_computer_use_cua_0_9.py
+4 -574
View File
@@ -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"]
tools/computer_use/backend.py
-29
View File
@@ -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."""
tools/computer_use/browser_route.py
-644
View File
@@ -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
tools/computer_use/cua_backend.py
+2 -52
View File
@@ -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:
tools/computer_use/schema.py
+1 -106
View File
@@ -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",
tools/computer_use/tool.py
+10 -206
View File
@@ -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