feat(computer_use): follow cua-driver's verify → escalate ladder (#67123)
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.
This commit is contained in:
Teknium
GitHub
parent
2637aa607f
commit
9d6d772837
7 changed files with 673 additions and 46 deletions
|
|
@ -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 "
|
||||
|
|
|
|||
|
|
@ -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 |
|
||||
|
|
|
|||
291
tests/tools/test_computer_use_delivery_ladder.py
Normal file
291
tests/tools/test_computer_use_delivery_ladder.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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 ───────────────────────────────────────────────
|
||||
|
|
|
|||
|
|
@ -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"))
|
||||
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue