From 9d6d7728376d110c468d15e0fdc9964b31099a00 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:59:35 -0700 Subject: [PATCH] =?UTF-8?q?feat(computer=5Fuse):=20follow=20cua-driver's?= =?UTF-8?q?=20verify=20=E2=86=92=20escalate=20ladder=20(#67123)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hermes' computer_use wrapper dropped cua-driver's structured action verdicts, exposed no delivery_mode, and injected background-only guidance — so the agent reported unverified no-ops as success and concluded cua-driver 'cannot drive' Electron/Chromium surfaces (observed live on tldraw offline). Fixes #67052. Phase A — preserve the result contract: - ActionResult carries verified/effect/escalation/path/degraded/code/delivery_mode - CuaDriverBackend._action() reads structuredContent (was data-only); a helper normalizes it, additive and None-safe on old drivers - _text_response surfaces the fields additively (ok stays transport-only) Phase B — bounded, model-reachable foreground: - delivery_mode (background|foreground) + bring_to_front on the schema, dispatcher, ABC, and all input methods - foreground is capability-gated (input.delivery_mode); old drivers get a structured foreground_unsupported refusal, never a silent background downgrade - no automatic/hidden foreground retry — the model selects it from the signal Phase C — guidance + isolation: - system prompt (prompt_builder) and bundled skills/computer-use/SKILL.md go from background-ONLY to background-FIRST, teaching the AX→PX→foreground ladder driven by returned effect/escalation, not predicted from the app being Electron - foreground approval scoped by (action, delivery_mode): a background approval never silently authorizes foreground - approval state keyed per session_id so concurrent gateway runs don't leak unlocks Tests: tests/tools/test_computer_use_delivery_ladder.py (15) cover confirmed/ unverifiable/suspected_noop/degraded/old-driver verdicts, delivery_mode gating + foreground_unsupported, and session-scoped foreground approval. Existing 265 computer_use tests still green. Live E2E (real cua-driver 0.8.3 + tldraw offline on Linux/X11): a background click returned effect='unverifiable'/path='ax' (no fabricated success), and a foreground request returned code='foreground_unsupported' — correct on a driver that predates the input.delivery_mode capability. --- agent/prompt_builder.py | 23 ++ skills/computer-use/SKILL.md | 52 +++- .../test_computer_use_delivery_ladder.py | 291 ++++++++++++++++++ tools/computer_use/backend.py | 41 ++- tools/computer_use/cua_backend.py | 165 +++++++++- tools/computer_use/schema.py | 29 ++ tools/computer_use/tool.py | 118 +++++-- 7 files changed, 673 insertions(+), 46 deletions(-) create mode 100644 tests/tools/test_computer_use_delivery_ladder.py diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index c3433b38b..25d900341 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -557,6 +557,29 @@ def computer_use_guidance(platform_name: Optional[str] = None) -> str: "4. After any state-changing action, re-capture to verify. You can " "pass `capture_after=true` to get the follow-up screenshot in one " "round-trip.\n\n" + "## Verify → escalate ladder (background-first, NOT background-only)\n" + "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: '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" + " - `'px'` → re-issue addressing the target by `coordinate=[x,y]` " + "read off the screenshot instead of `element`.\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 " + "when the user isn't actively working. Common for Electron/Chromium " + "consent dialogs, DirectInput games, and raw-input canvases.\n" + "- Escalate to foreground as a REACTION to a returned signal, never " + "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" "## 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/skills/computer-use/SKILL.md b/skills/computer-use/SKILL.md index 6c7fe9816..85c73dfa5 100644 --- a/skills/computer-use/SKILL.md +++ b/skills/computer-use/SKILL.md @@ -101,6 +101,56 @@ All actions accept optional `capture_after=True` to get a follow-up 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. + +## The verify → escalate ladder (background-first) + +cua-driver delivers input in the **background** by default (no focus steal), +but that is the first rung, not the only one. Every input action returns a +structured verdict; read it and climb only when the driver tells you to. + +Returned fields (present when the driver supports them): +- `effect`: `"confirmed"` (driver read the result back — done), `"unverifiable"` + (delivered, but confirm it yourself by re-capturing), or `"suspected_noop"` + (ran but almost certainly did nothing). +- `escalation`: `{recommended: "px" | "foreground" | "page", reason}` — present + only when there's a next rung to try. +- `code`: a structured refusal like `"background_unavailable"` or + `"foreground_unsupported"`. +- `verified`: `true` only on AX read-back. + +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, + 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 + (it's a visible focus change) and is only appropriate when the user isn't + actively working. Classic cases: Electron/Chromium consent dialogs (e.g. + tldraw offline's "Run Script"), DirectInput games, raw-input canvases. + +``` +computer_use(action="click", element=7) +# → {effect: "suspected_noop", escalation: {recommended: "foreground", ...}} +computer_use(action="click", element=7, delivery_mode="foreground") +# → {effect: "unverifiable", path: "x11_pixel_fg"} then re-capture to confirm +``` + +**Escalate to foreground as a REACTION to a returned signal, never as a +prediction** from the app being Electron/Chromium/GTK. 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. + ### Key shortcuts vary per platform Use the host's idiomatic modifier: @@ -205,7 +255,7 @@ 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 | Re-capture and verify. A modal that wasn't visible before may be blocking input. Dismiss it (usually `escape` or click its close button) before retrying | +| 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 | | 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 | diff --git a/tests/tools/test_computer_use_delivery_ladder.py b/tests/tools/test_computer_use_delivery_ladder.py new file mode 100644 index 000000000..4f03d2dbb --- /dev/null +++ b/tests/tools/test_computer_use_delivery_ladder.py @@ -0,0 +1,291 @@ +"""Regression tests for the cua-driver verify → escalate ladder. + +Covers NousResearch/hermes-agent#67052: + - Phase A: cua-driver structured verdicts (verified/effect/escalation/code/ + degraded/path) are preserved through ActionResult and surfaced in the + model-facing response, additively (old drivers omit them cleanly). + - Phase B: delivery_mode is model-reachable, capability-gated, and refuses + with foreground_unsupported on an old driver rather than silently + downgrading to background. + - Phase C: foreground approval is scoped by (action, delivery_mode) and by + session_id, so a background approval never silently authorizes foreground + and one run's unlock never leaks into another. + +Stdlib + pytest + unittest.mock only. No live cua-driver, no network. +""" + +from __future__ import annotations + +import json +import os +from typing import Any, Dict, Optional +from unittest.mock import patch + +import pytest + + +@pytest.fixture(autouse=True) +def _reset(): + from tools.computer_use.tool import reset_backend_for_tests + reset_backend_for_tests() + yield + reset_backend_for_tests() + + +# --------------------------------------------------------------------------- +# Phase A — structured verdict normalization (_action_result_from) +# --------------------------------------------------------------------------- + +class _FakeSession: + """Minimal cua-driver session stub returning a canned tool result.""" + + def __init__(self, out: Dict[str, Any], capabilities: Optional[set] = None): + self._out = out + self._caps = capabilities or set() + self.last_args: Dict[str, Any] = {} + + def call_tool(self, name: str, args: Dict[str, Any], timeout: float = 30.0): + self.last_args = args + return self._out + + def supports_capability(self, capability: str, tool: Optional[str] = None) -> bool: + return capability in self._caps + + +def _make_backend(session: _FakeSession): + from tools.computer_use.cua_backend import CuaDriverBackend + be = CuaDriverBackend.__new__(CuaDriverBackend) + be._session = session # type: ignore[attr-defined] + be._session_id = "test-run" # type: ignore[attr-defined] + be._snapshot_tokens = {} # type: ignore[attr-defined] + be._active_pid = 4242 # type: ignore[attr-defined] + be._active_window_id = 7 # type: ignore[attr-defined] + return be + + +def test_confirmed_verdict_is_preserved(): + out = { + "isError": False, "data": {"message": "ok"}, + "structuredContent": {"verified": True, "effect": "confirmed", "path": "ax"}, + } + be = _make_backend(_FakeSession(out)) + res = be.click(element=3) + assert res.ok is True + assert res.verified is True + assert res.effect == "confirmed" + assert res.path == "ax" + assert res.escalation is None + + +def test_suspected_noop_carries_escalation(): + out = { + "isError": False, "data": {}, + "structuredContent": { + "effect": "suspected_noop", + "escalation": {"recommended": "foreground", "reason": "occluded renderer"}, + "code": "background_unavailable", + }, + } + be = _make_backend(_FakeSession(out)) + res = be.click(element=3) + assert res.effect == "suspected_noop" + assert res.escalation == {"recommended": "foreground", "reason": "occluded renderer"} + assert res.code == "background_unavailable" + # transport ok, but semantically not confirmed + assert res.verified is None + + +def test_unverifiable_distinct_from_success_and_failure(): + out = { + "isError": False, "data": {}, + "structuredContent": {"effect": "unverifiable", "verified": False, "path": "x11_pixel"}, + } + be = _make_backend(_FakeSession(out)) + res = be.click(x=10, y=20) + assert res.ok is True # transport succeeded + assert res.verified is False # ... but not confirmed + assert res.effect == "unverifiable" + + +def test_degraded_capture_signal_preserved(): + out = { + "isError": False, "data": {}, + "structuredContent": {"effect": "suspected_noop", "degraded": True, + "escalation": {"recommended": "px", "reason": "empty tree"}}, + } + be = _make_backend(_FakeSession(out)) + res = be.scroll(direction="down", element=1) + assert res.degraded is True + assert res.escalation["recommended"] == "px" + + +def test_old_driver_without_structured_content_is_clean(): + """A driver that returns no structuredContent leaves every verdict field + None — unchanged behavior, no crash.""" + out = {"isError": False, "data": {"message": "done"}, "structuredContent": None} + be = _make_backend(_FakeSession(out)) + res = be.click(element=3) + assert res.ok is True + assert res.message == "done" + assert res.verified is None + assert res.effect is None + assert res.escalation is None + assert res.code is None + assert res.path is None + + +def test_text_response_surfaces_fields_additively(): + from tools.computer_use.backend import ActionResult + from tools.computer_use.tool import _text_response + + # Full verdict → all fields present. + r = ActionResult(ok=True, action="click", effect="suspected_noop", + escalation={"recommended": "foreground"}, code="background_unavailable", + path="ax", verified=False) + payload = json.loads(_text_response(r)) + assert payload["effect"] == "suspected_noop" + assert payload["escalation"] == {"recommended": "foreground"} + assert payload["code"] == "background_unavailable" + assert payload["verified"] is False + + # Bare result (old driver) → only ok/action, no None noise. + r2 = ActionResult(ok=True, action="click") + payload2 = json.loads(_text_response(r2)) + assert payload2 == {"ok": True, "action": "click"} + for k in ("effect", "escalation", "code", "verified", "path", "degraded", "delivery_mode"): + assert k not in payload2 + + +# --------------------------------------------------------------------------- +# Phase B — delivery_mode threading + capability gating +# --------------------------------------------------------------------------- + +def test_background_is_default_no_flag_sent(): + out = {"isError": False, "data": {}, "structuredContent": {"effect": "confirmed"}} + sess = _FakeSession(out) + be = _make_backend(sess) + be.click(element=1) # no delivery_mode + assert "delivery_mode" not in sess.last_args + + +def test_foreground_sent_when_capability_present(): + out = {"isError": False, "data": {}, "structuredContent": {"effect": "unverifiable"}} + sess = _FakeSession(out, capabilities={"input.delivery_mode"}) + be = _make_backend(sess) + res = be.click(element=1, delivery_mode="foreground", bring_to_front=True) + assert sess.last_args.get("delivery_mode") == "foreground" + assert sess.last_args.get("bring_to_front") is True + assert res.delivery_mode == "foreground" + + +def test_foreground_refused_on_old_driver(): + """Old driver lacking the capability 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 + 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 == {} + + +def test_bad_delivery_mode_rejected(): + out = {"isError": False, "data": {}, "structuredContent": {}} + sess = _FakeSession(out, capabilities={"input.delivery_mode"}) + be = _make_backend(sess) + res = be.type_text("hi", delivery_mode="sideways") + assert res.ok is False + assert res.code == "bad_delivery_mode" + + +def test_dispatcher_threads_delivery_mode_to_backend(): + """End-to-end through the tool dispatcher with the noop backend.""" + from tools.computer_use import tool as cu + with patch.dict(os.environ, {"HERMES_COMPUTER_USE_BACKEND": "noop"}, clear=False): + cu.reset_backend_for_tests() + be = cu._get_backend() + cu.handle_computer_use({"action": "click", "element": 5, + "delivery_mode": "foreground"}) + # noop records kwargs; find the click call + clicks = [kw for (name, kw) in be.calls if name == "click"] # type: ignore[attr-defined] + assert clicks and clicks[-1].get("delivery_mode") == "foreground" + + +# --------------------------------------------------------------------------- +# Phase C — foreground approval scoping (action + delivery_mode + session) +# --------------------------------------------------------------------------- + +def test_background_approval_does_not_authorize_foreground(): + from tools.computer_use import tool as cu + + seen = [] + + def cb(action, args, summary): + seen.append((action, args.get("delivery_mode"))) + return "approve_session" + + cu.set_approval_callback(cb) + try: + # Background click, approve for session. + assert cu._request_approval("click", {}, "sess-A") is None + # A second background click needs no prompt (cached). + assert cu._request_approval("click", {}, "sess-A") is None + assert len(seen) == 1 + # Foreground click on the SAME action must prompt again — the + # background approval does not cover it. + assert cu._request_approval("click", {"delivery_mode": "foreground"}, "sess-A") is None + assert len(seen) == 2 + assert seen[-1] == ("click", "foreground") + finally: + cu.set_approval_callback(None) + + +def test_approval_state_is_session_scoped(): + from tools.computer_use import tool as cu + + calls = [] + + def cb(action, args, summary): + calls.append((action, args.get("delivery_mode"))) + return "approve_session" + + cu.set_approval_callback(cb) + try: + # Run A approves foreground click. + cu._request_approval("click", {"delivery_mode": "foreground"}, "run-A") + # Run B has NOT — it must prompt independently. + n_before = len(calls) + cu._request_approval("click", {"delivery_mode": "foreground"}, "run-B") + assert len(calls) == n_before + 1 + finally: + cu.set_approval_callback(None) + + +def test_always_approve_covers_foreground(): + from tools.computer_use import tool as cu + + calls = [] + + def cb(action, args, summary): + calls.append(action) + return "always_approve" + + cu.set_approval_callback(cb) + try: + # First call unlocks everything for this session. + cu._request_approval("click", {}, "run-C") + # Foreground now sails through without another prompt. + cu._request_approval("click", {"delivery_mode": "foreground"}, "run-C") + assert len(calls) == 1 + finally: + cu.set_approval_callback(None) + + +def test_foreground_summary_warns_about_focus_change(): + from tools.computer_use.tool import _summarize_action + s = _summarize_action("click", {"element": 3, "delivery_mode": "foreground"}) + assert "FOREGROUND" in s + bg = _summarize_action("click", {"element": 3}) + assert "FOREGROUND" not in bg diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index ba109cf2e..c98726848 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -69,7 +69,16 @@ class CaptureResult: @dataclass class ActionResult: - """Result of any action (click / type / scroll / drag / key / wait).""" + """Result of any action (click / type / scroll / drag / key / wait). + + Beyond the transport-level ``ok`` flag, this carries cua-driver's + structured action verdict so the model can follow the documented + verify → escalate ladder (NousResearch/hermes-agent#67052). ``ok`` stays + tool/transport success only — it is NOT the semantic verdict. Read + ``effect`` / ``escalation`` to decide the next rung. All structured + fields are optional and additive: an older driver that omits + ``structuredContent`` leaves them ``None`` and behavior is unchanged. + """ ok: bool action: str @@ -79,6 +88,24 @@ class ActionResult: capture: Optional[CaptureResult] = None # Arbitrary extra fields for debugging / telemetry. meta: Dict[str, Any] = field(default_factory=dict) + # ── cua-driver structured verdict (additive; None on old drivers) ── + # AX read-back verification: True = driver read the effect back, + # False = ran but unconfirmed, None = tool doesn't carry the field. + verified: Optional[bool] = None + # Confidence signal: "confirmed" | "unverifiable" | "suspected_noop". + effect: Optional[str] = None + # Machine-readable next-rung hint: {"recommended": "px"|"foreground"|"page", + # "reason": str} — present only when the driver recommends climbing. + escalation: Optional[Dict[str, Any]] = None + # Delivery rung that actually ran (e.g. "ax", "x11_pixel", "cgevent_fg"). + path: Optional[str] = None + # True when an AX walk found no actionable elements (act by px instead). + degraded: Optional[bool] = None + # The delivery_mode the caller requested for this action, echoed back. + delivery_mode: Optional[str] = None + # A structured refusal code (e.g. "background_unavailable", + # "foreground_unsupported", "desktop_scope_disabled") when present. + code: Optional[str] = None class ComputerUseBackend(ABC): @@ -118,6 +145,8 @@ class ComputerUseBackend(ABC): button: str = "left", # left | right | middle click_count: int = 1, modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, # background (default) | foreground + bring_to_front: bool = False, ) -> ActionResult: ... @abstractmethod @@ -130,6 +159,8 @@ class ComputerUseBackend(ABC): to_xy: Optional[Tuple[int, int]] = None, button: str = "left", modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, + bring_to_front: bool = False, ) -> ActionResult: ... @abstractmethod @@ -142,14 +173,18 @@ class ComputerUseBackend(ABC): x: Optional[int] = None, y: Optional[int] = None, modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, + bring_to_front: bool = False, ) -> ActionResult: ... # ── Keyboard ──────────────────────────────────────────────────── @abstractmethod - def type_text(self, text: str) -> ActionResult: ... + def type_text(self, text: str, *, delivery_mode: Optional[str] = None, + bring_to_front: bool = False) -> ActionResult: ... @abstractmethod - def key(self, keys: str) -> ActionResult: + def key(self, keys: str, *, delivery_mode: Optional[str] = None, + bring_to_front: bool = False) -> ActionResult: """Send a key combo, e.g. 'cmd+s', 'ctrl+alt+t', 'return'.""" # ── Introspection ─────────────────────────────────────────────── diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index 1481d09fb..ad1f1cb52 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -59,6 +59,70 @@ from tools.computer_use.backend import ( logger = logging.getLogger(__name__) +def _action_result_from( + name: str, + ok: bool, + message: str, + meta: Dict[str, Any], + structured: Dict[str, Any], + *, + requested_delivery: Optional[str] = None, +) -> ActionResult: + """Build an ActionResult, lifting cua-driver's structured verdict. + + All structured fields are additive: a driver that omits + ``structuredContent`` (or any individual field) leaves the corresponding + ActionResult attribute ``None``, so callers and tests see unchanged + behavior on old drivers. See the action response shape in + cua-driver's mcp-tool-notes and NousResearch/hermes-agent#67052. + """ + sc = structured if isinstance(structured, dict) else {} + + def _pick(key: str) -> Any: + # structuredContent is canonical; fall back to a flattened meta copy. + if key in sc: + return sc.get(key) + return meta.get(key) + + verified = _pick("verified") + if not isinstance(verified, bool): + verified = None + effect = _pick("effect") + if not isinstance(effect, str): + effect = None + escalation = _pick("escalation") + if not isinstance(escalation, dict): + escalation = None + path = _pick("path") + if not isinstance(path, str): + path = None + degraded = _pick("degraded") + if not isinstance(degraded, bool): + degraded = None + # Refusal/limitation code — drivers spell it "code" or "reason_code". + code = _pick("code") or _pick("reason_code") + if not isinstance(code, str): + code = None + # Echo the delivery mode the caller actually requested (the driver's + # `path` records the rung that ran; this records what we asked for). + delivery_mode = requested_delivery if isinstance(requested_delivery, str) else None + + return ActionResult( + ok=ok, + action=name, + message=message, + meta=meta, + verified=verified, + effect=effect, + escalation=escalation, + path=path, + degraded=degraded, + delivery_mode=delivery_mode, + code=code, + ) + + + # --------------------------------------------------------------------------- # Update checking # --------------------------------------------------------------------------- @@ -1766,6 +1830,48 @@ class CuaDriverBackend(ComputerUseBackend): ) # ── Pointer ──────────────────────────────────────────────────── + def _apply_delivery( + self, + 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 + ``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 + to proceed. See NousResearch/hermes-agent#67052 phase B. + """ + if not delivery_mode or delivery_mode == "background": + return None + if delivery_mode != "foreground": + return ActionResult( + ok=False, action=action, code="bad_delivery_mode", + 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 + ): + 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." + ), + ) + args["delivery_mode"] = "foreground" + if bring_to_front: + args["bring_to_front"] = True + return None + def click( self, *, @@ -1775,6 +1881,8 @@ class CuaDriverBackend(ComputerUseBackend): button: str = "left", click_count: int = 1, modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, + bring_to_front: bool = False, ) -> ActionResult: pid = self._active_pid if pid is None: @@ -1815,6 +1923,9 @@ 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) def drag( @@ -1826,6 +1937,8 @@ class CuaDriverBackend(ComputerUseBackend): to_xy: Optional[Tuple[int, int]] = None, button: str = "left", modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, + bring_to_front: bool = False, ) -> ActionResult: pid = self._active_pid if pid is None: @@ -1849,6 +1962,9 @@ 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) def scroll( @@ -1860,6 +1976,8 @@ class CuaDriverBackend(ComputerUseBackend): x: Optional[int] = None, y: Optional[int] = None, modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, + bring_to_front: bool = False, ) -> ActionResult: pid = self._active_pid if pid is None: @@ -1889,18 +2007,27 @@ 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) # ── Keyboard ─────────────────────────────────────────────────── - def type_text(self, text: str) -> ActionResult: + def type_text(self, text: str, *, delivery_mode: Optional[str] = None, + bring_to_front: bool = False) -> ActionResult: pid = self._active_pid window_id = self._active_window_id if pid is None or window_id is None: return ActionResult(ok=False, action="type_text", message="No active window — call capture() first.") - return self._action("type_text", {"pid": pid, "window_id": window_id, "text": text}) + 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) - def key(self, keys: str) -> ActionResult: + def key(self, keys: str, *, delivery_mode: Optional[str] = None, + bring_to_front: bool = False) -> ActionResult: pid = self._active_pid window_id = self._active_window_id if pid is None or window_id is None: @@ -1914,11 +2041,18 @@ class CuaDriverBackend(ComputerUseBackend): if modifiers: # hotkey requires at least one modifier + one key. - return self._action("hotkey", {"pid": pid, "window_id": window_id, - "keys": modifiers + [key_name]}) + 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) else: - return self._action("press_key", {"pid": pid, "window_id": window_id, - "key": key_name}) + 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) # ── Value setter ──────────────────────────────────────────────── def set_value(self, value: str, element: Optional[int] = None) -> ActionResult: @@ -2325,11 +2459,22 @@ class CuaDriverBackend(ComputerUseBackend): logger.exception("cua-driver %s call failed", name) return ActionResult(ok=False, action=name, message=f"cua-driver error: {e}") ok = not out["isError"] - message = "" data = out["data"] + structured = out.get("structuredContent") or {} + message = "" if isinstance(data, dict): message = str(data.get("message", "")) elif isinstance(data, str): message = data - return ActionResult(ok=ok, action=name, message=message, - meta=data if isinstance(data, dict) else {}) + if not message and isinstance(structured, dict): + message = str(structured.get("message", "")) + # Merge data + structuredContent into meta for debugging, structured + # winning on key overlap (it is the canonical verdict surface). + meta: Dict[str, Any] = {} + if isinstance(data, dict): + meta.update(data) + if isinstance(structured, dict): + 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 613185d83..ed3eea224 100644 --- a/tools/computer_use/schema.py +++ b/tools/computer_use/schema.py @@ -218,6 +218,35 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "matching the background co-work model." ), }, + # ── delivery (verify → escalate ladder) ──────────────── + "delivery_mode": { + "type": "string", + "enum": ["background", "foreground"], + "description": ( + "How input is delivered, for the input actions (click, " + "double_click, right_click, drag, scroll, type, key). " + "`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." + ), + }, + "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." + ), + }, # ── return shape ─────────────────────────────────────── "capture_after": { "type": "boolean", diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index fec551b60..7097de848 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -139,9 +139,16 @@ def _is_blocked_type(text: str) -> Optional[str]: # Per-process cached backend; lazily instantiated on first call. _backend_lock = threading.Lock() _backend: Optional[ComputerUseBackend] = None -# Session-scoped approval state. -_session_auto_approve = False -_always_allow: set = set() # action names the user unlocked for the session +# 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 +# don't pass a session_id (e.g. the classic single-run CLI). Values: +# _session_auto_approve[sid] -> bool ("always_approve everything") +# _always_allow[sid] -> set of (action, delivery_mode) scope keys +# See NousResearch/hermes-agent#67052 gap 4. +_approval_lock = threading.Lock() +_session_auto_approve: Dict[str, bool] = {} +_always_allow: Dict[str, set] = {} def _get_backend() -> ComputerUseBackend: @@ -169,8 +176,8 @@ def _get_backend() -> ComputerUseBackend: def reset_backend_for_tests() -> None: # pragma: no cover - """Test helper — tear down the cached backend.""" - global _backend, _session_auto_approve, _always_allow + """Test helper — tear down the cached backend and per-session state.""" + global _backend with _backend_lock: if _backend is not None: try: @@ -178,8 +185,9 @@ def reset_backend_for_tests() -> None: # pragma: no cover except Exception: pass _backend = None - _session_auto_approve = False - _always_allow = set() + with _approval_lock: + _session_auto_approve.clear() + _always_allow.clear() class _NoopBackend(ComputerUseBackend): # pragma: no cover @@ -219,12 +227,12 @@ class _NoopBackend(ComputerUseBackend): # pragma: no cover self.calls.append(("scroll", kw)) return ActionResult(ok=True, action="scroll") - def type_text(self, text: str) -> ActionResult: - self.calls.append(("type", {"text": text})) + def type_text(self, text: str, **kw) -> ActionResult: + self.calls.append(("type", {"text": text, **kw})) return ActionResult(ok=True, action="type") - def key(self, keys: str) -> ActionResult: - self.calls.append(("key", {"keys": keys})) + def key(self, keys: str, **kw) -> ActionResult: + self.calls.append(("key", {"keys": keys, **kw})) return ActionResult(ok=True, action="key") def list_apps(self) -> List[Dict[str, Any]]: @@ -257,6 +265,8 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: action = (args.get("action") or "").strip().lower() if not action: return json.dumps({"error": "missing `action`"}) + # Per-run key for approval-state isolation across concurrent sessions. + session_id = str(kwargs.get("session_id") or "") # Safety: validate actions before approval prompt. if action == "type": @@ -280,7 +290,7 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: # Approval gate (destructive actions only). if action in _DESTRUCTIVE_ACTIONS: - err = _request_approval(action, args) + err = _request_approval(action, args, session_id) if err is not None: return err @@ -301,13 +311,26 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: return json.dumps({"error": f"{action} failed: {e}"}) -def _request_approval(action: str, args: Dict[str, Any]) -> Optional[str]: - """Return None if approved, or a JSON error string if denied.""" - global _session_auto_approve, _always_allow - if _session_auto_approve: - return None - if action in _always_allow: - return None +def _request_approval(action: str, args: Dict[str, Any], + session_id: str = "") -> Optional[str]: + """Return None if approved, or a JSON error string if denied. + + Approval is scoped by (action, delivery_mode) AND by session_id. + Foreground delivery is a visible focus change, so a prior background + approval — even ``approve_session`` on the same action — must NOT + silently authorize it (NousResearch/hermes-agent#67052). + ``always_approve`` (the blanket "auto-approve everything" unlock) still + covers foreground, since the user explicitly opted into unattended + operation. State is keyed on session_id so concurrent runs don't leak + unlocks into one another. + """ + is_foreground = args.get("delivery_mode") == "foreground" + scope_key = (action, "foreground" if is_foreground else "background") + with _approval_lock: + if _session_auto_approve.get(session_id): + return None + if scope_key in _always_allow.get(session_id, set()): + return None cb = _approval_callback if cb is None: # No CLI approval wired — default allow. Gateway approval is handled @@ -322,35 +345,38 @@ def _request_approval(action: str, args: Dict[str, Any]) -> Optional[str]: if verdict == "approve_once": return None if verdict == "approve_session" or verdict == "always_approve": - _always_allow.add(action) - if verdict == "always_approve": - _session_auto_approve = True + with _approval_lock: + _always_allow.setdefault(session_id, set()).add(scope_key) + if verdict == "always_approve": + _session_auto_approve[session_id] = True return None return json.dumps({"error": "denied by user", "action": action}) def _summarize_action(action: str, args: Dict[str, Any]) -> str: + fg = " [FOREGROUND — briefly raises the window / changes focus]" \ + if args.get("delivery_mode") == "foreground" else "" if action in {"click", "double_click", "right_click", "middle_click"}: if args.get("element") is not None: - return f"{action} element #{args['element']}" + return f"{action} element #{args['element']}{fg}" coord = args.get("coordinate") if coord: - return f"{action} at {tuple(coord)}" - return action + return f"{action} at {tuple(coord)}{fg}" + return action + fg if action == "drag": src = args.get("from_element") or args.get("from_coordinate") dst = args.get("to_element") or args.get("to_coordinate") - return f"drag {src} → {dst}" + return f"drag {src} → {dst}{fg}" if action == "scroll": - return f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}" + return f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}{fg}" if action == "type": text = args.get("text", "") - return f"type {text[:60]!r}" + ("..." if len(text) > 60 else "") + return f"type {text[:60]!r}" + ("..." if len(text) > 60 else "") + fg if action == "key": - return f"key {args.get('keys', '')!r}" + return f"key {args.get('keys', '')!r}{fg}" if action == "focus_app": return f"focus {args.get('app', '')!r}" + (" (raise)" if args.get("raise_window") else "") - return action + return action + fg def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> Any: @@ -389,6 +415,11 @@ 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) + # 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") + bring_to_front = bool(args.get("bring_to_front")) + if action in {"click", "double_click", "right_click", "middle_click"}: button = args.get("button") click_count = 1 @@ -407,6 +438,7 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> element=element if element is not None else None, x=x, y=y, button=button or "left", click_count=click_count, modifiers=args.get("modifiers"), + delivery_mode=delivery_mode, bring_to_front=bring_to_front, ) return _maybe_follow_capture(backend, res, capture_after) @@ -424,6 +456,7 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> to_xy=tuple(args["to_coordinate"]) if args.get("to_coordinate") else None, button=args.get("button", "left"), modifiers=args.get("modifiers"), + delivery_mode=delivery_mode, bring_to_front=bring_to_front, ) return _maybe_follow_capture(backend, res, capture_after) @@ -436,15 +469,18 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> x=coord[0] if coord and coord[0] is not None else None, y=coord[1] if coord and coord[1] is not None else None, modifiers=args.get("modifiers"), + delivery_mode=delivery_mode, bring_to_front=bring_to_front, ) return _maybe_follow_capture(backend, res, capture_after) if action == "type": - res = backend.type_text(args.get("text", "")) + res = backend.type_text(args.get("text", ""), + delivery_mode=delivery_mode, bring_to_front=bring_to_front) return _maybe_follow_capture(backend, res, capture_after) if action == "key": - res = backend.key(args.get("keys", "")) + res = backend.key(args.get("keys", ""), + delivery_mode=delivery_mode, bring_to_front=bring_to_front) return _maybe_follow_capture(backend, res, capture_after) if action == "set_value": @@ -465,6 +501,24 @@ def _text_response(res: ActionResult) -> str: payload: Dict[str, Any] = {"ok": res.ok, "action": res.action} if res.message: payload["message"] = res.message + # Surface cua-driver's structured verdict additively so the model can + # follow the verify → escalate ladder. Only include fields the driver + # actually returned (None = old driver / not carried). ok is transport + # success; effect/escalation are the semantic verdict. + if res.verified is not None: + payload["verified"] = res.verified + if res.effect is not None: + payload["effect"] = res.effect + if res.escalation is not None: + payload["escalation"] = res.escalation + if res.path is not None: + payload["path"] = res.path + if res.degraded is not None: + payload["degraded"] = res.degraded + if res.delivery_mode is not None: + payload["delivery_mode"] = res.delivery_mode + if res.code is not None: + payload["code"] = res.code if res.meta: payload["meta"] = res.meta return json.dumps(payload)