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

Merge branch 'simp/r2-tools-b-cua2-capt' into simp/r2-tools-b-cua2

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:51:31 -07:00
parent 26862482c5 303a8cafb0
commit 89f08dfdb6
4 changed files with 342 additions and 580 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/tools/test_computer_use.py
View File

@@ -2051,8 +2051,8 @@ class TestStructuredElementsConsumption:
`structuredContent.elements` part of every `get_window_state` MCP
response. The wrapper used to parse the markdown AX tree with a
regex — lossy because bounds always came back (0,0,0,0). The
structured path preserves real frames, so UIElement.center() works
against pixel coordinates instead of just an index lookup.
structured path preserves real frames, so UIElement.bounds carries
pixel coordinates instead of just an index lookup.
"""
def test_structured_parser_reads_frames(self):

99
tools/computer_use/backend.py
View File

@@ -1,13 +1,11 @@
"""Abstract backend interface for computer use.
Any implementation (cua-driver over MCP, pyautogui, noop, future Linux/Windows)
returns the shapes below. All methods are synchronous; async is handled inside
the backend implementation if needed.
"""
"""Abstract backend interface for computer use. Any implementation (cua-driver over MCP,
pyautogui, noop, future Linux/Windows) returns the shapes below. All methods are synchronous;
async is handled inside the backend implementation if needed."""
from __future__ import annotations
import struct
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
@@ -19,22 +17,17 @@ def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]:
segments (skipping 0xFF fill bytes) to the first SOF marker; stop at SOS. Used by the
tool layer's provider min-size guard."""
if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24:
try:
width, height = struct.unpack(">II", raw[16:24])
return int(width), int(height)
except Exception:
return None
width, height = struct.unpack(">II", raw[16:24]) # cannot fail: 8 bytes are guaranteed present
return int(width), int(height)
if raw.startswith(b"\xff\xd8") and len(raw) > 4:
i = 2
while i + 9 < len(raw):
if raw[i] != 0xFF:
i += 1
continue
marker = raw[i + 1]
i += 2
marker, i = raw[i + 1], i + 2
while marker == 0xFF and i < len(raw):
marker = raw[i]
i += 1
marker, i = raw[i], i + 1
if marker in {0xD8, 0xD9}:
continue
if marker == 0xDA or i + 2 > len(raw):
@@ -43,9 +36,7 @@ def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]:
if segment_len < 2 or i + segment_len > len(raw):
break
if marker in _JPEG_SOF_MARKERS and segment_len >= 7:
height = int.from_bytes(raw[i + 3:i + 5], "big")
width = int.from_bytes(raw[i + 5:i + 7], "big")
return int(width), int(height)
return int.from_bytes(raw[i + 5:i + 7], "big"), int.from_bytes(raw[i + 3:i + 5], "big")
i += segment_len
return None
@@ -62,15 +53,10 @@ class UIElement:
pid: int = 0 # owning process PID
window_id: int = 0 # SkyLight / CG window ID
attributes: Dict[str, Any] = field(default_factory=dict)
# Opaque per-snapshot handle from cua-driver. Passed alongside `index` for explicit
# stale-detection: a stale token errors instead of silently re-resolving to a
# different element. None for older drivers that lack the field.
# Opaque per-snapshot handle from cua-driver, passed alongside `index` for explicit stale-detection: a
# stale token errors instead of silently re-resolving to a different element. None on older drivers.
element_token: Optional[str] = None
def center(self) -> Tuple[int, int]:
x, y, w, h = self.bounds
return x + w // 2, y + h // 2
@dataclass
class CaptureResult:
@@ -86,30 +72,27 @@ class CaptureResult:
app: str = "" # target app/window the elements were captured for
window_title: str = ""
png_bytes_len: int = 0 # raw bytes sent to Anthropic, for token estimation
# MIME type of `png_b64` when the backend supplied it (cua-driver-rs emits `mimeType`
# on every image part). None → consumers fall back to base64-prefix sniffing (older drivers).
# MIME type of `png_b64` when the backend supplied it (cua-driver-rs emits `mimeType` on every image
# part). None → consumers fall back to base64-prefix sniffing (older drivers).
image_mime_type: Optional[str] = None
# Guidance appended to the summary by capture lanes that intentionally return no
# elements (e.g. full-screen composited grabs) to point the model at an interactive lane.
# Guidance appended to the summary by capture lanes that intentionally return no elements (e.g.
# full-screen composited grabs) to point the model at an interactive lane.
note: str = ""
@dataclass
class ActionResult:
"""Result of any action (click / type / scroll / drag / key / wait).
``ok`` is tool/transport success only — NOT the semantic verdict; read ``effect`` /
``escalation`` (cua-driver's structured verdict) to pick the next rung of the
verify → escalate ladder. Structured fields are optional and additive: an older
driver that omits ``structuredContent`` leaves them ``None``, behavior unchanged.
"""
"""Result of any action (click / type / scroll / drag / key / wait). ``ok`` is
tool/transport success only — NOT the semantic verdict; read ``effect`` / ``escalation``
(cua-driver's structured verdict) to pick the next rung of the verify → escalate ladder.
Structured fields are optional and additive: an older driver that omits
``structuredContent`` leaves them ``None``, behavior unchanged."""
ok: bool
action: str
message: str = "" # human-readable summary
capture: Optional[CaptureResult] = None # trailing screenshot, when requested / always-on
meta: Dict[str, Any] = field(default_factory=dict) # debugging / telemetry extras
# ── cua-driver structured verdict (additive; None on old drivers) ──
verified: Optional[bool] = None # AX read-back: True confirmed, False unconfirmed, None n/a
effect: Optional[str] = None # "confirmed" | "unverifiable" | "suspected_noop"
# {"recommended": "px"|"foreground"|"page", "reason": str} — only when driver recommends climbing
@@ -121,12 +104,11 @@ class ActionResult:
class ComputerUseBackend(ABC):
"""Lifecycle: `start()` before first use, `stop()` at shutdown.
Pointer/keyboard actions take ``delivery_mode`` (background (default) | foreground)
and ``bring_to_front``; ``button`` is left | right | middle; ``modifiers`` a list of
key names. ``element`` args are 1-based SOM indices from a prior capture.
"""
"""Lifecycle: `start()` before first use, `stop()` at shutdown. Pointer/keyboard actions
take ``delivery_mode`` (background (default) | foreground) and ``bring_to_front``;
``button`` is left | right | middle; ``modifiers`` a list of key names. ``element`` args
are 1-based SOM indices from a prior capture. `direction` is up | down | left | right and
`amount` is wheel ticks; `keys` is a combo such as 'cmd+s', 'ctrl+alt+t', 'return'."""
@abstractmethod
def start(self) -> None: ...
@@ -135,15 +117,12 @@ class ComputerUseBackend(ABC):
def stop(self) -> None: ...
@abstractmethod
def is_available(self) -> bool:
"""True if the backend can be used on this host right now (check_fn gating, setup wizard)."""
def is_available(self) -> bool: ... # usable on this host right now (check_fn gating, setup wizard)
# ── Capture ─────────────────────────────────────────────────────
@abstractmethod
def capture(self, mode: str = "som", app: Optional[str] = None, pid: Optional[int] = None,
window_id: Optional[int] = None) -> CaptureResult: ...
# ── Pointer actions ─────────────────────────────────────────────
@abstractmethod
def click(self, *, element: Optional[int] = None, x: Optional[int] = None, y: Optional[int] = None,
button: str = "left", click_count: int = 1, modifiers: Optional[List[str]] = None,
@@ -158,39 +137,29 @@ class ComputerUseBackend(ABC):
@abstractmethod
def scroll(self, *, direction: str, amount: int = 3, element: Optional[int] = None,
x: Optional[int] = None, y: Optional[int] = None, modifiers: Optional[List[str]] = None,
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult:
"""`direction` is up | down | left | right; `amount` is wheel ticks."""
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ...
# ── Keyboard ────────────────────────────────────────────────────
@abstractmethod
def type_text(self, text: str, *, delivery_mode: Optional[str] = None,
bring_to_front: bool = False) -> ActionResult: ...
@abstractmethod
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'."""
def key(self, keys: str, *, delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ...
# ── Introspection ───────────────────────────────────────────────
@abstractmethod
def list_apps(self) -> List[Dict[str, Any]]:
"""Return running apps with bundle IDs, PIDs, window counts."""
def list_apps(self) -> List[Dict[str, Any]]: ... # running apps with bundle IDs, PIDs, window counts
def list_windows(self) -> List[Dict[str, Any]]:
"""Visible native windows with PID and window identifiers. Optional compatibility
hook: backends that predate window discovery stay instantiable and report none."""
"""Visible native windows with PID and window identifiers. Optional compatibility hook: backends that
predate window discovery stay instantiable and report none."""
return []
@abstractmethod
def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
"""Route input to `app` (by name or bundle ID). Default: focus without raise."""
def focus_app(self, app: str, raise_window: bool = False) -> ActionResult: ... # route input to `app` (name / bundle ID)
@abstractmethod
def set_value(self, value: str, element: Optional[int] = None) -> ActionResult:
"""Set a native value on an element (e.g. AXPopUpButton selection)."""
def set_value(self, value: str, element: Optional[int] = None) -> ActionResult: ... # e.g. AXPopUpButton selection
def wait(self, seconds: float) -> ActionResult:
"""Default implementation: time.sleep."""
import time
def wait(self, seconds: float) -> ActionResult: # default implementation
time.sleep(max(0.0, min(seconds, 30.0)))
return ActionResult(ok=True, action="wait", message=f"waited {seconds:.2f}s")

567
tools/computer_use/cua_backend_capture.py
View File

@@ -1,10 +1,6 @@
"""Capture side of the cua-driver backend: window discovery, capture-target
selection and the capture()/list_windows()/list_apps()/focus_app() methods
(mixed into ``CuaDriverBackend``).
Logger name is kept as ``tools.computer_use.cua_backend`` so log-based tests
and operators see one backend logger.
"""
"""Capture side of the cua-driver backend: window discovery, capture-target selection and capture()/list_windows()/
list_apps()/focus_app() (mixed into ``CuaDriverBackend``). Logger name stays ``tools.computer_use.cua_backend`` so
log-based tests and operators see one backend logger."""
from __future__ import annotations
@@ -14,68 +10,41 @@ import os
import re
import subprocess
import sys
from typing import Any, Dict, List, Optional, Tuple
from contextlib import contextmanager
from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple
from tools.computer_use.backend import ActionResult, CaptureResult, UIElement
from tools.computer_use.cua_backend_input import _BTF_UNSUPPORTED_MSG
from tools.computer_use.cua_backend_parse import (
_apps_from_windows,
_image_dimensions_from_bytes,
_image_from_tool_result,
_ingest_windows,
_is_placeholder_id,
_is_real_app_window,
_parse_elements_from_structured,
_parse_elements_from_tree,
_parse_xprop_net_active_window,
_positive_int,
_split_tree_text,
_windows_from_tool_result,
_z_index_uninformative,
_apps_from_windows, _image_dimensions_from_bytes, _image_from_tool_result, _ingest_windows, _is_placeholder_id,
_is_real_app_window, _parse_elements_from_structured, _parse_elements_from_tree, _parse_xprop_net_active_window,
_positive_int, _split_tree_text, _windows_from_tool_result, _z_index_uninformative,
)
logger = logging.getLogger("tools.computer_use.cua_backend")
# Whole-screen intents: app="screen"/... -> composited `get_desktop_state`
# (pixels only); app="desktop" -> the OS shell window via list_windows, WITH
# interactable elements (desktop icons, taskbar).
# Whole-screen intents: app="screen"/... -> composited `get_desktop_state` (pixels only); app="desktop" -> the OS
# shell window via list_windows, WITH interactable elements (icons, taskbar).
_FULL_SCREEN_SENTINELS = {"screen", "fullscreen", "full screen", "all"}
_DESKTOP_SHELL_SENTINELS = {"desktop"}
# Shell window identifiers (substring of app_name + title, case-insensitive).
# Windows: Progman/WorkerW = desktop, Shell_TrayWnd = taskbar; macOS: Finder/Dock.
_DESKTOP_WINDOW_NAMES = (
"progman", "workerw", "program manager", "shell_traywnd", "taskbar",
"finder", "desktop", "dock",
)
# Backdrop subset preferred over the taskbar when both are present.
# Shell window identifiers (substring of app_name + title, case-insensitive). Windows: Progman/WorkerW =
# desktop, Shell_TrayWnd = taskbar; macOS: Finder/Dock. The backdrop subset is preferred over the taskbar.
_DESKTOP_WINDOW_NAMES = ("progman", "workerw", "program manager", "shell_traywnd", "taskbar", "finder", "desktop", "dock")
_DESKTOP_BACKDROP_NAMES = ("progman", "workerw", "program manager", "finder", "desktop")
_WINDOW_TITLE_RE = re.compile(r'AXWindow\s+"([^"]+)"')
_LEGACY_APP_LINE_RE = re.compile(r'(.+?)\s+\(pid\s+(\d+)\)')
_NO_DESKTOP_WINDOW_MSG = (
"<no desktop/shell window found for app={app!r}; cua-driver captures one "
"window at a time and exposes no whole-virtual-desktop or per-monitor "
"capture. Call list_apps / capture(app='<AppName>') to target a specific "
"window instead. On Windows the taskbar is 'Shell_TrayWnd' and the desktop "
"is 'Progman'.>"
)
_NO_APP_MATCH_MSG = (
"<no on-screen window matched app={app!r}; call list_apps to see available "
"app names or bundle IDs (macOS reports localized names, e.g. '計算機' "
"instead of 'Calculator'; some Linux/Qt apps only resolve via list_apps "
"metadata)>"
)
_NO_DESKTOP_IMAGE_MSG = (
"<get_desktop_state returned no image; the driver may predate the desktop "
"capture lane — try capture(app='<AppName>') for a specific window>"
)
_FULL_SCREEN_NOTE = (
"full-screen capture has no interactable elements; to act on what you see, "
"call capture(app='<AppName>') for that app's clickable element list, or "
"capture(app='desktop') for the desktop shell (wallpaper icons / taskbar) "
"with elements"
)
_NO_DESKTOP_WINDOW_MSG = ("<no desktop/shell window found for app={app!r}; cua-driver captures one window at a time "
"and exposes no whole-virtual-desktop or per-monitor capture. Call list_apps / "
"capture(app='<AppName>') to target a specific window instead. On Windows the taskbar is "
"'Shell_TrayWnd' and the desktop is 'Progman'.>")
_NO_APP_MATCH_MSG = ("<no on-screen window matched app={app!r}; call list_apps to see available app names or bundle "
"IDs (macOS reports localized names, e.g. '計算機' instead of 'Calculator'; some Linux/Qt apps "
"only resolve via list_apps metadata)>")
_NO_DESKTOP_IMAGE_MSG = ("<get_desktop_state returned no image; the driver may predate the desktop capture lane — "
"try capture(app='<AppName>') for a specific window>")
_FULL_SCREEN_NOTE = ("full-screen capture has no interactable elements; to act on what you see, call "
"capture(app='<AppName>') for that app's clickable element list, or capture(app='desktop') for "
"the desktop shell (wallpaper icons / taskbar) with elements")
def _linux_x11_active_window_id() -> Optional[int]:
"""Best-effort read of ``_NET_ACTIVE_WINDOW`` via xprop. Never raises."""
@@ -90,346 +59,256 @@ def _linux_x11_active_window_id() -> Optional[int]:
def _select_capture_target(windows: List[Dict[str, Any]], *, app_requested: bool,
exact_target: bool = False) -> Dict[str, Any]:
"""Best window from z-sorted (frontmost-first) list_windows output.
Unqualified default captures on Linux (no app filter, no exact target) skip
desktop/shell helper windows first — targetable but capture as empty — and
when every remaining candidate shares one ``z_index`` (the common X11 case)
``_NET_ACTIVE_WINDOW`` beats list order. Exact-target captures never pay
for the ``xprop`` probe.
"""
"""Best window from z-sorted (frontmost-first) list_windows output. Unqualified default captures on
Linux (no app filter, no exact target) skip desktop/shell helper windows first — targetable but capture
as empty — and when every remaining candidate shares one ``z_index`` (the common X11 case)
``_NET_ACTIVE_WINDOW`` beats list order. Exact-target captures never pay for the ``xprop`` probe."""
pool = [w for w in windows if not w["off_screen"]]
if not exact_target and not app_requested and sys.platform == "linux":
pool = [w for w in pool if _is_real_app_window(w)] or pool
if pool and _z_index_uninformative(pool):
active_id = _linux_x11_active_window_id()
if active_id is not None:
for w in pool:
if w.get("window_id") == active_id:
return w
if active_id is not None and (hit := [w for w in pool if w.get("window_id") == active_id]):
return hit[0]
return pool[0] if pool else windows[0]
def _sorted_windows(out: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Normalised windows from a list_windows result, ``z_index`` DESCENDING
(frontmost at index 0 — the default target for capture()/focus_app())."""
windows = _ingest_windows(_windows_from_tool_result(out))
windows.sort(key=lambda w: w["z_index"], reverse=True)
return windows
"""Normalised list_windows rows, ``z_index`` DESCENDING (frontmost first = default capture/focus target)."""
return sorted(_ingest_windows(_windows_from_tool_result(out)), key=lambda w: w["z_index"], reverse=True)
def _tree_and_title(out: Dict[str, Any]) -> Tuple[str, str]:
"""``(tree_markdown, window_title)`` from a get_window_state result."""
data = out.get("data")
_, tree = _split_tree_text(data if isinstance(data, str) else "")
match = _WINDOW_TITLE_RE.search(tree)
return tree, (match.group(1) if match else "")
tree = _split_tree_text(data if isinstance((data := out.get("data")), str) else "")[1]
return tree, (match.group(1) if (match := _WINDOW_TITLE_RE.search(tree)) else "")
def _gws_is_empty(out: Dict[str, Any]) -> bool:
"""True when a get_window_state result carries neither a screenshot nor a
parseable tree. Modern drivers put the payload in structuredContent with
no markdown tree — that is NOT empty."""
if out.get("images"):
return False
"""True when a get_window_state result carries neither a screenshot nor a parseable tree. Modern
drivers put the payload in structuredContent with no markdown tree — that is NOT empty."""
sc_ = out.get("structuredContent") or {}
if sc_.get("elements") or sc_.get("screenshot_png_b64"):
return False
tree, _ = _tree_and_title(out)
return not tree.strip()
return not (out.get("images") or sc_.get("elements") or sc_.get("screenshot_png_b64")
or _tree_and_title(out)[0].strip())
def _png_metrics(png_b64: str, width: int, height: int) -> Tuple[int, int, int]:
"""Return ``(png_bytes_len, width, height)``, replacing the given size with
the sniffed one when the bytes decode to a readable PNG/JPEG header."""
"""``(png_bytes_len, width, height)``; the sniffed size wins when the bytes carry a readable PNG/JPEG header."""
try:
raw = base64.b64decode(png_b64, validate=False)
png_bytes_len = len(raw)
detected_width, detected_height = _image_dimensions_from_bytes(raw)
if detected_width and detected_height:
width, height = detected_width, detected_height
except Exception:
png_bytes_len = len(png_b64) * 3 // 4
return png_bytes_len, width, height
return len(png_b64) * 3 // 4, width, height
detected_width, detected_height = _image_dimensions_from_bytes(raw)
return len(raw), *((detected_width, detected_height) if detected_width and detected_height else (width, height))
def _is_desktop_window(w: Dict[str, Any], names: Tuple[str, ...] = _DESKTOP_WINDOW_NAMES) -> bool:
haystack = f"{w.get('app_name', '')} {w.get('title', '')}".lower()
return any(name in haystack for name in names)
def _app_aliases(raw_app: Dict[str, Any]) -> set:
return {
value.strip().lower()
for key in ("bundle_id", "bundleId", "name", "app_name", "display_name")
if isinstance((value := raw_app.get(key)), str) and value.strip()
}
return any(name in f"{w.get('app_name', '')} {w.get('title', '')}".lower() for name in names)
class _CaptureMixin:
"""capture()/list_windows()/list_apps()/focus_app() and their window-discovery helpers."""
# ── Failure plumbing ───────────────────────────────────────────
def _failed_capture(self, mode: str, message: str = "") -> CaptureResult:
"""Return an empty capture after disarming any prior target context."""
self._clear_active_target()
return CaptureResult(mode=mode, width=0, height=0, png_b64=None, elements=[],
app="", window_title=message, png_bytes_len=0)
def _call_capture_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""Call a capture-stage tool and disarm state on transport or logical failure."""
@contextmanager
def _disarming(self) -> Iterator[None]:
"""Forget the sticky target when the wrapped capture-stage step raises."""
try:
out = self._session.call_tool(name, args)
yield
except Exception:
self._clear_active_target()
raise
if out.get("isError") is True:
message = out.get("data")
self._clear_active_target()
raise RuntimeError(f"cua-driver {name} failed"
+ (f": {message}" if isinstance(message, str) and message else ""))
def _failed_capture(self, mode: str, message: str = "") -> CaptureResult:
"""Return an empty capture after disarming any prior target context."""
self._clear_active_target()
return CaptureResult(mode=mode, width=0, height=0, window_title=message)
def _call_capture_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""Call a capture-stage tool and disarm state on transport or logical failure."""
with self._disarming():
out = self._session.call_tool(name, args)
if out.get("isError") is True:
message = out.get("data")
raise RuntimeError(f"cua-driver {name} failed"
+ (f": {message}" if isinstance(message, str) and message else ""))
return out
def _cli_refetch(self, name: str, args: Dict[str, Any], timeout: float,
what: str) -> Optional[Dict[str, Any]]:
"""One-shot call over the CLI transport (different daemon socket) after
MCP came back empty/imageless without raising. None on failure."""
def _cli_refetch(self, name: str, args: Dict[str, Any], timeout: float, what: str,
warning: str, *warning_args: Any) -> Optional[Dict[str, Any]]:
"""MCP came back empty/imageless without raising: log *warning*, then a one-shot call over the CLI
transport (different daemon socket). None on failure."""
logger.warning(warning, *warning_args)
try:
cli_out = self._session._call_tool_via_cli(name, args, timeout)
except Exception as cli_exc:
logger.error("cua-driver CLI re-fetch for %s failed: %s", what, cli_exc)
return None
if cli_out.get("isError") is True:
if name == "list_windows":
logger.error("cua-driver CLI re-fetch for list_windows returned an error")
self._clear_active_target()
return None
return cli_out
if cli_out.get("isError") is not True:
return cli_out
if name == "list_windows":
logger.error("cua-driver CLI re-fetch for list_windows returned an error")
self._clear_active_target()
return None
# ── Window discovery ───────────────────────────────────────────
def _list_windows_args(self) -> Dict[str, Any]:
return {"on_screen_only": True, "session": self._session_id}
def _fetch_or_refetch(self, name: str, args: Dict[str, Any], timeout: float, what: str,
empty: Callable[[Dict[str, Any]], bool], warning: str, *warning_args: Any) -> Dict[str, Any]:
"""``_call_capture_tool`` whose result, when *empty*, is replaced by a non-empty CLI re-fetch (the
MCP result stands when the CLI fails too)."""
out = self._call_capture_tool(name, args)
cli_out = self._cli_refetch(name, args, timeout, what, warning, *warning_args) if empty(out) else None
return cli_out if cli_out is not None and not empty(cli_out) else out
def _load_windows(self) -> List[Dict[str, Any]]:
"""Visible windows frontmost-first, re-fetching over the CLI transport
when MCP returns nothing."""
windows = _sorted_windows(self._call_capture_tool("list_windows", self._list_windows_args()))
if windows:
return windows
logger.warning("cua-driver list_windows returned no windows over MCP; re-fetching via CLI transport")
cli_out = self._cli_refetch("list_windows", self._list_windows_args(), 20.0, "list_windows")
return _sorted_windows(cli_out) if cli_out is not None else []
def _load_windows_or_disarm(self) -> List[Dict[str, Any]]:
"""``_load_windows`` that forgets the sticky target when discovery raises."""
try:
return self._load_windows()
except Exception:
self._clear_active_target()
raise
"""Visible windows frontmost-first, re-fetching over the CLI transport when MCP returns nothing."""
return _sorted_windows(self._fetch_or_refetch(
"list_windows", {"on_screen_only": True, "session": self._session_id}, 20.0, "list_windows",
lambda out: not _sorted_windows(out),
"cua-driver list_windows returned no windows over MCP; re-fetching via CLI transport"))
def _match_windows_for_app(self, windows: List[Dict[str, Any]], app: str) -> List[Dict[str, Any]]:
"""Resolve ``app=``: exact window names, then exact list_apps aliases
(Linux ``list_windows`` can omit the app name that ``list_apps`` keeps),
then substrings — querying ``Code`` must not silently select
``Visual Studio Code`` because it is frontmost."""
"""Resolve ``app=``: exact window names, then exact list_apps aliases (Linux ``list_windows`` can
omit the app name that ``list_apps`` keeps), then substrings — querying ``Code`` must not silently
select ``Visual Studio Code`` because it is frontmost."""
app_lower = app.strip().lower()
if not app_lower:
return []
def _name(w: Dict[str, Any]) -> str:
return str(w.get("app_name", "")).lower()
direct_exact = [w for w in windows if app_lower == _name(w).strip()]
if direct_exact:
_name = lambda w: str(w.get("app_name", "")).lower() # noqa: E731
direct_exact = [w for w in windows if app_lower and app_lower == _name(w).strip()]
if not app_lower or direct_exact:
return direct_exact
try:
running_apps = self.list_apps()
except Exception as exc:
# A title can still be the only usable identity on X11 when app
# enumeration is unavailable, so keep the title fallback below.
# A title can still be the only usable identity on X11 when app enumeration is unavailable,
# so keep the title fallback below.
logger.debug("computer_use list_apps fallback failed for %r: %s", app, exc)
running_apps = []
exact_pids: set[int] = set()
partial_pids: set[int] = set()
exact_pids, partial_pids = set(), set()
for raw_app in running_apps:
pid = _positive_int(raw_app.get("pid")) if isinstance(raw_app, dict) else None
if pid is None or raw_app.get("running") is False:
continue
aliases = _app_aliases(raw_app)
if app_lower in aliases:
exact_pids.add(pid)
elif any(app_lower in alias for alias in aliases):
partial_pids.add(pid)
for matched in ([w for w in windows if w.get("pid") in exact_pids],
[w for w in windows if app_lower in _name(w)],
[w for w in windows if w.get("pid") in partial_pids]):
if matched:
return matched
# Some X11 backends expose a title but no app name. Restrict this final
# fallback to nameless rows so a localized app name is not overridden
# merely because its title happens to be in the caller's language.
return [w for w in windows
if not _name(w).strip() and app_lower in str(w.get("title", "")).lower()]
if pid is not None and raw_app.get("running") is not False:
aliases = {value.strip().lower() for key in ("bundle_id", "bundleId", "name", "app_name", "display_name")
if isinstance((value := raw_app.get(key)), str) and value.strip()}
if app_lower in aliases:
exact_pids.add(pid)
elif any(app_lower in alias for alias in aliases):
partial_pids.add(pid)
# Some X11 backends expose a title but no app name. Restrict the final fallback to nameless rows so
# a localized app name is not overridden merely because its title happens to be in the caller's language.
tiers = ([w for w in windows if w.get("pid") in exact_pids],
[w for w in windows if app_lower in _name(w)],
[w for w in windows if w.get("pid") in partial_pids],
[w for w in windows if not _name(w).strip() and app_lower in str(w.get("title", "")).lower()])
return next((matched for matched in tiers if matched), [])
def _resolve_capture_windows(self, mode: str, app: Optional[str], pid: Optional[int],
window_id: Optional[int]) -> "List[Dict[str, Any]] | CaptureResult":
"""Candidate windows for capture(), or a failed CaptureResult."""
if pid is not None or window_id is not None:
# An exact pid/window pair is both the stable capture_after target
# and the escape hatch when discovery is unavailable on X11.
# An exact pid/window pair is both the stable capture_after target and the escape hatch when
# discovery is unavailable on X11.
if pid is None or window_id is None:
return self._failed_capture(mode, "<capture targeting requires both pid and window_id>")
target_pid, target_window_id = _positive_int(pid), _positive_int(window_id)
if target_pid is None or target_window_id is None:
return self._failed_capture(
mode, "<capture targeting requires positive integer pid and window_id>",
)
return [{"app_name": app or "", "pid": target_pid, "window_id": target_window_id,
"off_screen": False, "title": "", "z_index": 0}]
windows = self._load_windows_or_disarm()
if (target_pid := _positive_int(pid)) is None or (target_window_id := _positive_int(window_id)) is None:
return self._failed_capture(mode, "<capture targeting requires positive integer pid and window_id>")
return [{"app_name": app or "", "pid": target_pid, "window_id": target_window_id, "off_screen": False,
"title": "", "z_index": 0}]
with self._disarming():
windows = self._load_windows()
if not windows:
# Diagnose instead of a bare 0x0: the dominant real-world cause on
# Linux is a locked desktop session.
# Diagnose instead of a bare 0x0: the dominant real-world cause on Linux is a locked desktop session.
from tools.computer_use import cua_backend as _cb
return self._failed_capture(mode, _cb._empty_discovery_reason())
if not app:
return windows
if app.strip().lower() in _DESKTOP_SHELL_SENTINELS:
# Desktop-shell request: the OS shell window WITH its interactable
# elements (desktop icons), so "click the taskbar" works. Prefer the
# backdrop (Progman/WorkerW/Finder) over the taskbar so the capture
# shows the full desktop rather than the task strip.
desktop = [w for w in windows if _is_desktop_window(w)]
if not desktop:
return self._failed_capture(mode, _NO_DESKTOP_WINDOW_MSG.format(app=app))
return sorted(desktop, key=lambda w: 0 if _is_desktop_window(w, _DESKTOP_BACKDROP_NAMES) else 1)
# Desktop-shell request: the OS shell window WITH its interactable elements (desktop icons), so
# "click the taskbar" works. Prefer the backdrop (Progman/WorkerW/Finder) over the taskbar so the
# capture shows the full desktop rather than the task strip.
desktop = sorted((w for w in windows if _is_desktop_window(w)),
key=lambda w: not _is_desktop_window(w, _DESKTOP_BACKDROP_NAMES))
return desktop or self._failed_capture(mode, _NO_DESKTOP_WINDOW_MSG.format(app=app))
# When the filter matches nothing, say so instead of silently capturing the frontmost window — on
# macOS list_windows returns the localized app name (e.g. "計算機"), so `app="Calculator"` legitimately misses.
return self._match_windows_for_app(windows, app) or self._failed_capture(mode, _NO_APP_MATCH_MSG.format(app=app))
# When the filter matches nothing, say so instead of silently capturing
# the frontmost window — on macOS list_windows returns the localized
# app name (e.g. "計算機"), so `app="Calculator"` legitimately misses.
return (self._match_windows_for_app(windows, app)
or self._failed_capture(mode, _NO_APP_MATCH_MSG.format(app=app)))
# ── Capture ────────────────────────────────────────────────────
def _gws_args(self) -> Dict[str, Any]:
return {"pid": self._active_pid, "window_id": self._active_window_id, "session": self._session_id}
def _capture_vision(self) -> Tuple[Optional[str], Optional[str], str]:
"""Pixels only, no elements: ``(png_b64, mime, window_title)``.
Drivers advertising the cheaper standalone ``screenshot`` tool use it;
current drivers folded PNG capture into ``get_window_state`` (tree
DISCARDED here). Before discovery ran we still try ``screenshot`` first
and fall back, so the path self-heals on any driver version.
"""
png_b64: Optional[str] = None
image_mime_type: Optional[str] = None
window_title = ""
def _capture_vision(self) -> Tuple[Optional[str], Optional[str], List[UIElement], str]:
"""Pixels only, ``elements`` always empty: ``(png_b64, mime, [], window_title)``. Drivers advertising the
cheaper standalone ``screenshot`` tool use it; current drivers folded PNG capture into ``get_window_state``
(tree DISCARDED here). Before discovery ran we still try ``screenshot`` first and fall back, so the path
self-heals on any driver version."""
png_b64, image_mime_type, window_title = None, None, ""
if self._session._has_tool("screenshot") or not self._session.capabilities_discovered:
sc_out = self._call_capture_tool("screenshot", {
"window_id": self._active_window_id, "format": "jpeg", "quality": 85,
"session": self._session_id,
})
png_b64, image_mime_type = _image_from_tool_result(sc_out)
png_b64, image_mime_type = _image_from_tool_result(self._call_capture_tool("screenshot", {
"window_id": self._active_window_id, "format": "jpeg", "quality": 85, "session": self._session_id}))
if not png_b64:
# "Unknown tool: screenshot" or an empty image part -> get_window_state.
# "Unknown tool: screenshot" or an empty image part -> get_window_state. The title is cheap
# and useful; `elements` stays empty by contract.
gws_out = self._call_capture_tool("get_window_state", self._gws_args())
png_b64, image_mime_type = _image_from_tool_result(gws_out)
# The title is cheap and useful; `elements` stays empty by contract.
_, window_title = _tree_and_title(gws_out)
(png_b64, image_mime_type), (_, window_title) = _image_from_tool_result(gws_out), _tree_and_title(gws_out)
if not png_b64:
logger.warning("cua-driver vision capture returned no image over MCP (window_id=%s); "
"re-fetching via CLI transport", self._active_window_id)
cli_out = self._cli_refetch("get_window_state", self._gws_args(), 30.0, "vision screenshot")
if cli_out is not None and cli_out.get("images"):
cli_out = self._cli_refetch(
"get_window_state", self._gws_args(), 30.0, "vision screenshot",
"cua-driver vision capture returned no image over MCP (window_id=%s); re-fetching via CLI transport",
self._active_window_id) or {}
if cli_out.get("images"):
png_b64, image_mime_type = cli_out["images"][0], "image/png"
return png_b64, image_mime_type, window_title
return png_b64, image_mime_type, [], window_title
def _capture_window_state(self) -> Tuple[Optional[str], Optional[str], List[UIElement], str]:
"""AX tree + screenshot. Returns ``(png_b64, mime, elements, window_title)``."""
gws_out = self._call_capture_tool("get_window_state", self._gws_args())
# A flaky bridge can return a degenerate result (no screenshot AND no
# parseable tree) WITHOUT raising — a silent 0x0 to the model. Distinct
# from the EAGAIN path handled in call_tool: here MCP "succeeded".
if _gws_is_empty(gws_out):
logger.warning("cua-driver get_window_state returned an empty result over MCP "
"(pid=%s window_id=%s); re-fetching via CLI transport",
self._active_pid, self._active_window_id)
cli_out = self._cli_refetch("get_window_state", self._gws_args(), 30.0, "get_window_state")
if cli_out is not None and not _gws_is_empty(cli_out):
gws_out = cli_out
# A flaky bridge can return a degenerate result (no screenshot AND no parseable tree) WITHOUT raising
# — a silent 0x0 to the model. Distinct from the EAGAIN path handled in call_tool: here MCP "succeeded".
gws_out = self._fetch_or_refetch(
"get_window_state", self._gws_args(), 30.0, "get_window_state", _gws_is_empty,
"cua-driver get_window_state returned an empty result over MCP (pid=%s window_id=%s); re-fetching via CLI "
"transport", self._active_pid, self._active_window_id)
tree, window_title = _tree_and_title(gws_out)
# Prefer the canonical structuredContent.elements (real frames); the
# markdown regex fallback yields (0,0,0,0) bounds.
# Prefer the canonical structuredContent.elements (real frames); the markdown regex fallback yields
# (0,0,0,0) bounds.
sc_elements = (gws_out.get("structuredContent") or {}).get("elements")
if isinstance(sc_elements, list) and sc_elements:
elements = _parse_elements_from_structured(sc_elements)
else:
elements = _parse_elements_from_tree(tree) if tree else []
# Tokens are tied to this snapshot: overwrite the whole map (and clear
# it when the new capture carries none).
elements = (_parse_elements_from_structured(sc_elements) if isinstance(sc_elements, list) and sc_elements
else _parse_elements_from_tree(tree) if tree else [])
# Tokens are tied to this snapshot: overwrite the whole map (and clear it when the new capture carries none).
self._snapshot_tokens = {e.index: e.element_token for e in elements if e.element_token}
png_b64, image_mime_type = _image_from_tool_result(gws_out)
return png_b64, image_mime_type, elements, window_title
return *_image_from_tool_result(gws_out), elements, window_title
def capture(self, mode: str = "som", app: Optional[str] = None, pid: Optional[int] = None,
window_id: Optional[int] = None) -> CaptureResult:
"""Capture the frontmost on-screen window or an exact known target:
`list_windows` + `get_window_state` (ax/som) or `screenshot` (vision).
Only the structured ``structuredContent.windows`` shape is supported."""
# Schema-filler ids (models zero-fill optional properties) must not read
# as a targeting request.
pid = None if _is_placeholder_id(pid) else pid
window_id = None if _is_placeholder_id(window_id) else window_id
"""Capture the frontmost on-screen window or an exact known target: `list_windows` +
`get_window_state` (ax/som) or `screenshot` (vision). Only the structured
``structuredContent.windows`` shape is supported."""
# Schema-filler ids (models zero-fill optional properties) must not read as a targeting request.
pid, window_id = [None if _is_placeholder_id(v) else v for v in (pid, window_id)]
exact_target = pid is not None or window_id is not None
# Full-screen lane bypasses enumeration entirely (also keeps
# screenshots working when Windows UIA enumeration hangs).
# app='desktop' deliberately does NOT take it: desktop icons stay clickable.
# Full-screen lane bypasses enumeration entirely (also keeps screenshots working when Windows UIA
# enumeration hangs). app='desktop' deliberately does NOT take it: desktop icons stay clickable.
if not exact_target and app and app.strip().lower() in _FULL_SCREEN_SENTINELS:
return self._capture_full_screen(mode)
windows = self._resolve_capture_windows(mode, app, pid, window_id)
if isinstance(windows, CaptureResult):
return windows
target = _select_capture_target(windows, app_requested=bool(app), exact_target=exact_target)
self._set_active_target(target)
self._set_active_target(target := _select_capture_target(windows, app_requested=bool(app), exact_target=exact_target))
app_name = target["app_name"]
# Record the resolved app so capture_after= follow-ups re-target the
# same app rather than falling back to the frontmost window.
# Record the resolved app so capture_after= follow-ups re-target the same app rather than falling back
# to the frontmost window.
if app or not self._last_app:
self._last_app = app_name or app or ""
elements: List[UIElement] = []
if mode == "vision":
png_b64, image_mime_type, window_title = self._capture_vision()
else:
png_b64, image_mime_type, elements, window_title = self._capture_window_state()
png_b64, image_mime_type, elements, window_title = (
self._capture_vision() if mode == "vision" else self._capture_window_state())
png_bytes_len, width, height = _png_metrics(png_b64, 0, 0) if png_b64 else (0, 0, 0)
return CaptureResult(mode=mode, width=width, height=height, png_b64=png_b64,
elements=elements, app=app_name, window_title=window_title,
png_bytes_len=png_bytes_len, image_mime_type=image_mime_type)
return CaptureResult(mode=mode, width=width, height=height, png_b64=png_b64, elements=elements, app=app_name,
window_title=window_title, png_bytes_len=png_bytes_len, image_mime_type=image_mime_type)
def _capture_full_screen(self, mode: str) -> CaptureResult:
"""Composited PrtScn-style grab via `get_desktop_state` (the shell window
would only show wallpaper + icons). Never enumerates, so it also works
when Windows UIA hangs. Pixels only — `elements` is empty and `note`
points the model at the interactive lanes. ``capture_scope`` is switched
to desktop for the call and restored afterwards."""
"""Composited PrtScn-style grab via `get_desktop_state` (the shell window would only show wallpaper + icons).
Never enumerates, so it also works when Windows UIA hangs. Pixels only — `elements` is empty and `note` points
the model at the interactive lanes. ``capture_scope`` is switched to desktop for the call and restored afterwards."""
self._clear_active_target()
previous_scope: Optional[str] = None
try:
cfg = self._session.call_tool("get_config", {"session": self._session_id}, timeout=10.0)
sc = cfg.get("structuredContent") or {}
if isinstance(sc, dict) and isinstance(sc.get("capture_scope"), str):
previous_scope = sc["capture_scope"]
sc = self._session.call_tool("get_config", {"session": self._session_id}, timeout=10.0).get("structuredContent")
previous_scope = sc["capture_scope"] if isinstance(sc, dict) and isinstance(sc.get("capture_scope"), str) else None
except Exception as e:
logger.debug("cua-driver get_config before full-screen capture failed: %s", e)
def _set_scope(value: str) -> None:
self._session.call_tool("set_config", {"key": "capture_scope", "value": value,
"session": self._session_id}, timeout=10.0)
_set_scope = lambda value: self._session.call_tool( # noqa: E731
"set_config", {"key": "capture_scope", "value": value, "session": self._session_id}, timeout=10.0)
try:
if previous_scope != "desktop":
_set_scope("desktop")
@@ -440,80 +319,54 @@ class _CaptureMixin:
_set_scope(previous_scope)
except Exception as e:
logger.debug("cua-driver restore capture_scope failed: %s", e)
png_b64, image_mime_type = _image_from_tool_result(out)
if not png_b64:
return self._failed_capture(mode, _NO_DESKTOP_IMAGE_MSG)
structured = out.get("structuredContent") or {}
png_bytes_len, width, height = _png_metrics(
png_b64,
int(structured.get("screenshot_width") or structured.get("screen_width") or 0),
int(structured.get("screenshot_height") or structured.get("screen_height") or 0),
)
return CaptureResult(
mode="vision", width=width, height=height, png_b64=png_b64, elements=[],
app="screen", window_title="Full screen (composited)",
png_bytes_len=png_bytes_len, image_mime_type=image_mime_type, note=_FULL_SCREEN_NOTE,
)
sc = out.get("structuredContent") or {}
png_bytes_len, width, height = _png_metrics(png_b64, int(sc.get("screenshot_width") or sc.get("screen_width") or 0),
int(sc.get("screenshot_height") or sc.get("screen_height") or 0))
return CaptureResult(mode="vision", width=width, height=height, png_b64=png_b64, app="screen",
window_title="Full screen (composited)", png_bytes_len=png_bytes_len,
image_mime_type=image_mime_type, note=_FULL_SCREEN_NOTE)
# ── Introspection ──────────────────────────────────────────────
def list_windows(self) -> List[Dict[str, Any]]:
return self._load_windows()
def list_apps(self) -> List[Dict[str, Any]]:
out = self._session.call_tool("list_apps", {"session": self._session_id})
structured = out.get("structuredContent")
data = out.get("data")
# structuredContent is canonical; empty lists fall through so a
# populated compatibility envelope (older drivers, CLI fallback) can
# still recover.
def _apps_in(container: Any) -> List[Any]:
apps = container.get("apps") if isinstance(container, dict) else None
return apps if isinstance(apps, list) else []
if _apps_in(structured):
return _apps_in(structured)
if isinstance(data, list) and data:
return data
for container in (data, out):
if _apps_in(container):
return _apps_in(container)
derived = _apps_from_windows(_windows_from_tool_result(out))
if derived:
return derived
structured, data = out.get("structuredContent"), out.get("data")
# structuredContent is canonical; empty lists fall through so a populated compatibility envelope
# (older drivers, CLI fallback) can still recover, then apps derived from the windows payload.
candidates = (lambda: structured.get("apps") if isinstance(structured, dict) else None,
lambda: data, lambda: data.get("apps") if isinstance(data, dict) else None,
lambda: out.get("apps"), lambda: _apps_from_windows(_windows_from_tool_result(out)))
for candidate in candidates:
if isinstance((apps := candidate()), list) and apps:
return apps
# Old text-only drivers retain a small, name/PID-only fallback.
if isinstance(data, str):
return [
{"name": m.group(1).strip(), "pid": int(m.group(2))}
for m in map(_LEGACY_APP_LINE_RE.search, data.splitlines()) if m
]
return []
return [{"name": m.group(1).strip(), "pid": int(m.group(2))}
for m in map(_LEGACY_APP_LINE_RE.search, data.splitlines()) if m] if isinstance(data, str) else []
def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
"""Pure window-selector (store pid/window_id so later input hits the
right process) — background automation never needs to raise a window.
``raise_window=True`` is explicit, separately approved, and uses the
standalone ``bring_to_front`` tool."""
matched = self._match_windows_for_app(self._load_windows_or_disarm(), app)
# No silent fallback to the frontmost window: that hides the real
# failure (often a localized macOS app-name mismatch).
"""Pure window-selector (store pid/window_id so later input hits the right process) — background
automation never needs to raise a window. ``raise_window=True`` is explicit, separately approved,
and uses the standalone ``bring_to_front`` tool."""
with self._disarming():
matched = self._match_windows_for_app(self._load_windows(), app)
# No silent fallback to the frontmost window: that hides the real failure (often a localized macOS
# app-name mismatch).
if not matched:
self._clear_active_target()
return ActionResult(ok=False, action="focus_app",
message=f"No on-screen window found for app '{app}'.")
target = matched[0]
self._set_active_target(target)
return ActionResult(ok=False, action="focus_app", message=f"No on-screen window found for app '{app}'.")
self._set_active_target(target := matched[0])
self._last_app = target["app_name"] or app # retained for back-compat diagnostics
if raise_window:
if not self._session._has_tool("bring_to_front"):
return ActionResult(ok=False, action="focus_app", code="bring_to_front_unsupported",
message=_BTF_UNSUPPORTED_MSG)
focused = self.bring_to_front(pid=self._active_pid, window_id=self._active_window_id)
if not focused.ok:
return focused
if not raise_window:
return ActionResult(ok=True, action="focus_app", message=f"Targeted {target['app_name']} (pid "
f"{self._active_pid}, window {self._active_window_id}) without raising window.")
if not self._session._has_tool("bring_to_front"):
return ActionResult(ok=False, action="focus_app", code="bring_to_front_unsupported", message=_BTF_UNSUPPORTED_MSG)
focused = self.bring_to_front(pid=self._active_pid, window_id=self._active_window_id)
if focused.ok:
focused.action = "focus_app"
focused.meta["target_selected"] = True
return focused
return ActionResult(ok=True, action="focus_app",
message=f"Targeted {target['app_name']} (pid {self._active_pid}, "
f"window {self._active_window_id}) without raising window.")
return focused

252
tools/computer_use/cua_backend_input.py
View File

@@ -1,21 +1,21 @@
"""Input side of the cua-driver backend: delivery-mode handling and the
pointer / keyboard / value-setter methods (mixed into ``CuaDriverBackend``).
"""
"""Input side of the cua-driver backend: delivery-mode handling and the pointer / keyboard /
value-setter methods (mixed into ``CuaDriverBackend``)."""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
from tools.computer_use.backend import ActionResult
from tools.computer_use.cua_backend_parse import _parse_key_combo
_NO_TARGET_MSG = "No active window — call capture() first."
_BTF_UNSUPPORTED_MSG = "The connected cua-driver does not advertise the standalone bring_to_front tool."
_FOREGROUND_UNSUPPORTED_MSG = (
"The connected cua-driver action schema does not accept delivery_mode, so "
"foreground delivery is unavailable. Use another verified rung without "
"assuming the reported package version describes the live schema."
)
_FOREGROUND_UNSUPPORTED_MSG = ("The connected cua-driver action schema does not accept delivery_mode, so foreground "
"delivery is unavailable. Use another verified rung without assuming the reported "
"package version describes the live schema.")
# (what, extra args) pointer addressing form; ``extra`` is None when the caller did not supply that form and
# may be a callable when computing it has side effects (capability probes) that must follow the refusal checks.
_Variant = Tuple[str, Union[None, Dict[str, Any], Callable[[], Dict[str, Any]]]]
def _refuse(action: str, message: str, **fields: Any) -> ActionResult:
return ActionResult(ok=False, action=action, message=message, **fields)
@@ -24,47 +24,44 @@ def _refuse(action: str, message: str, **fields: Any) -> ActionResult:
class _InputMixin:
"""Pointer / keyboard / value-setter actions against the sticky target."""
def _no_target(self, action: str, *, need_window: bool = False) -> Optional[ActionResult]:
def _target_args(self, action: str, *, need_window: bool = False) -> Tuple[Optional[ActionResult], Dict[str, Any]]:
"""``(refusal, base args)`` for an input action against the sticky target."""
if self._active_pid is None or (need_window and self._active_window_id is None):
return _refuse(action, _NO_TARGET_MSG)
return None
return _refuse(action, _NO_TARGET_MSG), {}
return None, {"pid": self._active_pid, **({"window_id": self._active_window_id} if need_window else {})}
def _need_window(self, action: str, what: str) -> Optional[ActionResult]:
"""Refusal when a targeted call has a pid but no window_id yet."""
if self._active_window_id is None:
return _refuse(action, f"No active window_id for {what}.")
return None
def _pointer_args(self, tool: str, args: Dict[str, Any], variants: Sequence[_Variant],
missing_msg: Optional[str]) -> Optional[ActionResult]:
"""Fill *args* from the first supplied addressing variant (element or coordinates) plus ``window_id``; refuse
when the target has a pid but no window_id yet. No variant -> refuse with *missing_msg* (None = bare window)."""
for what, extra in variants:
if extra is not None:
if self._active_window_id is None:
return _refuse(tool, f"No active window_id for {what}.")
args.update(extra() if callable(extra) else extra, window_id=self._active_window_id)
return None
return _refuse(tool, missing_msg) if missing_msg else None
# ── Input delivery ─────────────────────────────────────────────
def _apply_delivery(self, action: str, args: Dict[str, Any],
delivery_mode: Optional[str]) -> Optional[ActionResult]:
"""Attach delivery_mode to an input-action args dict.
Background is the default and needs no flag. Foreground is only sent
when the live action schema accepts it; on an older driver we refuse
with ``foreground_unsupported`` instead of silently downgrading to
background (which would land input where the model didn't expect).
Returns an ActionResult to short-circuit on refusal, or None to proceed.
"""
def _apply_delivery(self, action: str, args: Dict[str, Any], delivery_mode: Optional[str]) -> Optional[ActionResult]:
"""Attach delivery_mode to an input-action args dict. Background is the default and needs no flag.
Foreground is only sent when the live action schema accepts it; on an older driver we refuse with
``foreground_unsupported`` instead of silently downgrading to background (which would land input
where the model didn't expect)."""
if not delivery_mode or delivery_mode == "background":
return None
if delivery_mode != "foreground":
return _refuse(action, f"unknown delivery_mode {delivery_mode!r} — use background|foreground.",
code="bad_delivery_mode")
if not self._session.supports_input_property(action, "delivery_mode"):
return _refuse(action, _FOREGROUND_UNSUPPORTED_MSG,
code="foreground_unsupported", delivery_mode="foreground")
return _refuse(action, _FOREGROUND_UNSUPPORTED_MSG, code="foreground_unsupported", delivery_mode="foreground")
args["delivery_mode"] = "foreground"
return None
def _run_input_action(self, action: str, args: Dict[str, Any],
delivery_mode: Optional[str], bring_to_front: bool) -> ActionResult:
"""Apply one delivery rung, optionally focusing via its own tool.
``bring_to_front`` is never an input-action property: when requested,
the separately approved standalone focus action runs first, then the
original foreground input runs unchanged.
"""
def _run_input_action(self, action: str, args: Dict[str, Any], delivery_mode: Optional[str],
bring_to_front: bool) -> ActionResult:
"""Apply one delivery rung, optionally focusing via its own tool. ``bring_to_front`` is never an
input-action property: when requested, the separately approved standalone focus action runs first,
then the original foreground input runs unchanged."""
refusal = self._apply_delivery(action, args, delivery_mode)
if refusal is not None:
return refusal
@@ -73,8 +70,7 @@ class _InputMixin:
return _refuse(action, "bring_to_front requires delivery_mode='foreground'.",
code="bring_to_front_requires_foreground")
if not self._session._has_tool("bring_to_front"):
return _refuse(action, _BTF_UNSUPPORTED_MSG,
code="bring_to_front_unsupported", delivery_mode="foreground")
return _refuse(action, _BTF_UNSUPPORTED_MSG, code="bring_to_front_unsupported", delivery_mode="foreground")
if self._active_pid is None or self._active_window_id is None:
return _refuse(action, "Capture an exact target before requesting persistent foreground focus.",
code="bring_to_front_target_required", delivery_mode="foreground")
@@ -86,138 +82,82 @@ class _InputMixin:
result.meta["foreground_focus"] = {"invoked": True, "tool": "bring_to_front"}
return result
# ── Pointer ────────────────────────────────────────────────────
def click(
self,
*,
element: Optional[int] = None,
x: Optional[int] = None,
y: Optional[int] = None,
button: str = "left",
click_count: int = 1,
modifiers: Optional[List[str]] = None,
delivery_mode: Optional[str] = None,
bring_to_front: bool = False,
) -> ActionResult:
missing = self._no_target("click")
if missing is not None:
return missing
# Tool is chosen by click_count only; `button` goes through click's
# enum (the driver rejects unknown buttons). `right_click` /
# `middle_click` MCP tools are deprecated aliases and never invoked here.
def click(self, *, element: Optional[int] = None, x: Optional[int] = None, y: Optional[int] = None,
button: str = "left", click_count: int = 1, modifiers: Optional[List[str]] = None,
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult:
refusal, args = self._target_args("click")
if refusal is not None:
return refusal
# Tool is chosen by click_count only; `button` goes through click's enum (the driver rejects unknown
# buttons). `right_click` / `middle_click` MCP tools are deprecated aliases and never invoked here.
button_norm = (button or "left").lower()
if button_norm not in {"left", "right", "middle"}:
return _refuse("click", f"unknown button {button!r} — expected left, right, middle.")
tool = "double_click" if click_count == 2 else "click"
args: Dict[str, Any] = {"pid": self._active_pid, "button": button_norm}
if element is not None:
refusal = self._need_window(tool, "element_index click")
args["element_index"] = element
elif x is not None and y is not None:
refusal = self._need_window(tool, "coordinate click")
args.update(x=x, y=y)
else:
return _refuse(tool, "click requires element= or x/y.")
if refusal is not None:
return refusal
args["window_id"] = self._active_window_id
tool, args["button"] = ("double_click" if click_count == 2 else "click"), button_norm
refusal = self._pointer_args(tool, args, (
("element_index click", {"element_index": element} if element is not None else None),
("coordinate click", {"x": x, "y": y} if x is not None and y is not None else None),
), "click requires element= or x/y.")
if modifiers:
args["modifier"] = modifiers
return self._run_input_action(tool, args, delivery_mode, bring_to_front)
return refusal if refusal is not None else self._run_input_action(tool, args, delivery_mode, bring_to_front)
def drag(
self,
*,
from_element: Optional[int] = None,
to_element: Optional[int] = None,
from_xy: Optional[Tuple[int, int]] = None,
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:
missing = self._no_target("drag")
if missing is not None:
return missing
args: Dict[str, Any] = {"pid": self._active_pid}
if from_element is not None and to_element is not None:
refusal = self._need_window("drag", "element-based drag")
args.update(from_element=from_element, to_element=to_element)
elif from_xy is not None and to_xy is not None:
refusal = self._need_window("drag", "coordinate drag")
args.update(from_x=int(from_xy[0]), from_y=int(from_xy[1]),
to_x=int(to_xy[0]), to_y=int(to_xy[1]))
else:
return _refuse("drag", "drag requires from_element/to_element or from_coordinate/to_coordinate.")
def drag(self, *, from_element: Optional[int] = None, to_element: Optional[int] = None,
from_xy: Optional[Tuple[int, int]] = None, 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:
refusal, args = self._target_args("drag")
if refusal is None:
refusal = self._pointer_args("drag", args, (
("element-based drag", {"from_element": from_element, "to_element": to_element}
if from_element is not None and to_element is not None else None),
("coordinate drag", {"from_x": int(from_xy[0]), "from_y": int(from_xy[1]),
"to_x": int(to_xy[0]), "to_y": int(to_xy[1])}
if from_xy is not None and to_xy is not None else None),
), "drag requires from_element/to_element or from_coordinate/to_coordinate.")
return refusal if refusal is not None else self._run_input_action("drag", args, delivery_mode, bring_to_front)
def scroll(self, *, direction: str, amount: int = 3, element: Optional[int] = None,
x: Optional[int] = None, y: Optional[int] = None, modifiers: Optional[List[str]] = None,
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult:
refusal, args = self._target_args("scroll")
if refusal is not None:
return refusal
args["window_id"] = self._active_window_id
return self._run_input_action("drag", args, delivery_mode, bring_to_front)
args.update(direction=direction, amount=max(1, min(50, amount)))
# An element without a known window_id is not an addressing form here; scrolling then falls through
# to the coordinate form or the bare window. Some driver schemas reject x/y on scroll: only send
# coordinates when the driver advertises support; otherwise it scrolls the targeted window
# (window_id is still sent for routing).
xy = lambda: ({"x": x, "y": y} # noqa: E731
if self._session.supports_capability("input.scroll.coordinates", tool="scroll") else {})
refusal = self._pointer_args("scroll", args, (
("element scroll", {"element_index": element}
if element is not None and self._active_window_id is not None else None),
("coordinate scroll", xy if x is not None and y is not None else None),
), None)
return refusal if refusal is not None else self._run_input_action("scroll", args, delivery_mode, bring_to_front)
def scroll(
self,
*,
direction: str,
amount: int = 3,
element: Optional[int] = None,
x: Optional[int] = None,
y: Optional[int] = None,
modifiers: Optional[List[str]] = None,
delivery_mode: Optional[str] = None,
bring_to_front: bool = False,
) -> ActionResult:
missing = self._no_target("scroll")
if missing is not None:
return missing
args: Dict[str, Any] = {"pid": self._active_pid, "direction": direction,
"amount": max(1, min(50, amount))}
if element is not None and self._active_window_id is not None:
args.update(element_index=element, window_id=self._active_window_id)
elif x is not None and y is not None:
refusal = self._need_window("scroll", "coordinate scroll")
if refusal is not None:
return refusal
# Some driver schemas reject x/y on scroll: only send coordinates
# when the driver advertises support; otherwise it scrolls the
# targeted window (window_id is still sent for routing).
if self._session.supports_capability("input.scroll.coordinates", tool="scroll"):
args.update(x=x, y=y)
args["window_id"] = self._active_window_id
return self._run_input_action("scroll", args, delivery_mode, bring_to_front)
def type_text(self, text: str, *, delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult:
refusal, args = self._target_args("type_text", need_window=True)
return refusal if refusal is not None else self._run_input_action("type_text", {**args, "text": text},
delivery_mode, bring_to_front)
# ── Keyboard ───────────────────────────────────────────────────
def type_text(self, text: str, *, delivery_mode: Optional[str] = None,
bring_to_front: bool = False) -> ActionResult:
missing = self._no_target("type_text", need_window=True)
if missing is not None:
return missing
args: Dict[str, Any] = {"pid": self._active_pid, "window_id": self._active_window_id, "text": text}
return self._run_input_action("type_text", args, delivery_mode, bring_to_front)
def key(self, keys: str, *, delivery_mode: Optional[str] = None,
bring_to_front: bool = False) -> ActionResult:
missing = self._no_target("key", need_window=True)
if missing is not None:
return missing
def key(self, keys: str, *, delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult:
refusal, args = self._target_args("key", need_window=True)
if refusal is not None:
return refusal
key_name, modifiers = _parse_key_combo(keys)
if not key_name:
return _refuse("key", f"Could not parse key from '{keys}'.")
args: Dict[str, Any] = {"pid": self._active_pid, "window_id": self._active_window_id}
if modifiers: # hotkey requires at least one modifier + one key
args["keys"] = modifiers + [key_name]
return self._run_input_action("hotkey", args, delivery_mode, bring_to_front)
args["key"] = key_name
return self._run_input_action("press_key", args, delivery_mode, bring_to_front)
return self._run_input_action("hotkey", {**args, "keys": modifiers + [key_name]}, delivery_mode, bring_to_front)
return self._run_input_action("press_key", {**args, "key": key_name}, delivery_mode, bring_to_front)
# ── Value setter ────────────────────────────────────────────────
def set_value(self, value: str, element: Optional[int] = None) -> ActionResult:
"""Set a value on an element. Handles AXPopUpButton selects natively."""
missing = self._no_target("set_value", need_window=True)
if missing is not None:
return missing
refusal, args = self._target_args("set_value", need_window=True)
if refusal is not None:
return refusal
if element is None:
return _refuse("set_value", "set_value requires element= (element index).")
return self._action("set_value", {"pid": self._active_pid, "window_id": self._active_window_id,
"element_index": element, "value": value})
return self._action("set_value", {**args, "element_index": element, "value": value})