feat(computer_use): align cua-driver 0.9 contracts

Salvaged from PR #67807 by @f-trycua onto current main.

- Foreground gate: discover delivery_mode support from the live tools/list
  inputSchema.properties (fail closed), not the never-shipped
  input.delivery_mode capability token
- bring_to_front: standalone strict-schema MCP tool (inject_session=False),
  separate approval scope, requires foreground
- Verdict precedence: confirmed > unverifiable (verify before retry) >
  suspected_noop/refusal (escalate); surfaced as explicit verdict field
- Typed cua_browser_* route inside computer_use (browser_route.py) with
  exact-binding, adapter-injected session, snapshot-scoped refs
- Per-Hermes-session backend isolation + release_computer_use_session seam
  wired into AIAgent.close()
- Recorded 0.9 tools/list fixture replaces fabricated capability tokens
This commit is contained in:
Francesco Bonacci 2026-07-29 00:06:37 -07:00 committed by Teknium
parent bcb352eeab
commit 847e401b74
19 changed files with 3302 additions and 143 deletions

View file

@ -567,16 +567,18 @@ def computer_use_guidance(platform_name: Optional[str] = None) -> str:
"Background delivery is the DEFAULT and the co-work path, but it is "
"the first rung, not the only one. Read each action's structured "
"result and climb only when the driver tells you to:\n"
"- `effect: 'confirmed'` + `verified: true` — the driver read the "
"result back. Done.\n"
"- `effect: 'confirmed'` (or `verified: true`) — done, even if an "
"advisory escalation is also present. Never repeat successful input.\n"
"- `effect: 'unverifiable'` — the input was delivered but the driver "
"can't confirm it. Re-capture and check the screenshot/tree yourself "
"before deciding it worked.\n"
"- `effect: 'suspected_noop'`, `code: 'background_unavailable'`, or an "
"`escalation.recommended` field — the action did NOT land. Follow "
"`escalation.recommended`:\n"
"can't confirm it. Get fresh state and check it before any retry; an "
"escalation recommendation does not override this rule.\n"
"- `effect: 'suspected_noop'` or a structured refusal such as "
"`code: 'background_unavailable'` — escalation is allowed. Follow "
"the recommended rung when present:\n"
" - `'px'` → re-issue addressing the target by `coordinate=[x,y]` "
"read off the screenshot instead of `element`.\n"
" - `'page'` → use the exact-bound typed browser page rung below "
"before native foreground escalation. Do not start a legacy page workflow.\n"
" - `'foreground'` (or a pixel click still didn't land) → re-issue "
"the SAME action with `delivery_mode='foreground'`. This briefly "
"raises the window; it needs its own approval and is only appropriate "
@ -586,6 +588,18 @@ def computer_use_guidance(platform_name: Optional[str] = None) -> str:
"as a prediction from the app being Electron/Chromium/GTK. Do not "
"silently retry the same rung expecting a different result, and do "
"not conclude 'cua-driver can't drive this app' — climb the ladder.\n\n"
"## Typed browser page rung\n"
"For `recommended='page'` or supported browser PAGE content, use the namespaced "
"`cua_browser_*` actions: bind with `cua_browser_state` using the exact "
"native `(pid, window_id)`, require `binding_quality='exact'` and "
"`mutation_allowed=true`, select its opaque `tab_id`, then take a "
"fresh semantic snapshot before using a current `ref`. After every "
"typed mutation, call `cua_browser_state` again before another action. "
"Input defaults to trusted; `input_route='dom_event'` is an explicit "
"downgrade, never an automatic retry. Use native capture/input for "
"browser chrome, OS permission prompts, native dialogs, and unsupported "
"targets. Browser setup is a separately approved action; attaching an "
"existing profile requires cua-driver's own interactive grant.\n\n"
"## Background mode rules\n"
"- Do NOT use `raise_window=true` on `focus_app` unless the user "
"explicitly asked you to bring a window to front. Input routing to "

View file

@ -881,7 +881,11 @@ def _cua_install_target_writable() -> bool:
return True
def install_cua_driver(upgrade: bool = False, require_confirmed_update: bool = False) -> bool:
def install_cua_driver(
upgrade: bool = False,
require_confirmed_update: bool = False,
show_installer_progress: bool = True,
) -> bool:
"""Install or refresh the cua-driver binary used by Computer Use.
The upstream installer always pulls the latest release tag, so re-running
@ -907,6 +911,10 @@ def install_cua_driver(upgrade: bool = False, require_confirmed_update: bool = F
--upgrade`` leaves it False an explicit upgrade request should still
reinstall when the check is indeterminate.
``show_installer_progress`` controls the installer's own progress line.
``hermes update`` already prints a contextual line before its update
check, so it disables this to avoid printing the refresh twice.
Returns True iff cua-driver is installed (or successfully refreshed)
when the function returns. Supported on macOS, Windows, and Linux
(Linux is alpha). Silently returns False on unsupported platforms.
@ -1054,7 +1062,10 @@ def install_cua_driver(upgrade: bool = False, require_confirmed_update: bool = F
before = ""
ok = _run_cua_driver_installer(
label="Refreshing", verbose=False, pin_version=confirmed_version
label="Refreshing",
verbose=False,
pin_version=confirmed_version,
show_progress=show_installer_progress,
)
if ok and before:
try:
@ -1331,6 +1342,7 @@ def _run_cua_driver_installer(
label: str = "Installing",
verbose: bool = True,
pin_version: Optional[str] = None,
show_progress: bool = True,
) -> bool:
"""Run the upstream cua-driver installer for this platform.
@ -1412,10 +1424,11 @@ def _run_cua_driver_installer(
install_cmd = ["/bin/bash", script_path]
use_shell = False
if verbose:
_print_info(f" {label} cua-driver (background computer-use)...")
else:
_print_info(f" {label} cua-driver...")
if show_progress:
if verbose:
_print_info(f" {label} cua-driver (background computer-use)...")
else:
_print_info(f"{label} cua-driver (Computer Use)...")
driver_cmd = _cua_driver_cmd()
installer_env = _cua_driver_env()

View file

@ -4130,7 +4130,11 @@ def _cmd_update_impl(args, gateway_mode: bool):
# driver) keeps the installed version — `hermes update`
# must stay fast; `hermes computer-use install --upgrade`
# remains the force path.
install_cua_driver(upgrade=True, require_confirmed_update=True)
install_cua_driver(
upgrade=True,
require_confirmed_update=True,
show_installer_progress=False,
)
except Exception as e:
logger.debug("cua-driver refresh failed: %s", e)

View file

@ -3891,6 +3891,7 @@ class AIAgent:
- process_registry entries for task_id (user's bg shells)
- terminal sandbox for task_id (cwd, env, shell state)
- browser daemon for task_id (open tabs, cookies)
- computer-use backend for task_id (native target and browser refs)
- memory provider (has its own lifecycle; keeps running)
We DO close:
@ -3947,6 +3948,7 @@ class AIAgent:
- Background processes tracked in ProcessRegistry
- Terminal sandbox environments
- Browser daemon sessions
- Computer-use backend sessions and target/ref state
- Active child agents (subagent delegation)
- OpenAI/httpx client connections
@ -3974,7 +3976,15 @@ class AIAgent:
except Exception:
pass
# 4. Close active child agents
# 4. Release the session-owned computer-use backend. The lazy import
# keeps sessions that never enabled computer use on the narrow path.
try:
from tools.computer_use import release_computer_use_session
release_computer_use_session(task_id)
except Exception:
pass
# 5. Close active child agents
try:
with self._active_children_lock:
children = list(self._active_children)
@ -3987,7 +3997,7 @@ class AIAgent:
except Exception:
pass
# 5. Close the OpenAI/httpx client
# 6. Close the OpenAI/httpx client
try:
client = getattr(self, "client", None)
if client is not None:
@ -3996,14 +4006,14 @@ class AIAgent:
except Exception:
pass
# 5b. Close the cached per-request wire client (reused across
# 6b. Close the cached per-request wire client (reused across
# sequential LLM calls; see _create_request_openai_client).
try:
self._close_cached_request_openai_client(reason="agent_close")
except Exception:
pass
# 6. Free conversation history. Mirrors _release_evicted_agent_soft's
# 7. Free conversation history. Mirrors _release_evicted_agent_soft's
# soft-eviction clear — close() is the hard teardown for true session
# boundaries (/new, /reset, session expiry), so the message list won't
# be reused. Drops the reference proactively rather than waiting for
@ -4014,7 +4024,7 @@ class AIAgent:
except Exception:
pass
# 7. Finalize the owned SQLite session row unless this agent is only a
# 8. Finalize the owned SQLite session row unless this agent is only a
# temporary helper that deliberately handed session ownership forward
# (manual compression helpers that rotate to a continuation session_id,
# or background-review forks that share the live parent's session_id and

View file

@ -102,8 +102,9 @@ screenshot in the same tool call. All actions that target an element
accept `modifiers=[…]` for held keys.
The input actions (`click`, `double_click`, `right_click`, `middle_click`,
`drag`, `scroll`, `type`, `key`) also accept `delivery_mode` and
`bring_to_front` — see "The verify → escalate ladder" below.
`drag`, `scroll`, `type`, `key`) also accept `delivery_mode`. The optional
`bring_to_front=True` request invokes a separately approved standalone focus
tool before foreground input; it is never an input-action property.
## The verify → escalate ladder (background-first)
@ -125,11 +126,17 @@ Walk it in order:
1. **Element, background (default).** `click(element=N)`. If `effect:"confirmed"`,
you're done.
2. **Pixel, background.** On `escalation.recommended == "px"` (or a `degraded`
capture with an empty element list), click by `coordinate=[x,y]` read off the
screenshot instead of `element`.
3. **Foreground.** On `escalation.recommended == "foreground"`,
`code:"background_unavailable"`, or a pixel click that still didn't land,
2. **Fresh verification.** `effect:"unverifiable"` means inspect a fresh
capture/state before any retry. Do this even when `escalation.recommended`
is present; it is advisory, not proof that successful input should repeat.
3. **Pixel, background.** After `effect:"suspected_noop"` or a structured
refusal recommends `"px"` (or a `degraded` capture has no elements), click
by `coordinate=[x,y]` instead of `element`.
4. **Typed page.** When `escalation.recommended == "page"` and the exact
browser-page contract below is available, use the namespaced typed route
before native foreground. This is not the legacy `page` workflow.
5. **Foreground.** After `effect:"suspected_noop"`,
`code:"background_unavailable"`, or a verified pixel no-op,
re-issue the SAME action with `delivery_mode="foreground"`. This briefly
raises the window and restores focus after; pair with `bring_to_front=True`
for a short sequence to avoid per-call flashes. It needs its own approval
@ -145,11 +152,44 @@ computer_use(action="click", element=7, delivery_mode="foreground")
```
**Escalate to foreground as a REACTION to a returned signal, never as a
prediction** from the app being Electron/Chromium/GTK. Different controls in
prediction** from the app being Electron/Chromium/GTK. A confirmed effect is
done and must not be duplicated. Different controls in
the same app behave differently. Do NOT silently retry the same rung, and do
NOT conclude "cua-driver can't drive this app" — climb the ladder. If
`delivery_mode="foreground"` returns `code:"foreground_unsupported"`, the
driver is too old; tell the user to update cua-driver.
`delivery_mode="foreground"` returns `code:"foreground_unsupported"`, the live
action schema lacks that property; choose another verified rung without
inferring support from the executable's reported version.
## Typed browser page rung
For page content in a supported GUI browser, the same `computer_use` tool
exposes namespaced `cua_browser_*` actions. They do not collide with other
browser tools. The contract is capability-based:
1. Discover the exact native browser `(pid, window_id)` with `list_windows` or
native capture, then call `cua_browser_state` with both values.
2. Continue only when it returns `status:"ok"`, `binding_quality:"exact"`, and
`mutation_allowed:true`. Select an opaque `tab_id` from that response.
3. Call `cua_browser_state` with the `tab_id` for a fresh `semantic_v2`
snapshot. Use only refs from that newest snapshot and only for their
declared actions.
4. Use the matching namespaced action (`cua_browser_click`,
`cua_browser_type`, `cua_browser_navigate`, or `cua_browser_pointer`).
Trusted input is the default. `input_route="dom_event"` is an explicit
trust downgrade; never choose it silently after a refusal.
5. Every mutation invalidates refs. Take a fresh state snapshot before another
typed action. Never chain actions from remembered refs.
`cua_browser_prepare` is a separate approved setup action. Driver-owned
`isolated_new`/`isolated_named` profiles require explicit `allow_launch=true`.
An `existing_profile` requires cua-driver's own exact, interactive grant;
ordinary Hermes approval is not a substitute and no grant token may be
invented, stored, logged, or reused.
Use the native capture/AX/pixel/foreground ladder for browser chrome, browser
permission UI, OS prompts, native dialogs, extension surfaces, unsupported
engines, and any typed route that cannot prove exact binding or mutation
permission. `cua_browser_dialog` covers page JavaScript dialogs only.
### Key shortcuts vary per platform
@ -255,14 +295,14 @@ in your conversation context.
| `cua-driver not installed` | Run `hermes computer-use install`, or `hermes tools` and enable Computer Use |
| Captures consistently return empty / "no on-screen window" | On Linux: DISPLAY may not be set (X11) or you're on pure Wayland — ask the user to run `hermes computer-use doctor`. On Windows: you may be in Session 0 (SSH session) instead of the interactive desktop — see the cua-driver `WINDOWS.md` deep-dive |
| Element index stale ("Element N not in cache") | SOM indices are only valid until the next `capture`. Re-capture before clicking. The wrapper carries opaque `element_token`s for stale-detection; you'll see an explicit error rather than a wrong click |
| Click had no effect | Read the structured verdict, don't just recapture. `effect:"unverifiable"` → re-capture and confirm yourself. `effect:"suspected_noop"` / `code:"background_unavailable"` / `escalation.recommended` → climb the ladder: try `coordinate=[x,y]` (px), then `delivery_mode="foreground"`. A modal (e.g. an Electron consent dialog) may be blocking input — foreground delivery is how you dismiss it. Don't conclude the app is undrivable |
| Click had no effect | Read the structured verdict. `effect:"unverifiable"` → fresh capture/state before retry, even with an escalation hint. `effect:"suspected_noop"` or a structured refusal → climb the recommended ladder: coordinate (px), typed page route when exact, then foreground. Browser chrome/native prompts remain native. Don't conclude the app is undrivable |
| Type text disappears into a terminal emulator | cua-driver detects terminals (Ghostty, iTerm2, Terminal.app, Windows Terminal, mintty, etc.) and routes through key-event synthesis — should "just work" on a recent cua-driver. If it doesn't, ask the user to run `hermes computer-use doctor` |
| `blocked pattern in type text` | You tried to `type` a shell command matching the dangerous-pattern block list (`curl ... \| bash`, `sudo rm -rf`, etc.). Break the command up or reconsider |
| Anything else weird | **First action: ask the user to run `hermes computer-use doctor`.** It runs the cua-driver `health_report` MCP tool and prints a structured per-check matrix. Their output tells you (and them) exactly what's wrong |
## When NOT to use `computer_use`
- **Web automation you can do via `browser_*` tools** — those use a
- **Web automation you can do via separate headless `browser_*` tools** — those use a
real headless Chromium and are more reliable than driving the user's
GUI browser. Reach for `computer_use` specifically when the task
needs the user's actual native apps (Finder/Explorer/Files, Mail/

View file

@ -0,0 +1,471 @@
"""Opt-in macOS smoke test for the installed cua-driver live MCP contract.
This script never installs, updates, or grants an existing browser profile. Start
an isolated daemon separately, then point this script at its socket:
cua-driver serve --embedded --socket /tmp/hermes-cua-0-9-live.sock \
--no-permissions-gate --no-overlay
CUA_DRIVER_LIVE_SOCKET=/tmp/hermes-cua-0-9-live.sock \
.venv/bin/python tests/computer_use/live_cua_0_9_smoke.py
The output deliberately excludes process IDs, window IDs, socket paths, and
driver payloads. Each cell is classified as pass, structured_refusal,
environment_unavailable, or unproven.
"""
import asyncio
import json
import os
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
def structured(result: Any) -> dict[str, Any]:
value = getattr(result, "structuredContent", None)
if isinstance(value, dict):
return value
dumped = result.model_dump(by_alias=True) if hasattr(result, "model_dump") else {}
for key in ("structuredContent", "structured_content"):
value = dumped.get(key)
if isinstance(value, dict):
return value
for block in getattr(result, "content", []) or []:
text = getattr(block, "text", None)
if not isinstance(text, str):
continue
try:
value = json.loads(text)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
return value
return {}
def refusal_code(payload: dict[str, Any]) -> str | None:
refusal = payload.get("refusal")
return payload.get("code") or (
refusal.get("code") if isinstance(refusal, dict) else None
)
def textedit_process_contains(pid: int, marker: str) -> bool:
"""Read the exact throwaway process through the native AX script bridge."""
script = """
on run argv
set targetPid to item 1 of argv as integer
set markerText to item 2 of argv
tell application "System Events"
tell first application process whose unix id is targetPid
set documentText to value of text area 1 of scroll area 1 of window 1
end tell
end tell
return (documentText contains markerText) as text
end run
"""
try:
result = subprocess.run(
["osascript", "-e", script, "--", str(pid), marker],
capture_output=True,
text=True,
timeout=5,
check=False,
)
except (OSError, subprocess.SubprocessError):
return False
return result.returncode == 0 and result.stdout.strip().lower() == "true"
async def run_smoke(socket_path: str) -> dict[str, dict[str, Any]]:
session_id = f"hermes-cua-live-{uuid.uuid4().hex[:8]}"
params = StdioServerParameters(
command="cua-driver",
args=["mcp", "--embedded", "--socket", socket_path],
)
report: dict[str, dict[str, Any]] = {
"foreground": {"classification": "unproven"},
"typed_browser": {"classification": "unproven"},
}
launched_pid: int | None = None
isolated_browser_pid: int | None = None
browser_pid: int | None = None
prior_foreground_pids: set[int] = set()
file_descriptor, temporary_name = tempfile.mkstemp(
prefix="hermes-cua-live-", suffix=".txt"
)
os.close(file_descriptor)
smoke_path = Path(temporary_name)
try:
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as client:
await client.initialize()
await client.call_tool("start_session", {"session": session_id})
try:
before_windows = structured(
await client.call_tool(
"list_windows",
{"on_screen_only": True, "session": session_id},
)
)
prior_foreground_pids = {
pid
for row in before_windows.get("windows") or []
if "textedit" in str(row.get("app_name") or "").lower()
and isinstance((pid := row.get("pid")), int)
}
launched = structured(
await client.call_tool(
"launch_app",
{
"name": "TextEdit",
"urls": [smoke_path.as_uri()],
"creates_new_application_instance": True,
"session": session_id,
},
)
)
launched_pid = launched.get("pid")
windows = launched.get("windows") or []
if isinstance(launched_pid, int) and not windows:
await client.call_tool(
"wait", {"seconds": 1, "session": session_id}
)
refreshed = structured(
await client.call_tool(
"list_windows",
{"on_screen_only": True, "session": session_id},
)
)
windows = [
row
for row in refreshed.get("windows") or []
if row.get("pid") == launched_pid
]
window_id = windows[0].get("window_id") if windows else None
if (
not isinstance(launched_pid, int)
or launched_pid in prior_foreground_pids
or not isinstance(window_id, int)
):
report["foreground"] = {
"classification": "environment_unavailable",
"stage": "throwaway_target",
}
else:
focus = await client.call_tool(
"bring_to_front",
{"pid": launched_pid, "window_id": window_id},
)
before = structured(
await client.call_tool(
"get_window_state",
{
"pid": launched_pid,
"window_id": window_id,
"session": session_id,
},
)
)
editor = next(
(
element
for element in before.get("elements") or []
if str(element.get("role") or "").lower()
in {"axtextarea", "axtextfield"}
),
None,
)
if not isinstance(editor, dict):
report["foreground"] = {
"classification": "unproven",
"stage": "editor_discovery",
}
else:
marker = "hermes foreground smoke"
type_args = {
"pid": launched_pid,
"window_id": window_id,
"element_index": editor.get("index"),
"text": marker,
"delivery_mode": "foreground",
"session": session_id,
}
token = editor.get("element_token")
if isinstance(token, str) and token:
type_args["element_token"] = token
typed = structured(
await client.call_tool("type_text", type_args)
)
saved = structured(
await client.call_tool(
"hotkey",
{
"pid": launched_pid,
"window_id": window_id,
"keys": ["cmd", "s"],
"delivery_mode": "foreground",
"session": session_id,
},
)
)
await client.call_tool(
"wait", {"seconds": 0.5, "session": session_id}
)
after = structured(
await client.call_tool(
"get_window_state",
{
"pid": launched_pid,
"window_id": window_id,
"session": session_id,
},
)
)
fresh_contains_marker = marker in json.dumps(
after.get("elements") or []
)
native_document_confirmed = textedit_process_contains(
launched_pid, marker
)
file_contains_marker = marker in smoke_path.read_text(
encoding="utf-8"
)
report["foreground"] = {
"classification": (
"pass"
if not focus.isError
and not refusal_code(typed)
and not refusal_code(saved)
and (
typed.get("verified") is True
or fresh_contains_marker
or native_document_confirmed
or file_contains_marker
)
else "unproven"
),
"focus_transport_ok": not focus.isError,
"effect": typed.get("effect"),
"verified": typed.get("verified"),
"fresh_state": bool(after.get("elements")),
"fresh_state_confirmed": fresh_contains_marker,
"native_document_confirmed": (
native_document_confirmed
),
"saved_file_confirmed": file_contains_marker,
"action_schema_omitted_bring_to_front": (
"bring_to_front" not in type_args
),
}
# Use only a driver-owned isolated profile. Never request,
# mint, print, or persist an existing-profile grant token.
listed = structured(
await client.call_tool(
"list_windows",
{"on_screen_only": True, "session": session_id},
)
)
browser_row = next(
(
row
for row in listed.get("windows") or []
if "chrome" in str(row.get("app_name") or "").lower()
),
None,
)
browser_pid = browser_row.get("pid") if browser_row else None
browser_window = (
browser_row.get("window_id") if browser_row else None
)
if not isinstance(browser_pid, int) or not isinstance(
browser_window, int
):
report["typed_browser"] = {
"classification": "environment_unavailable",
"stage": "browser_target",
}
else:
prepared = structured(
await client.call_tool(
"browser_prepare",
{
"pid": browser_pid,
"window_id": browser_window,
"allow_launch": True,
"profile": {"mode": "isolated_new"},
"session": session_id,
},
)
)
isolated_browser_pid = prepared.get("prepared_pid")
code = refusal_code(prepared)
if prepared.get("status") == "refused" or code:
report["typed_browser"] = {
"classification": "structured_refusal",
"code": code,
}
else:
prepared_pid = prepared.get("prepared_pid") or browser_pid
await client.call_tool(
"wait", {"seconds": 1, "session": session_id}
)
prepared_windows = structured(
await client.call_tool(
"list_windows",
{
"on_screen_only": True,
"session": session_id,
},
)
)
prepared_row = next(
(
row
for row in prepared_windows.get("windows") or []
if row.get("pid") == prepared_pid
),
None,
)
prepared_window = (
prepared_row.get("window_id")
if prepared_row
else browser_window
)
bound = structured(
await client.call_tool(
"get_browser_state",
{
"pid": prepared_pid,
"window_id": prepared_window,
"session": session_id,
},
)
)
tabs = bound.get("tabs") or []
tab_id = tabs[0].get("tab_id") if tabs else None
target_id = bound.get("target_id")
if (
bound.get("status") == "ok"
and bound.get("binding_quality") == "exact"
and bound.get("mutation_allowed") is True
and isinstance(tab_id, str)
and isinstance(target_id, str)
):
snapshot = structured(
await client.call_tool(
"get_browser_state",
{
"target_id": target_id,
"tab_id": tab_id,
"snapshot_format": "semantic_v2",
"session": session_id,
},
)
)
navigated = structured(
await client.call_tool(
"browser_navigate",
{
"target_id": target_id,
"tab_id": tab_id,
"url": "about:blank",
"session": session_id,
},
)
)
fresh = structured(
await client.call_tool(
"get_browser_state",
{
"target_id": target_id,
"tab_id": tab_id,
"snapshot_format": "semantic_v2",
"session": session_id,
},
)
)
report["typed_browser"] = {
"classification": "pass",
"exact_binding": True,
"mutation_allowed": True,
"initial_snapshot": snapshot.get("status")
in (None, "ok"),
"mutation_transport": navigated.get("status")
in (None, "ok"),
"fresh_verification": fresh.get("status")
in (None, "ok"),
}
else:
report["typed_browser"] = {
"classification": "unproven",
"stage": "exact_binding",
"code": refusal_code(bound),
}
finally:
if (
isinstance(launched_pid, int)
and launched_pid not in prior_foreground_pids
):
await client.call_tool(
"kill_app", {"pid": launched_pid, "session": session_id}
)
if (
isinstance(isolated_browser_pid, int)
and isolated_browser_pid != browser_pid
):
await client.call_tool(
"kill_app",
{"pid": isolated_browser_pid, "session": session_id},
)
await client.call_tool("end_session", {"session": session_id})
finally:
smoke_path.unlink(missing_ok=True)
return report
def main() -> int:
report: dict[str, dict[str, Any]] = {
"foreground": {"classification": "environment_unavailable"},
"typed_browser": {"classification": "environment_unavailable"},
}
if sys.platform != "darwin":
for cell in report.values():
cell["stage"] = "macos_host_required"
else:
socket_path = os.environ.get(
"CUA_DRIVER_LIVE_SOCKET", "/tmp/hermes-cua-0-9-live.sock"
)
if not Path(socket_path).is_socket():
for cell in report.values():
cell["stage"] = "isolated_daemon_required"
else:
try:
report = asyncio.run(run_smoke(socket_path))
except Exception as exc: # pragma: no cover - host/driver boundary
report = {
"foreground": {
"classification": "environment_unavailable",
"stage": "driver_connection",
"error_type": type(exc).__name__,
},
"typed_browser": {
"classification": "environment_unavailable",
"stage": "driver_connection",
"error_type": type(exc).__name__,
},
}
print(json.dumps(report, indent=2, sort_keys=True))
return int(any(cell.get("classification") != "pass" for cell in report.values()))
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -55,6 +55,18 @@ class TestAtexitTeardown:
cu_tool._shutdown_backend_atexit() # must not raise
assert cu_tool._backend is None
def test_shutdown_stops_every_session_backend(self):
"""Session-scoped caches are all drained, not only the legacy slot."""
first = MagicMock()
second = MagicMock()
with patch.object(cu_tool, "_backend", None), \
patch.object(cu_tool, "_backends", {"one": first, "two": second}), \
patch.object(cu_tool, "_backend_call_locks", {}):
cu_tool._shutdown_backend_atexit()
first.stop.assert_called_once()
second.stop.assert_called_once()
assert cu_tool._backends == {}
def test_hook_is_registered_with_atexit(self):
"""Importing the tool module registers the teardown hook.

View file

@ -0,0 +1,570 @@
{
"format": "normalized-selected-tools-list-v1",
"contract_epoch": "cua-driver-0.9",
"observed_reported_version": "0.8.3",
"capability_version": "1",
"observed_tool_count": 49,
"tools": [
{
"capabilities": [
"window.activate"
],
"inputSchema": {
"additionalProperties": false,
"properties": {
"pid": {
"type": "integer"
},
"window_id": {
"type": "integer"
}
},
"required": [
"pid"
],
"type": "object"
},
"name": "bring_to_front"
},
{
"capabilities": [
"browser.input.click"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"input_route": {
"enum": [
"trusted",
"dom_event"
],
"type": "string"
},
"ref": {
"type": "string"
},
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
},
"x": {
"type": "number"
},
"y": {
"type": "number"
}
},
"required": [
"target_id",
"tab_id"
],
"type": "object"
},
"name": "browser_click"
},
{
"capabilities": [
"browser.dialog"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"action": {
"enum": [
"inspect",
"accept",
"dismiss"
],
"type": "string"
},
"delivery_mode": {
"enum": [
"background",
"foreground"
],
"type": "string"
},
"dialog_id": {
"type": "string"
},
"prompt_text": {
"type": "string"
},
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
}
},
"required": [
"target_id",
"tab_id",
"action"
],
"type": "object"
},
"name": "browser_dialog"
},
{
"capabilities": [
"browser.download"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"destination_root": {
"type": "string"
},
"ref": {
"type": "string"
},
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
}
},
"required": [
"session",
"target_id",
"tab_id",
"ref",
"destination_root"
],
"type": "object"
},
"name": "browser_download"
},
{
"capabilities": [
"browser.navigate"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
},
"url": {
"type": "string"
}
},
"required": [
"target_id",
"tab_id",
"url"
],
"type": "object"
},
"name": "browser_navigate"
},
{
"capabilities": [
"browser.input.pointer"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"action": {
"enum": [
"hover",
"right_click",
"double_click",
"scroll",
"drag"
],
"type": "string"
},
"delta_x": {
"type": "number"
},
"delta_y": {
"type": "number"
},
"destination_ref": {
"type": "string"
},
"input_route": {
"enum": [
"trusted",
"dom_event"
],
"type": "string"
},
"ref": {
"type": "string"
},
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
},
"to_x": {
"type": "number"
},
"to_y": {
"type": "number"
},
"x": {
"type": "number"
},
"y": {
"type": "number"
}
},
"required": [
"target_id",
"tab_id",
"session",
"action"
],
"type": "object"
},
"name": "browser_pointer"
},
{
"capabilities": [
"browser.prepare"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"allow_launch": {
"type": "boolean"
},
"approval_token": {
"type": "string"
},
"pid": {
"type": "integer"
},
"profile": {
"additionalProperties": false,
"properties": {
"mode": {
"enum": [
"isolated_new",
"isolated_named"
],
"type": "string"
},
"name": {
"type": "string"
}
},
"required": [
"mode"
],
"type": "object"
},
"session": {
"type": "string"
},
"strategy": {
"additionalProperties": false,
"properties": {
"kind": {
"enum": [
"existing_profile"
],
"type": "string"
}
},
"required": [
"kind"
],
"type": "object"
},
"window_id": {
"type": "integer"
}
},
"required": [
"pid"
],
"type": "object"
},
"name": "browser_prepare"
},
{
"capabilities": [
"browser.input.files"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"files": {
"items": {
"type": "string"
},
"maxItems": 32,
"minItems": 1,
"type": "array"
},
"ref": {
"type": "string"
},
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
}
},
"required": [
"target_id",
"tab_id",
"ref",
"files"
],
"type": "object"
},
"name": "browser_set_input_files"
},
{
"capabilities": [
"browser.input.type"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"mode": {
"enum": [
"insert_text",
"keystrokes"
],
"type": "string"
},
"ref": {
"type": "string"
},
"session": {
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
},
"text": {
"type": "string"
}
},
"required": [
"target_id",
"tab_id",
"ref",
"text"
],
"type": "object"
},
"name": "browser_type"
},
{
"capabilities": [
"input.pointer.click",
"input.pointer.click.left",
"accessibility.element_tokens"
],
"inputSchema": {
"additionalProperties": false,
"properties": {
"action": {
"type": "string"
},
"button": {
"enum": [
"left",
"right",
"middle"
],
"type": "string"
},
"count": {
"type": "integer"
},
"debug_image_out": {
"type": "string"
},
"delivery_mode": {
"enum": [
"background",
"foreground"
],
"type": "string"
},
"element_index": {
"type": "integer"
},
"element_token": {
"type": "string"
},
"from_zoom": {
"type": "boolean"
},
"modifier": {
"items": {
"type": "string"
},
"type": "array"
},
"pid": {
"type": "integer"
},
"scope": {
"enum": [
"window",
"desktop"
],
"type": "string"
},
"session": {
"type": "string"
},
"window_id": {
"type": "integer"
},
"x": {
"type": "number"
},
"y": {
"type": "number"
}
},
"required": [],
"type": "object"
},
"name": "click"
},
{
"capabilities": [
"browser.state"
],
"inputSchema": {
"additionalProperties": true,
"properties": {
"continuation": {
"type": "string"
},
"pid": {
"type": "integer"
},
"query": {
"type": "string"
},
"scope_ref": {
"type": "string"
},
"session": {
"type": "string"
},
"snapshot_format": {
"enum": [
"dom_refs_v1",
"semantic_v2"
],
"type": "string"
},
"tab_id": {
"type": "string"
},
"target_id": {
"type": "string"
},
"window_id": {
"type": "integer"
}
},
"type": "object"
},
"name": "get_browser_state"
},
{
"capabilities": [
"input.keyboard.type",
"input.keyboard.type.terminal_safe",
"accessibility.element_tokens"
],
"inputSchema": {
"additionalProperties": false,
"properties": {
"delay_ms": {
"maximum": 200,
"minimum": 0,
"type": "integer"
},
"delivery_mode": {
"enum": [
"background",
"foreground"
],
"type": "string"
},
"element_index": {
"type": "integer"
},
"element_token": {
"type": "string"
},
"pid": {
"type": "integer"
},
"scope": {
"enum": [
"window",
"desktop"
],
"type": "string"
},
"session": {
"type": "string"
},
"text": {
"type": "string"
},
"window_id": {
"type": "integer"
},
"x": {
"type": "number"
},
"y": {
"type": "number"
}
},
"required": [
"text"
],
"type": "object"
},
"name": "type_text"
}
]
}

View file

@ -71,6 +71,96 @@ class TestInstallCuaDriverUpgrade:
assert tools_config.install_cua_driver(upgrade=True) is True
runner.assert_called_once()
def test_quiet_refresh_prints_single_contextual_progress_line(self):
import subprocess
from unittest.mock import MagicMock
from hermes_cli import tools_config
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
with patch("platform.system", return_value="Linux"), \
patch(
"subprocess.run",
return_value=MagicMock(returncode=0, stderr=""),
), \
patch("subprocess.Popen", return_value=fake_proc), \
patch.object(
tools_config.shutil,
"which",
return_value="/usr/local/bin/cua-driver",
), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_info") as info:
assert tools_config._run_cua_driver_installer(
label="Refreshing",
verbose=False,
) is True
info.assert_called_once_with(
"→ Refreshing cua-driver (Computer Use)..."
)
def test_quiet_refresh_can_suppress_progress_line(self):
from unittest.mock import MagicMock
from hermes_cli import tools_config
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
with patch("platform.system", return_value="Linux"), \
patch(
"subprocess.run",
return_value=MagicMock(returncode=0, stderr=""),
), \
patch("subprocess.Popen", return_value=fake_proc), \
patch.object(
tools_config.shutil,
"which",
return_value="/usr/local/bin/cua-driver",
), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_info") as info:
assert tools_config._run_cua_driver_installer(
label="Refreshing",
verbose=False,
show_progress=False,
) is True
info.assert_not_called()
def test_upgrade_can_suppress_installer_progress(self):
from hermes_cli import tools_config
with patch("platform.system", return_value="Darwin"), \
patch.object(
tools_config.shutil,
"which",
side_effect=lambda name: (
f"/usr/local/bin/{name}"
if name in {"cua-driver", "curl"}
else None
),
), \
patch.object(
tools_config,
"_run_cua_driver_installer",
return_value=True,
) as runner, \
patch("subprocess.run"):
assert tools_config.install_cua_driver(
upgrade=True,
show_installer_progress=False,
) is True
assert runner.call_args.kwargs["show_progress"] is False
def test_upgrade_on_macos_non_writable_applications_skips_refresh(self):
from hermes_cli import tools_config

View file

@ -0,0 +1,857 @@
"""Behavior contracts for cua-driver's verify/escalate and typed-browser ladder.
The fixture used here is a deliberately selected and normalized ``tools/list``
capture. It contains schemas, not machine/user state, and records the 0.9-era
contract where input properties are the discovery surface.
"""
from __future__ import annotations
import asyncio
import json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Dict, Optional
from unittest.mock import MagicMock, Mock, patch
import pytest
FIXTURE = Path(__file__).parents[1] / "fixtures" / "cua_driver_0_9_tools_list.json"
@pytest.fixture(autouse=True)
def _reset_computer_use_state():
from tools.computer_use.tool import reset_backend_for_tests
reset_backend_for_tests()
yield
reset_backend_for_tests()
class _FakeSession:
def __init__(
self,
out: Optional[Dict[str, Any]] = None,
*,
input_properties: Optional[Dict[str, set[str]]] = None,
tools: Optional[set[str]] = None,
) -> None:
self.out = out or {
"isError": False,
"data": {},
"structuredContent": {"effect": "confirmed"},
}
self.input_properties = input_properties or {}
self.tools = tools or {"bring_to_front", *self.input_properties}
self.calls: list[tuple[str, Dict[str, Any]]] = []
def call_tool(self, name: str, args: Dict[str, Any], timeout: float = 30.0):
self.calls.append((name, dict(args)))
return self.out
def supports_capability(self, capability: str, tool: Optional[str] = None) -> bool:
return False
def supports_input_property(self, tool: str, prop: str) -> bool:
return prop in self.input_properties.get(tool, set())
def _has_tool(self, name: str) -> bool:
return name in self.tools
def _make_backend(session: _FakeSession):
from tools.computer_use.cua_backend import CuaDriverBackend
backend = CuaDriverBackend.__new__(CuaDriverBackend)
backend._session = session
backend._session_id = "hermes-session"
backend._snapshot_tokens = {}
backend._active_pid = 42
backend._active_window_id = 7
return backend
def _driver_result(payload: Dict[str, Any]) -> Dict[str, Any]:
return {"isError": False, "data": {}, "structuredContent": payload}
# ---------------------------------------------------------------------------
# Selected live schema and foreground delivery
# ---------------------------------------------------------------------------
def test_normalized_fixture_is_sanitized_and_records_the_selected_contract():
fixture = json.loads(FIXTURE.read_text(encoding="utf-8"))
tools = {tool["name"]: tool for tool in fixture["tools"]}
assert fixture["contract_epoch"] == "cua-driver-0.9"
assert fixture["observed_reported_version"] == "0.8.3"
assert fixture["capability_version"] == "1"
assert fixture["observed_tool_count"] == 49
assert "delivery_mode" in tools["click"]["inputSchema"]["properties"]
assert "delivery_mode" in tools["type_text"]["inputSchema"]["properties"]
assert all(
"input.delivery_mode" not in tool["capabilities"] for tool in tools.values()
)
assert "bring_to_front" in tools
assert "bring_to_front" not in tools["click"]["inputSchema"]["properties"]
assert {
"get_browser_state",
"browser_prepare",
"browser_navigate",
"browser_click",
"browser_type",
"browser_pointer",
}.issubset(tools)
serialized = json.dumps(fixture)
for forbidden in (
"/Users/",
"\\Users\\",
"localhost",
"http://",
"https://",
"token-",
):
assert forbidden not in serialized
def test_foreground_support_is_discovered_from_tool_input_schema():
from tools.computer_use.cua_backend import _CuaDriverSession
fixture = json.loads(FIXTURE.read_text(encoding="utf-8"))
listed = []
for item in fixture["tools"]:
listed.append(
SimpleNamespace(
name=item["name"],
capabilities=item["capabilities"],
inputSchema=item["inputSchema"],
model_extra={},
)
)
class _McpSession:
async def list_tools(self):
return SimpleNamespace(tools=listed, model_extra={})
session = _CuaDriverSession.__new__(_CuaDriverSession)
session._capabilities = {}
session._input_properties = {}
session._capability_version = ""
asyncio.run(session._populate_capabilities(_McpSession()))
assert session.supports_input_property("click", "delivery_mode") is True
assert session.supports_input_property("type_text", "delivery_mode") is True
assert session.supports_input_property("bring_to_front", "delivery_mode") is False
assert session.supports_capability("input.delivery_mode", tool="click") is False
def test_foreground_focus_is_a_separate_call_before_action():
session = _FakeSession(input_properties={"click": {"delivery_mode"}})
backend = _make_backend(session)
result = backend.click(
element=3,
delivery_mode="foreground",
bring_to_front=True,
)
assert result.ok is True
assert [name for name, _ in session.calls] == ["bring_to_front", "click"]
focus_args = session.calls[0][1]
action_args = session.calls[1][1]
assert focus_args == {"pid": 42, "window_id": 7}
assert action_args["delivery_mode"] == "foreground"
assert "bring_to_front" not in action_args
def test_foreground_refuses_only_when_schema_lacks_delivery_property():
backend = _make_backend(_FakeSession())
result = backend.click(element=3, delivery_mode="foreground")
assert result.ok is False
assert result.code == "foreground_unsupported"
assert "update" not in result.message.lower()
assert backend._session.calls == []
def test_invalid_delivery_mode_is_rejected_before_driver_call():
session = _FakeSession(input_properties={"type_text": {"delivery_mode"}})
backend = _make_backend(session)
result = backend.type_text("hello", delivery_mode="sideways")
assert result.ok is False
assert result.code == "bad_delivery_mode"
assert session.calls == []
# ---------------------------------------------------------------------------
# Deterministic verdict precedence and backend isolation
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("result_kwargs", "decision"),
[
({"ok": True, "effect": "confirmed", "verified": True}, "done"),
(
{
"ok": True,
"effect": "unverifiable",
"verified": False,
"escalation": {"recommended": "foreground"},
},
"verify_fresh_state",
),
({"ok": True, "effect": "suspected_noop"}, "escalate"),
({"ok": False, "code": "browser_input_trust_unavailable"}, "escalate"),
],
)
def test_action_verdict_precedence(result_kwargs, decision):
from tools.computer_use.backend import ActionResult
from tools.computer_use.tool import _classify_action_result
result = ActionResult(action="click", **result_kwargs)
assert _classify_action_result(result)["decision"] == decision
def test_backends_are_isolated_by_hermes_session_and_reused_within_it():
from tools.computer_use import tool as computer_use
created = []
class _Backend:
def __init__(self):
created.append(self)
def start(self):
pass
def stop(self):
pass
with patch("tools.computer_use.cua_backend.CuaDriverBackend", _Backend):
first = computer_use._get_backend(session_id="conversation-a")
first_again = computer_use._get_backend(session_id="conversation-a")
second = computer_use._get_backend(session_id="conversation-b")
assert first is first_again
assert first is not second
assert created == [first, second]
def test_release_seam_stops_exact_backend_and_clears_session_state():
from tools.computer_use import tool as computer_use
first = MagicMock()
second = MagicMock()
computer_use._backends.update({
"conversation-a": first,
"conversation-b": second,
})
computer_use._backend_call_locks.update({
"conversation-a": computer_use.threading.RLock(),
"conversation-b": computer_use.threading.RLock(),
})
computer_use._session_auto_approve["conversation-a"] = True
computer_use._always_allow["conversation-a"] = {
("click", "background"),
}
assert computer_use.release_computer_use_session("conversation-a") is True
assert computer_use.release_computer_use_session("conversation-a") is False
first.stop.assert_called_once_with()
second.stop.assert_not_called()
assert "conversation-a" not in computer_use._backends
assert "conversation-a" not in computer_use._backend_call_locks
assert "conversation-a" not in computer_use._session_auto_approve
assert "conversation-a" not in computer_use._always_allow
assert computer_use._backends["conversation-b"] is second
def test_release_seam_evicts_state_even_when_backend_stop_fails():
from tools.computer_use import tool as computer_use
backend = MagicMock()
backend.stop.side_effect = RuntimeError("driver teardown failed")
computer_use._backends["failed-run"] = backend
computer_use._backend_call_locks["failed-run"] = computer_use.threading.RLock()
computer_use._session_auto_approve["failed-run"] = True
assert computer_use.release_computer_use_session("failed-run") is True
assert "failed-run" not in computer_use._backends
assert "failed-run" not in computer_use._backend_call_locks
assert "failed-run" not in computer_use._session_auto_approve
def test_release_seam_waits_for_in_flight_action_before_stopping_backend():
from tools.computer_use import tool as computer_use
backend = MagicMock()
call_lock = computer_use.threading.RLock()
computer_use._backends["cancelled-run"] = backend
computer_use._backend_call_locks["cancelled-run"] = call_lock
pool = ThreadPoolExecutor(max_workers=1)
try:
call_lock.acquire()
try:
released = pool.submit(
computer_use.release_computer_use_session,
"cancelled-run",
)
with pytest.raises(FutureTimeoutError):
released.result(timeout=0.05)
backend.stop.assert_not_called()
finally:
call_lock.release()
assert released.result(timeout=1) is True
finally:
pool.shutdown(wait=True)
backend.stop.assert_called_once_with()
def test_concurrent_hermes_sessions_do_not_share_backend_state():
from tools.computer_use import tool as computer_use
created = []
class _Backend:
def __init__(self):
self.marker = len(created)
created.append(self)
def start(self):
pass
def stop(self):
pass
def typed_browser_state(self, **kwargs):
return {"marker": self.marker, "pid": kwargs.get("pid")}
def invoke(session_id):
return json.loads(
computer_use.handle_computer_use(
{"action": "cua_browser_state", "pid": 101, "window_id": 202},
session_id=session_id,
)
)["marker"]
with patch("tools.computer_use.cua_backend.CuaDriverBackend", _Backend):
with ThreadPoolExecutor(max_workers=4) as executor:
markers = list(
executor.map(invoke, ["conversation-a", "conversation-b"] * 4)
)
assert set(markers[0::2]).isdisjoint(set(markers[1::2]))
assert len(set(markers[0::2])) == 1
assert len(set(markers[1::2])) == 1
assert len(created) == 2
def test_persistent_focus_has_a_separate_approval_scope():
from tools.computer_use import tool as computer_use
seen = []
def approve(action, args, summary):
seen.append(action)
return "approve_once" if action == "click" else "deny"
computer_use.set_approval_callback(approve)
try:
result = json.loads(
computer_use.handle_computer_use(
{
"action": "click",
"element": 1,
"delivery_mode": "foreground",
"bring_to_front": True,
},
session_id="approval-session",
)
)
finally:
computer_use.set_approval_callback(None)
assert seen == ["click", "bring_to_front"]
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_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_requires_interactive_driver_grant():
driver = _BrowserDriver()
route = _browser_route(driver)
result = route.prepare(
pid=101,
window_id=202,
profile_mode="existing_profile",
allow_launch=True,
)
assert result["code"] == "browser_consent_required"
assert result["interactive_grant_required"] is True
assert driver.calls == []
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": True,
"action": "click",
"verdict": {"decision": "verify_fresh_state"},
}
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]

View file

@ -39,18 +39,32 @@ def _reset():
class _FakeSession:
"""Minimal cua-driver session stub returning a canned tool result."""
def __init__(self, out: Dict[str, Any], capabilities: Optional[set] = None):
def __init__(
self,
out: Dict[str, Any],
capabilities: Optional[set] = None,
input_properties: Optional[Dict[str, set]] = None,
):
self._out = out
self._caps = capabilities or set()
self._input_properties = input_properties or {}
self.last_args: Dict[str, Any] = {}
self.calls = []
def call_tool(self, name: str, args: Dict[str, Any], timeout: float = 30.0):
self.last_args = args
self.calls.append((name, dict(args)))
return self._out
def supports_capability(self, capability: str, tool: Optional[str] = None) -> bool:
return capability in self._caps
def supports_input_property(self, tool: str, property_name: str) -> bool:
return property_name in self._input_properties.get(tool, set())
def _has_tool(self, name: str) -> bool:
return name == "bring_to_front"
def _make_backend(session: _FakeSession):
from tools.computer_use.cua_backend import CuaDriverBackend
@ -148,10 +162,14 @@ def test_text_response_surfaces_fields_additively():
assert payload["code"] == "background_unavailable"
assert payload["verified"] is False
# Bare result (old driver) → only ok/action, no None noise.
# Bare transport success still requires fresh verification, without None noise.
r2 = ActionResult(ok=True, action="click")
payload2 = json.loads(_text_response(r2))
assert payload2 == {"ok": True, "action": "click"}
assert payload2 == {
"ok": True,
"action": "click",
"verdict": {"decision": "verify_fresh_state"},
}
for k in ("effect", "escalation", "code", "verified", "path", "degraded", "delivery_mode"):
assert k not in payload2
@ -168,32 +186,34 @@ def test_background_is_default_no_flag_sent():
assert "delivery_mode" not in sess.last_args
def test_foreground_sent_when_capability_present():
def test_foreground_sent_when_schema_property_present():
out = {"isError": False, "data": {}, "structuredContent": {"effect": "unverifiable"}}
sess = _FakeSession(out, capabilities={"input.delivery_mode"})
sess = _FakeSession(out, input_properties={"click": {"delivery_mode"}})
be = _make_backend(sess)
res = be.click(element=1, delivery_mode="foreground", bring_to_front=True)
assert [name for name, _ in sess.calls] == ["bring_to_front", "click"]
assert sess.calls[0][1] == {"pid": 4242, "window_id": 7}
assert sess.last_args.get("delivery_mode") == "foreground"
assert sess.last_args.get("bring_to_front") is True
assert "bring_to_front" not in sess.last_args
assert res.delivery_mode == "foreground"
def test_foreground_refused_on_old_driver():
"""Old driver lacking the capability must NOT silently downgrade — it
"""A live action schema lacking the property must NOT silently downgrade — it
returns a structured foreground_unsupported result."""
out = {"isError": False, "data": {}, "structuredContent": {}}
sess = _FakeSession(out, capabilities=set()) # no input.delivery_mode
sess = _FakeSession(out)
be = _make_backend(sess)
res = be.click(element=1, delivery_mode="foreground")
assert res.ok is False
assert res.code == "foreground_unsupported"
# crucially: no tool call was made with a silent background downgrade
assert sess.last_args == {}
assert sess.calls == []
def test_bad_delivery_mode_rejected():
out = {"isError": False, "data": {}, "structuredContent": {}}
sess = _FakeSession(out, capabilities={"input.delivery_mode"})
sess = _FakeSession(out, input_properties={"type_text": {"delivery_mode"}})
be = _make_backend(sess)
res = be.type_text("hi", delivery_mode="sideways")
assert res.ok is False

View file

@ -96,7 +96,7 @@ class TestAgentCloseMethod:
"""Verify AIAgent.close() exists, is idempotent, and calls cleanup."""
def test_close_calls_cleanup_functions(self):
"""close() should call kill_all, cleanup_vm, cleanup_browser."""
"""close() should release every session-owned execution backend."""
from unittest.mock import patch
with patch("run_agent.AIAgent.__init__", return_value=None):
@ -109,7 +109,8 @@ class TestAgentCloseMethod:
with patch("tools.process_registry.process_registry") as mock_registry, \
patch("run_agent.cleanup_vm") as mock_cleanup_vm, \
patch("run_agent.cleanup_browser") as mock_cleanup_browser:
patch("run_agent.cleanup_browser") as mock_cleanup_browser, \
patch("tools.computer_use.release_computer_use_session") as mock_cleanup_cua:
agent.close()
mock_registry.kill_all.assert_called_once_with(
@ -117,6 +118,7 @@ class TestAgentCloseMethod:
)
mock_cleanup_vm.assert_called_once_with("test-close-cleanup")
mock_cleanup_browser.assert_called_once_with("test-close-cleanup")
mock_cleanup_cua.assert_called_once_with("test-close-cleanup")
def test_close_is_idempotent(self):
"""close() can be called multiple times without error."""
@ -134,6 +136,49 @@ class TestAgentCloseMethod:
agent.close()
agent.close()
def test_close_releases_computer_use_when_earlier_cleanup_fails(self):
"""One failed cleanup step must not strand the computer-use session."""
from unittest.mock import patch
with patch("run_agent.AIAgent.__init__", return_value=None):
from run_agent import AIAgent
agent = AIAgent.__new__(AIAgent)
agent.session_id = "test-close-after-failure"
agent._active_children = []
agent._active_children_lock = threading.Lock()
agent.client = None
with patch(
"tools.process_registry.process_registry.kill_all",
side_effect=RuntimeError("process cleanup failed"),
), patch(
"tools.computer_use.release_computer_use_session",
) as mock_cleanup_cua:
agent.close()
mock_cleanup_cua.assert_called_once_with(
"test-close-after-failure"
)
def test_soft_client_release_preserves_computer_use_session(self):
"""Cache eviction is not a hard session boundary."""
from unittest.mock import patch
with patch("run_agent.AIAgent.__init__", return_value=None):
from run_agent import AIAgent
agent = AIAgent.__new__(AIAgent)
agent.session_id = "test-soft-release"
agent._active_children = []
agent._active_children_lock = threading.Lock()
agent.client = None
with patch(
"tools.computer_use.release_computer_use_session",
) as mock_cleanup_cua:
agent.release_clients()
mock_cleanup_cua.assert_not_called()
def test_close_propagates_to_children(self):
"""close() should call close() on all active child agents."""
from unittest.mock import MagicMock, patch

View file

@ -40,4 +40,5 @@ from tools.computer_use.tool import ( # noqa: F401
set_approval_callback,
check_computer_use_requirements,
get_computer_use_schema,
release_computer_use_session,
)

View file

@ -212,6 +212,35 @@ 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."""

View file

@ -0,0 +1,560 @@
"""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 not performed here because it needs cua-driver's documented
interactive grant, not ordinary tool approval.
"""
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 the structured driver payload without discarding refusals."""
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)
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.
"""
normalized: Dict[str, Set[str]] = {}
snapshot = payload.get("snapshot")
# semantic_v2 carries the authoritative action-bearing entries in
# ``content_refs``; some transitional builds also emitted a ``refs`` list
# or map. Prefer the richer live shape, then accept both older forms.
raw = payload.get("content_refs")
if not raw:
raw = payload.get("refs")
if raw is None and isinstance(snapshot, dict):
raw = snapshot.get("refs")
if isinstance(raw, dict):
entries: Iterable[tuple[Optional[str], Any]] = raw.items()
elif isinstance(raw, list):
entries = ((None, item) for item in raw)
else:
entries = ()
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
normalized[ref] = {
action for action in (actions or []) if isinstance(action, str)
}
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,
) -> 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,
)
payload = self._call(
"get_browser_state",
{"pid": exact_pid, "window_id": exact_window},
)
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
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
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,
) -> Dict[str, Any]:
"""Run explicit isolated setup; refuse existing-profile attachment."""
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":
return _refusal(
"browser_consent_required",
"Existing-profile attachment requires cua-driver's interactive browser-approve grant bound to the exact pid, window, and session; ordinary tool approval is insufficient.",
interactive_grant_required=True,
)
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.",
)
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

View file

@ -58,6 +58,7 @@ from tools.computer_use.backend import (
ComputerUseBackend,
UIElement,
)
from tools.computer_use.browser_route import CuaTypedBrowserRoute
logger = logging.getLogger(__name__)
@ -937,6 +938,11 @@ class _CuaDriverSession:
# Empty until the session starts; consumers should call
# `supports_capability` rather than reading directly.
self._capabilities: Dict[str, set] = {}
# Raw input schemas are the compatibility source of truth for action
# properties. cua-driver 0.9-era builds advertise delivery_mode in
# inputSchema while intentionally omitting the old, fabricated
# ``input.delivery_mode`` capability token.
self._tool_schemas: Dict[str, Dict[str, Any]] = {}
self._capability_version: str = ""
# Lifecycle plumbing — see class docstring above.
self._ready_event = threading.Event()
@ -1045,6 +1051,9 @@ class _CuaDriverSession:
"""Surface 4: cache per-tool capability sets + capability_version
from tools/list. Soft prerequisite discovery failure leaves
the map empty and supports_capability degrades to False."""
self._capabilities = {}
self._tool_schemas = {}
self._capability_version = ""
try:
tools_list = await session.list_tools()
for tool in getattr(tools_list, "tools", []) or []:
@ -1063,6 +1072,14 @@ class _CuaDriverSession:
}
else:
self._capabilities[tool_name] = set()
schema = getattr(tool, "inputSchema", None)
if schema is None:
schema = (getattr(tool, "model_extra", None) or {}).get(
"inputSchema"
)
self._tool_schemas[tool_name] = (
dict(schema) if isinstance(schema, dict) else {}
)
# capability_version is a top-level sibling of `tools` on the
# tools/list response. cua-driver-core/src/tool.rs:354 emits
# it; cua-driver-core/src/protocol.rs:150 leaves it OUT of
@ -1194,6 +1211,17 @@ class _CuaDriverSession:
"""
return name in self._capabilities
def supports_input_property(self, tool: str, property_name: str) -> bool:
"""Return whether a live action schema accepts ``property_name``.
This deliberately inspects tools/list rather than guessing from the
package version or requiring a capability token the driver never
shipped. A missing/invalid schema fails closed.
"""
schema = getattr(self, "_tool_schemas", {}).get(tool, {})
properties = schema.get("properties") if isinstance(schema, dict) else None
return isinstance(properties, dict) and property_name in properties
@property
def capabilities_discovered(self) -> bool:
"""True once ``tools/list`` populated the per-tool map. When False,
@ -1734,6 +1762,23 @@ class CuaDriverBackend(ComputerUseBackend):
# degrade to the anonymous / unsynced path documented in the
# MCP server instructions.
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,
)
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:
@ -2313,13 +2358,12 @@ class CuaDriverBackend(ComputerUseBackend):
action: str,
args: Dict[str, Any],
delivery_mode: Optional[str],
bring_to_front: bool,
) -> Optional[ActionResult]:
"""Attach delivery_mode to an input-action args dict.
Background is the default and never needs a flag. Foreground is only
sent when the driver advertises support for it; on an older driver
that lacks the capability we refuse with a structured
sent when the live action schema accepts it; on an older driver that
lacks the property we refuse with a structured
``foreground_unsupported`` result instead of silently downgrading to
background (which would land the input somewhere the model didn't
expect). Returns an ActionResult to short-circuit on refusal, or None
@ -2333,23 +2377,74 @@ class CuaDriverBackend(ComputerUseBackend):
message=f"unknown delivery_mode {delivery_mode!r} — use background|foreground.",
)
# Foreground requested. Only send it if the driver understands it.
if not self._session.supports_capability(
"input.delivery_mode", tool=action
):
if not self._session.supports_input_property(action, "delivery_mode"):
return ActionResult(
ok=False, action=action, code="foreground_unsupported",
delivery_mode="foreground",
message=(
"This cua-driver build does not support foreground "
"delivery (no `input.delivery_mode` capability). Update "
"cua-driver to escalate to the foreground rung."
"The connected cua-driver action schema does not accept "
"delivery_mode, so foreground delivery is unavailable. "
"Use another verified rung without assuming the reported "
"package version describes the live schema."
),
)
args["delivery_mode"] = "foreground"
if bring_to_front:
args["bring_to_front"] = True
return None
def _run_input_action(
self,
action: str,
args: Dict[str, Any],
delivery_mode: Optional[str],
bring_to_front: bool,
) -> ActionResult:
"""Apply one delivery rung, optionally focusing via its own tool.
``bring_to_front`` is never an input-action property. When explicitly
requested, the separately approved standalone focus action runs first,
then the original foreground input runs unchanged.
"""
refusal = self._apply_delivery(action, args, delivery_mode)
if refusal is not None:
return refusal
if bring_to_front:
if delivery_mode != "foreground":
return ActionResult(
ok=False,
action=action,
code="bring_to_front_requires_foreground",
message="bring_to_front requires delivery_mode='foreground'.",
)
if not self._session._has_tool("bring_to_front"):
return ActionResult(
ok=False,
action=action,
code="bring_to_front_unsupported",
delivery_mode="foreground",
message="The connected cua-driver does not advertise the standalone bring_to_front tool.",
)
if self._active_pid is None or self._active_window_id is None:
return ActionResult(
ok=False,
action=action,
code="bring_to_front_target_required",
delivery_mode="foreground",
message="Capture an exact target before requesting persistent foreground focus.",
)
focused = self.bring_to_front(
pid=self._active_pid,
window_id=self._active_window_id,
)
if not focused.ok:
return focused
result = self._action(action, args)
if bring_to_front:
result.meta["foreground_focus"] = {
"invoked": True,
"tool": "bring_to_front",
}
return result
def click(
self,
*,
@ -2401,10 +2496,7 @@ class CuaDriverBackend(ComputerUseBackend):
if modifiers:
args["modifier"] = modifiers
refusal = self._apply_delivery(tool, args, delivery_mode, bring_to_front)
if refusal is not None:
return refusal
return self._action(tool, args)
return self._run_input_action(tool, args, delivery_mode, bring_to_front)
def drag(
self,
@ -2440,10 +2532,7 @@ class CuaDriverBackend(ComputerUseBackend):
else:
return ActionResult(ok=False, action="drag",
message="drag requires from_element/to_element or from_coordinate/to_coordinate.")
refusal = self._apply_delivery("drag", args, delivery_mode, bring_to_front)
if refusal is not None:
return refusal
return self._action("drag", args)
return self._run_input_action("drag", args, delivery_mode, bring_to_front)
def scroll(
self,
@ -2485,10 +2574,7 @@ class CuaDriverBackend(ComputerUseBackend):
args["x"] = x
args["y"] = y
args["window_id"] = self._active_window_id
refusal = self._apply_delivery("scroll", args, delivery_mode, bring_to_front)
if refusal is not None:
return refusal
return self._action("scroll", args)
return self._run_input_action("scroll", args, delivery_mode, bring_to_front)
# ── Keyboard ───────────────────────────────────────────────────
def type_text(self, text: str, *, delivery_mode: Optional[str] = None,
@ -2499,10 +2585,7 @@ class CuaDriverBackend(ComputerUseBackend):
return ActionResult(ok=False, action="type_text",
message="No active window — call capture() first.")
args: Dict[str, Any] = {"pid": pid, "window_id": window_id, "text": text}
refusal = self._apply_delivery("type_text", args, delivery_mode, bring_to_front)
if refusal is not None:
return refusal
return self._action("type_text", args)
return self._run_input_action("type_text", args, delivery_mode, bring_to_front)
def key(self, keys: str, *, delivery_mode: Optional[str] = None,
bring_to_front: bool = False) -> ActionResult:
@ -2521,16 +2604,10 @@ class CuaDriverBackend(ComputerUseBackend):
# hotkey requires at least one modifier + one key.
args: Dict[str, Any] = {"pid": pid, "window_id": window_id,
"keys": modifiers + [key_name]}
refusal = self._apply_delivery("hotkey", args, delivery_mode, bring_to_front)
if refusal is not None:
return refusal
return self._action("hotkey", args)
return self._run_input_action("hotkey", args, delivery_mode, bring_to_front)
else:
args = {"pid": pid, "window_id": window_id, "key": key_name}
refusal = self._apply_delivery("press_key", args, delivery_mode, bring_to_front)
if refusal is not None:
return refusal
return self._action("press_key", args)
return self._run_input_action("press_key", args, delivery_mode, bring_to_front)
# ── Value setter ────────────────────────────────────────────────
def set_value(self, value: str, element: Optional[int] = None) -> ActionResult:
@ -2592,7 +2669,7 @@ class CuaDriverBackend(ComputerUseBackend):
return self._load_windows()
def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
"""Target an app for subsequent actions without stealing system focus.
"""Target an app, optionally invoking standalone foreground focus.
cua-driver background-automation never needs to bring a window to the
front: capture(app=...) already selects the right window via
@ -2601,8 +2678,9 @@ class CuaDriverBackend(ComputerUseBackend):
its pid/window_id so that subsequent click/type calls hit the right
process.
raise_window=True is intentionally ignored: stealing the user's focus
is exactly what this backend is designed to avoid.
The default remains non-disruptive. ``raise_window=True`` is explicit,
separately approved by the Hermes adapter, and uses cua-driver's
standalone ``bring_to_front`` tool rather than an action property.
"""
try:
windows = self._load_windows()
@ -2625,6 +2703,23 @@ class CuaDriverBackend(ComputerUseBackend):
"pid": self._active_pid,
"window_id": self._active_window_id,
}
if raise_window:
if not self._session._has_tool("bring_to_front"):
return ActionResult(
ok=False,
action="focus_app",
code="bring_to_front_unsupported",
message="The connected cua-driver does not advertise the standalone bring_to_front tool.",
)
focused = self.bring_to_front(
pid=self._active_pid,
window_id=self._active_window_id,
)
if not focused.ok:
return focused
focused.action = "focus_app"
focused.meta["target_selected"] = True
return focused
return ActionResult(
ok=True, action="focus_app",
message=f"Targeted {target['app_name']} (pid {self._active_pid}, "
@ -2686,7 +2781,29 @@ class CuaDriverBackend(ComputerUseBackend):
args: Dict[str, Any] = {"pid": int(pid)}
if window_id is not None:
args["window_id"] = int(window_id)
return self._action("bring_to_front", args)
# The live 0.9-era schema is strict and deliberately has no session
# property. It is a standalone native focus operation, not a
# 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."""
return self._browser_route().prepare(**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 ─────────────────────────────
@ -2936,7 +3053,13 @@ class CuaDriverBackend(ComputerUseBackend):
return
args["element_token"] = token
def _action(self, name: str, args: Dict[str, Any]) -> ActionResult:
def _action(
self,
name: str,
args: Dict[str, Any],
*,
inject_session: bool = True,
) -> ActionResult:
# Attach the snapshot's element_token whenever the call carries
# an element_index and the target tool advertises support.
self._maybe_attach_element_token(name, args)
@ -2944,7 +3067,8 @@ class CuaDriverBackend(ComputerUseBackend):
# and per-session state (config overrides, recording ownership)
# stay tied to this run. setdefault preserves any explicit
# session a caller already supplied.
args.setdefault("session", self._session_id)
if inject_session:
args.setdefault("session", self._session_id)
try:
out = self._session.call_tool(name, args)
except Exception as e:
@ -2969,4 +3093,3 @@ class CuaDriverBackend(ComputerUseBackend):
meta.update(structured)
return _action_result_from(name, ok, message, meta, structured,
requested_delivery=args.get("delivery_mode"))

View file

@ -46,6 +46,15 @@ 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 "
@ -228,25 +237,100 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = {
"`background` (DEFAULT) routes input to the target without "
"raising it or stealing focus — the co-work model. "
"`foreground` briefly fronts the window, acts, then "
"restores the prior frontmost app. Only escalate to "
"`foreground` when a background attempt did NOT land — i.e. "
"a prior result had `effect: 'suspected_noop'`, "
"`code: 'background_unavailable'`, or "
"`escalation.recommended: 'foreground'`. Do not predict it "
"from the app being Electron/Chromium; react to the "
"returned signal. Foreground is a visible focus change and "
"needs its own approval."
"restores the prior frontmost app. A `confirmed` effect is "
"done. For `unverifiable`, inspect fresh state before any "
"retry even if escalation is recommended. Escalate only "
"after `suspected_noop` or a structured refusal. Do not "
"predict the rung from the app being Electron/Chromium. "
"Foreground is a visible focus change and needs its own "
"approval."
),
},
"bring_to_front": {
"type": "boolean",
"description": (
"Optional, pairs with delivery_mode='foreground'. Keep the "
"target fronted after the action instead of restoring the "
"previous app, to avoid a per-call flash across a short "
"sequence of foreground actions. Default false."
"Optional and only valid with delivery_mode='foreground'. "
"Explicitly invokes cua-driver's standalone bring_to_front "
"tool before the input; it is never passed as an input "
"property. This persistent focus change has a separate "
"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.",
},
"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. existing_profile always requires "
"the driver's separate interactive grant."
),
},
"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.",
},
"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",

View file

@ -78,12 +78,17 @@ def set_approval_callback(cb) -> None:
# Actions that read, not mutate. Always allowed.
_SAFE_ACTIONS = frozenset({"capture", "wait", "list_apps"})
_SAFE_ACTIONS = frozenset({
"capture", "wait", "list_apps", "list_windows", "cua_browser_state",
})
# 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
@ -141,11 +146,15 @@ def _is_blocked_type(text: str) -> Optional[str]:
# Backend selection — env-swappable for tests
# ---------------------------------------------------------------------------
# Per-process cached backend; lazily instantiated on first call.
# Per-Hermes-session cached backends. Each backend owns its own cua-driver
# session, native target, typed-browser binding, refs, and grant namespace.
_backend_lock = threading.Lock()
# Backward-compatible empty-session injection hook used by older tests.
# Process-scoped aux-vision routing cache: (provider, model) → bool.
_AUX_VISION_ROUTE_CACHE: Dict[Tuple[str, str], bool] = {}
_backend: Optional[ComputerUseBackend] = None
_backends: Dict[str, ComputerUseBackend] = {}
_backend_call_locks: Dict[str, threading.RLock] = {}
# Approval state, scoped per conversation/run (keyed by session_id) so a
# gateway serving concurrent sessions can't leak one run's "always approve"
# unlock into another. Falls back to a shared "" bucket for callers that
@ -158,37 +167,92 @@ _session_auto_approve: Dict[str, bool] = {}
_always_allow: Dict[str, set] = {}
def _get_backend() -> ComputerUseBackend:
def _get_backend(session_id: str = "") -> ComputerUseBackend:
global _backend
sid = str(session_id or "")
with _backend_lock:
if _backend is None:
backend_name = os.environ.get("HERMES_COMPUTER_USE_BACKEND", "cua").lower()
if backend_name in {"cua", "cua-driver", ""}:
from tools.computer_use.cua_backend import CuaDriverBackend
_backend = CuaDriverBackend()
elif backend_name == "noop": # pragma: no cover
_backend = _NoopBackend()
else:
raise RuntimeError(f"Unknown HERMES_COMPUTER_USE_BACKEND={backend_name!r}")
try:
_backend.start()
except Exception:
# Don't cache a backend whose start() failed (e.g. a lazy
# dependency install was declined / failed). The next call
# retries cleanly instead of returning a half-initialised
# backend.
_backend = None
raise
return _backend
if sid == "" and _backend is not None:
return _backend
cached = _backends.get(sid)
if cached is not None:
return cached
backend_name = os.environ.get("HERMES_COMPUTER_USE_BACKEND", "cua").lower()
if backend_name in {"cua", "cua-driver", ""}:
from tools.computer_use.cua_backend import CuaDriverBackend
backend = CuaDriverBackend()
elif backend_name == "noop": # pragma: no cover
backend = _NoopBackend()
else:
raise RuntimeError(f"Unknown HERMES_COMPUTER_USE_BACKEND={backend_name!r}")
try:
backend.start()
except Exception:
# Don't cache a backend whose start() failed (e.g. a lazy
# dependency install was declined / failed). The next call
# retries cleanly instead of returning a half-initialised backend.
raise
_backends[sid] = backend
_backend_call_locks[sid] = threading.RLock()
if sid == "":
_backend = backend
return backend
def release_computer_use_session(session_id: str) -> bool:
"""Release one session-owned computer-use backend.
This is the production lifecycle seam for hosts and policy plugins. It
removes the exact session backend and its call lock before stopping the
backend, so new lookups cannot retain the stale target/ref namespace.
Approval state is cleared even when no backend was started.
Returns ``True`` when a backend was found and released, ``False`` when the
session was already absent. Safe to call repeatedly.
"""
global _backend
sid = str(session_id or "")
with _backend_lock:
backend = _backends.pop(sid, None)
call_lock = _backend_call_locks.pop(sid, None)
# Preserve the backward-compatible empty-session injection hook:
# older callers/tests may populate only `_backend`.
if sid == "" and backend is None:
backend = _backend
if sid == "" and _backend is backend:
_backend = None
with _approval_lock:
_session_auto_approve.pop(sid, None)
_always_allow.pop(sid, None)
if backend is None:
return False
try:
# Let an in-flight action finish before ending the driver session and
# dropping its target/ref state. Do not hold the global cache lock
# while waiting: unrelated Hermes sessions remain independent.
if call_lock is not None:
with call_lock:
backend.stop()
else:
backend.stop()
except Exception:
logger.debug(
"computer_use backend release failed for session %s",
sid,
exc_info=True,
)
return True
def _shutdown_backend_atexit() -> None:
"""Stop the cached backend so the cua-driver child doesn't outlive us.
"""Stop all cached backends so cua-driver children don't outlive us.
The backend is cached per-process and holds a long-lived ``cua-driver``
subprocess, so without this the driver survives the Hermes process that
spawned it (#28152 item 3). #69903 kept the orphan from burning a core by
disabling the cursor overlay; the process itself still lingered.
Each session backend holds a long-lived ``cua-driver`` subprocess, so
without this a driver can survive the Hermes process that spawned it
(#28152 item 3). #69903 kept the orphan from burning a core by disabling
the cursor overlay; the process itself still lingered.
Mirrors ``browser_tool``'s ``atexit.register(_emergency_cleanup_all_sessions)``
same spawn-and-drive-a-subprocess shape. atexit only, no signal handlers:
@ -197,16 +261,35 @@ def _shutdown_backend_atexit() -> None:
exception escaping atexit prints a traceback on every exit.
"""
global _backend
# Drop the lock before stop() — teardown budgets 5s and shouldn't block
# an unrelated caller waiting to spawn.
# Drop the global lock before stop() — teardown budgets 5s and shouldn't
# block an unrelated caller waiting to spawn.
with _backend_lock:
backend, _backend = _backend, None
if backend is None:
return
try:
backend.stop()
except Exception as e:
logger.debug("cua-driver atexit teardown failed: %s", e)
unique = {
id(backend): (backend, _backend_call_locks.get(sid))
for sid, backend in _backends.items()
}
if _backend is not None:
unique.setdefault(
id(_backend),
(_backend, _backend_call_locks.get("")),
)
_backend = None
_backends.clear()
_backend_call_locks.clear()
with _approval_lock:
_session_auto_approve.clear()
_always_allow.clear()
for backend, call_lock in unique.values():
try:
if call_lock is not None:
with call_lock:
backend.stop()
else:
backend.stop()
except Exception as e:
logger.debug("cua-driver atexit teardown failed: %s", e)
atexit.register(_shutdown_backend_atexit)
@ -300,7 +383,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 == "type":
if action in {"type", "cua_browser_type"}:
text = args.get("text", "")
pat = _is_blocked_type(text)
if pat:
@ -319,15 +402,30 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
"hint": "Destructive system shortcuts are hard-blocked.",
})
if args.get("bring_to_front") and args.get("delivery_mode") != "foreground":
return json.dumps({
"error": "bring_to_front requires delivery_mode='foreground'",
"code": "bring_to_front_requires_foreground",
})
# Approval gate (destructive actions only).
if action in _DESTRUCTIVE_ACTIONS:
err = _request_approval(action, args, session_id)
if err is not None:
return err
# Persistent focus is a separate, visible side effect from the input
# itself. Keep its approval scope distinct even when the input rung has
# already been approved for this session.
if args.get("bring_to_front") or (
action == "focus_app" and args.get("raise_window")
):
err = _request_approval("bring_to_front", args, session_id)
if err is not None:
return err
# Dispatch to backend.
try:
backend = _get_backend()
backend = _get_backend(session_id=session_id)
except Exception as e:
return json.dumps({
"error": f"computer_use backend unavailable: {e}",
@ -336,7 +434,10 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
})
try:
return _dispatch(backend, action, args)
with _backend_lock:
call_lock = _backend_call_locks.setdefault(session_id, threading.RLock())
with call_lock:
return _dispatch(backend, action, args)
except Exception as e:
logger.exception("computer_use %s failed", action)
return json.dumps({"error": f"{action} failed: {e}"})
@ -446,6 +547,93 @@ 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"),
):
if args.get(public) is not None:
state_args[internal] = args[public]
return json.dumps(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"),
"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")
@ -528,7 +716,27 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
# Response shaping
# ---------------------------------------------------------------------------
def _text_response(res: ActionResult) -> str:
def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
"""Choose the next ladder step from semantic evidence, in precedence order.
An escalation recommendation is advisory. It never overrides a confirmed
effect and it never turns an unverifiable action into permission to repeat
input. The model must first obtain fresh evidence.
"""
if res.effect == "confirmed" or res.verified is True:
return {"decision": "done"}
if res.effect == "unverifiable":
return {"decision": "verify_fresh_state"}
if res.effect == "suspected_noop" or not res.ok or res.code is not None:
decision: Dict[str, Any] = {"decision": "escalate"}
if isinstance(res.escalation, dict):
decision["recommended"] = res.escalation.get("recommended")
return decision
# Transport success without semantic proof is not proof of effect.
return {"decision": "verify_fresh_state"}
def _action_payload(res: ActionResult) -> Dict[str, Any]:
payload: Dict[str, Any] = {"ok": res.ok, "action": res.action}
if res.message:
payload["message"] = res.message
@ -552,7 +760,12 @@ def _text_response(res: ActionResult) -> str:
payload["code"] = res.code
if res.meta:
payload["meta"] = res.meta
return json.dumps(payload)
payload["verdict"] = _classify_action_result(res)
return payload
def _text_response(res: ActionResult) -> str:
return json.dumps(_action_payload(res))
# Default cap for the AX `elements` array returned by capture. Dense UIs
@ -989,19 +1202,20 @@ def _maybe_follow_capture(
# Combine action summary with the capture.
resp = _capture_response(cap)
if isinstance(resp, dict) and resp.get("_multimodal"):
prefix = f"[{res.action}] ok={res.ok}" + (f"{res.message}" if res.message else "")
# Keep the complete evidence/verdict contract visible when an image is
# attached; otherwise capture_after would accidentally discard the
# very signal that governs whether repeating input is allowed.
prefix = json.dumps(_action_payload(res))
resp["content"][0]["text"] = prefix + "\n\n" + resp["content"][0]["text"]
resp["text_summary"] = prefix + "\n\n" + resp["text_summary"]
resp["action_result"] = _action_payload(res)
return resp
# Fallback: action + text capture merged.
try:
data = json.loads(resp)
except (TypeError, json.JSONDecodeError):
data = {"capture": resp}
data["action"] = res.action
data["ok"] = res.ok
if res.message:
data["message"] = res.message
data.update(_action_payload(res))
return json.dumps(data)

View file

@ -11,6 +11,7 @@ from tools.computer_use.schema import COMPUTER_USE_SCHEMA
from tools.computer_use.tool import (
check_computer_use_requirements,
handle_computer_use,
release_computer_use_session,
set_approval_callback,
)
from tools.registry import registry
@ -36,4 +37,5 @@ __all__ = [
"handle_computer_use",
"set_approval_callback",
"check_computer_use_requirements",
"release_computer_use_session",
]