diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index 845e4260ddb..92080053990 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -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 " diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 0f3c1d666e8..85a62aeea70 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -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() diff --git a/hermes_cli/update_cmd.py b/hermes_cli/update_cmd.py index f2f503374be..5e70487698f 100644 --- a/hermes_cli/update_cmd.py +++ b/hermes_cli/update_cmd.py @@ -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) diff --git a/run_agent.py b/run_agent.py index dde8a8797cc..cb80963595b 100644 --- a/run_agent.py +++ b/run_agent.py @@ -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 diff --git a/skills/autonomous-ai-agents/computer-use/SKILL.md b/skills/autonomous-ai-agents/computer-use/SKILL.md index 07fe306c7dc..0e1d2491aa9 100644 --- a/skills/autonomous-ai-agents/computer-use/SKILL.md +++ b/skills/autonomous-ai-agents/computer-use/SKILL.md @@ -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/ diff --git a/tests/computer_use/live_cua_0_9_smoke.py b/tests/computer_use/live_cua_0_9_smoke.py new file mode 100644 index 00000000000..fb51c513ff9 --- /dev/null +++ b/tests/computer_use/live_cua_0_9_smoke.py @@ -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()) diff --git a/tests/computer_use/test_cua_atexit_teardown.py b/tests/computer_use/test_cua_atexit_teardown.py index ff02d94fc5a..64150da8471 100644 --- a/tests/computer_use/test_cua_atexit_teardown.py +++ b/tests/computer_use/test_cua_atexit_teardown.py @@ -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. diff --git a/tests/fixtures/cua_driver_0_9_tools_list.json b/tests/fixtures/cua_driver_0_9_tools_list.json new file mode 100644 index 00000000000..8a218c2a2a4 --- /dev/null +++ b/tests/fixtures/cua_driver_0_9_tools_list.json @@ -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" + } + ] +} diff --git a/tests/hermes_cli/test_install_cua_driver.py b/tests/hermes_cli/test_install_cua_driver.py index cc3f3012dd8..b7687587002 100644 --- a/tests/hermes_cli/test_install_cua_driver.py +++ b/tests/hermes_cli/test_install_cua_driver.py @@ -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 diff --git a/tests/tools/test_computer_use_cua_0_9.py b/tests/tools/test_computer_use_cua_0_9.py new file mode 100644 index 00000000000..99350caaa0e --- /dev/null +++ b/tests/tools/test_computer_use_cua_0_9.py @@ -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] diff --git a/tests/tools/test_computer_use_delivery_ladder.py b/tests/tools/test_computer_use_delivery_ladder.py index 5facd3e847c..ee1c2ff9a88 100644 --- a/tests/tools/test_computer_use_delivery_ladder.py +++ b/tests/tools/test_computer_use_delivery_ladder.py @@ -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 diff --git a/tests/tools/test_zombie_process_cleanup.py b/tests/tools/test_zombie_process_cleanup.py index b4679ffbe3a..ba3c20eebce 100644 --- a/tests/tools/test_zombie_process_cleanup.py +++ b/tests/tools/test_zombie_process_cleanup.py @@ -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 diff --git a/tools/computer_use/__init__.py b/tools/computer_use/__init__.py index 3c3404a6480..6a9028f5b53 100644 --- a/tools/computer_use/__init__.py +++ b/tools/computer_use/__init__.py @@ -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, ) diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index c98726848b7..9e233459f75 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -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.""" diff --git a/tools/computer_use/browser_route.py b/tools/computer_use/browser_route.py new file mode 100644 index 00000000000..1f4ad4f791b --- /dev/null +++ b/tools/computer_use/browser_route.py @@ -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 diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index dbc71ab58f8..d0d685484b6 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -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")) - diff --git a/tools/computer_use/schema.py b/tools/computer_use/schema.py index ed3eea2249d..656e5f24b32 100644 --- a/tools/computer_use/schema.py +++ b/tools/computer_use/schema.py @@ -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", diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index 59b5e1820d4..cdc9e1b646d 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -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) diff --git a/tools/computer_use_tool.py b/tools/computer_use_tool.py index e9f4f4f8e2b..318813d5945 100644 --- a/tools/computer_use_tool.py +++ b/tools/computer_use_tool.py @@ -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", ]