From 48c517a04f63cf304894787c6b5d176d400ee5fc Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:12:24 -0700 Subject: [PATCH 1/4] refactor(computer_use): input mixin target/pointer arg resolution via _target_args + _pointer_args --- tools/computer_use/cua_backend_input.py | 198 +++++++++++------------- 1 file changed, 88 insertions(+), 110 deletions(-) diff --git a/tools/computer_use/cua_backend_input.py b/tools/computer_use/cua_backend_input.py index 7022379b0b..9f11c2de1f 100644 --- a/tools/computer_use/cua_backend_input.py +++ b/tools/computer_use/cua_backend_input.py @@ -4,7 +4,7 @@ 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 from tools.computer_use.backend import ActionResult from tools.computer_use.cua_backend_parse import _parse_key_combo @@ -16,6 +16,9 @@ _FOREGROUND_UNSUPPORTED_MSG = ( "foreground delivery is unavailable. Use another verified rung without " "assuming the reported package version describes the live schema." ) +# (what, lazy extra-args) pointer variants: ``extra`` is None when the caller +# did not supply that addressing form. +_Variant = Tuple[str, Optional[Callable[[], Dict[str, Any]]]] def _refuse(action: str, message: str, **fields: Any) -> ActionResult: return ActionResult(ok=False, action=action, message=message, **fields) @@ -24,28 +27,38 @@ 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]: + # ── Target resolution ────────────────────────────────────────── + 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), {} + args: Dict[str, Any] = {"pid": self._active_pid} + if need_window: + args["window_id"] = self._active_window_id + return None, args - 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 = proceed).""" + 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()) + args["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. - """ + """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": @@ -60,11 +73,9 @@ class _InputMixin: 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. - """ + original foreground input runs unchanged.""" refusal = self._apply_delivery(action, args, delivery_mode) if refusal is not None: return refusal @@ -87,21 +98,12 @@ class _InputMixin: 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 + 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. @@ -109,102 +111,78 @@ class _InputMixin: 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.") + args["button"] = button_norm + refusal = self._pointer_args(tool, args, ( + ("element_index click", (lambda: {"element_index": element}) if element is not None else None), + ("coordinate click", (lambda: {"x": x, "y": y}) if x is not None and y is not None else None), + ), "click requires element= or x/y.") if refusal is not None: return refusal - args["window_id"] = self._active_window_id if modifiers: args["modifier"] = modifiers return 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", (lambda: {"from_element": from_element, "to_element": to_element}) + if from_element is not None and to_element is not None else None), + ("coordinate drag", (lambda: {"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.") 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) - 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 + 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.update(direction=direction, amount=max(1, min(50, amount))) + + def _xy() -> Dict[str, Any]: # 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 {"x": x, "y": y} + return {} + + # 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. + refusal = self._pointer_args("scroll", args, ( + ("element scroll", (lambda: {"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) + if refusal is not None: + return refusal return self._run_input_action("scroll", args, 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} + refusal, args = self._target_args("type_text", need_window=True) + if refusal is not None: + return refusal + args["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 + 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) @@ -214,10 +192,10 @@ class _InputMixin: # ── 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}) + args.update(element_index=element, value=value) + return self._action("set_value", args) From a7c1ec6ecb6e4855cd6a8e4d6889a067d1ceed6e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:24:06 -0700 Subject: [PATCH 2/4] refactor(computer_use): capture mixin disarm-on-failure contextmanager + shared MCP-empty->CLI re-fetch helper; table-driven list_apps/_match_windows_for_app --- tools/computer_use/cua_backend_capture.py | 185 ++++++++++------------ 1 file changed, 83 insertions(+), 102 deletions(-) diff --git a/tools/computer_use/cua_backend_capture.py b/tools/computer_use/cua_backend_capture.py index 23c70e5f35..361260fe88 100644 --- a/tools/computer_use/cua_backend_capture.py +++ b/tools/computer_use/cua_backend_capture.py @@ -1,9 +1,7 @@ """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. +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,7 +12,8 @@ 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 @@ -91,30 +90,25 @@ 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. - """ + 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 + for w in pool: + if active_id is not None and w.get("window_id") == active_id: + return w 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 + 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.""" @@ -132,12 +126,11 @@ def _gws_is_empty(out: Dict[str, Any]) -> bool: 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 _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)``, replacing the given size with the + sniffed one when the bytes decode to a readable PNG/JPEG header.""" try: raw = base64.b64decode(png_b64, validate=False) png_bytes_len = len(raw) @@ -164,6 +157,15 @@ class _CaptureMixin: """capture()/list_windows()/list_apps()/focus_app() and their window-discovery helpers.""" # ── Failure plumbing ─────────────────────────────────────────── + @contextmanager + def _disarming(self) -> Iterator[None]: + """Forget the sticky target when the wrapped capture-stage step raises.""" + try: + yield + except Exception: + self._clear_active_target() + raise + def _failed_capture(self, mode: str, message: str = "") -> CaptureResult: """Return an empty capture after disarming any prior target context.""" self._clear_active_target() @@ -172,22 +174,19 @@ class _CaptureMixin: 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.""" - try: + with self._disarming(): out = self._session.call_tool(name, args) - 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 "")) + 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: @@ -200,27 +199,25 @@ class _CaptureMixin: return None return cli_out - # ── 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) + if empty(out): + cli_out = self._cli_refetch(name, args, timeout, what, warning, *warning_args) + if cli_out is not None and not empty(cli_out): + out = cli_out + return out + # ── Window discovery ─────────────────────────────────────────── 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 + args = {"on_screen_only": True, "session": self._session_id} + return _sorted_windows(self._fetch_or_refetch( + "list_windows", args, 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 @@ -255,16 +252,14 @@ class _CaptureMixin: 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 + # 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. - return [w for w in windows - if not _name(w).strip() and app_lower in str(w.get("title", "")).lower()] + 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": @@ -276,13 +271,12 @@ class _CaptureMixin: return self._failed_capture(mode, "") 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, "", - ) + return self._failed_capture(mode, "") 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() + 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. @@ -291,7 +285,6 @@ class _CaptureMixin: 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 @@ -300,8 +293,7 @@ class _CaptureMixin: 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) - + return sorted(desktop, key=lambda w: not _is_desktop_window(w, _DESKTOP_BACKDROP_NAMES)) # 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. @@ -314,12 +306,10 @@ class _CaptureMixin: 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. - """ + 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 = "" @@ -336,27 +326,23 @@ class _CaptureMixin: # The title is cheap and useful; `elements` stays empty by contract. _, window_title = _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") + 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) if cli_out is not None and cli_out.get("images"): png_b64, image_mime_type = cli_out["images"][0], "image/png" 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 - + 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. @@ -390,7 +376,6 @@ class _CaptureMixin: 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) app_name = target["app_name"] @@ -404,7 +389,6 @@ class _CaptureMixin: png_b64, image_mime_type, window_title = self._capture_vision() else: png_b64, image_mime_type, elements, window_title = 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, @@ -419,8 +403,7 @@ class _CaptureMixin: 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 {} + sc = self._session.call_tool("get_config", {"session": self._session_id}, timeout=10.0).get("structuredContent") or {} if isinstance(sc, dict) and isinstance(sc.get("capture_scope"), str): previous_scope = sc["capture_scope"] except Exception as e: @@ -462,25 +445,21 @@ class _CaptureMixin: 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") + 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. - 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 + # 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: + apps = candidate() + if isinstance(apps, list) and apps: + return apps # Old text-only drivers retain a small, name/PID-only fallback. if isinstance(data, str): return [ @@ -494,7 +473,9 @@ class _CaptureMixin: 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) + with self._disarming(): + windows = self._load_windows() + matched = self._match_windows_for_app(windows, app) # No silent fallback to the frontmost window: that hides the real # failure (often a localized macOS app-name mismatch). if not matched: From 94b9b3860f50ba0313bbca5ae3a9f51ad63a8ca4 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:29:02 -0700 Subject: [PATCH 3/4] =?UTF-8?q?refactor(computer=5Fuse):=20backend=20ABC?= =?UTF-8?q?=20=E2=80=94=20merge=20method=20docstrings=20into=20class=20con?= =?UTF-8?q?tract,=20module-level=20time=20import?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/computer_use/backend.py | 55 ++++++++++++----------------------- 1 file changed, 19 insertions(+), 36 deletions(-) diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index dcd5e903f3..dcdd5756c6 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -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 @@ -30,11 +28,9 @@ def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]: 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 +39,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 @@ -96,13 +90,11 @@ class CaptureResult: @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 @@ -121,12 +113,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: ... @@ -138,12 +129,10 @@ class ComputerUseBackend(ABC): def is_available(self) -> bool: """True if the backend can be used 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,20 +147,15 @@ 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.""" @@ -191,6 +175,5 @@ class ComputerUseBackend(ABC): def wait(self, seconds: float) -> ActionResult: """Default implementation: time.sleep.""" - import time time.sleep(max(0.0, min(seconds, 30.0))) return ActionResult(ok=True, action="wait", message=f"waited {seconds:.2f}s") From 303a8cafb07ff07aec89c7d368957c509d5f3511 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:28:25 -0700 Subject: [PATCH 4/4] =?UTF-8?q?refactor(computer=5Fuse):=20compact=20captu?= =?UTF-8?q?re/input/backend=20=E2=80=94=20drop=20dead=20UIElement.center,?= =?UTF-8?q?=20section=20banners,=20reflow=20docstrings=20to=20118=20cols?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/tools/test_computer_use.py | 4 +- tools/computer_use/backend.py | 44 +-- tools/computer_use/cua_backend_capture.py | 458 ++++++++-------------- tools/computer_use/cua_backend_input.py | 140 +++---- 4 files changed, 233 insertions(+), 413 deletions(-) diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index 0ac3fed17e..844bb3dffa 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -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): diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index dcdd5756c6..842ee737c1 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -17,11 +17,8 @@ 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): @@ -56,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: @@ -80,11 +72,11 @@ 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 = "" @@ -101,7 +93,6 @@ class ActionResult: 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 @@ -126,8 +117,7 @@ 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) @abstractmethod def capture(self, mode: str = "som", app: Optional[str] = None, pid: Optional[int] = None, @@ -157,23 +147,19 @@ class ComputerUseBackend(ABC): def key(self, keys: str, *, delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ... @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.""" + 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") diff --git a/tools/computer_use/cua_backend_capture.py b/tools/computer_use/cua_backend_capture.py index 361260fe88..0fb8b9b3b4 100644 --- a/tools/computer_use/cua_backend_capture.py +++ b/tools/computer_use/cua_backend_capture.py @@ -1,8 +1,6 @@ -"""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. -""" +"""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 @@ -18,63 +16,35 @@ 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_APP_MATCH_MSG = ( - "" -) -_NO_DESKTOP_IMAGE_MSG = ( - "" -) -_FULL_SCREEN_NOTE = ( - "full-screen capture has no interactable elements; to act on what you see, " - "call capture(app='') 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_APP_MATCH_MSG = ("") +_NO_DESKTOP_IMAGE_MSG = ("") +_FULL_SCREEN_NOTE = ("full-screen capture has no interactable elements; to act on what you see, call " + "capture(app='') 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.""" @@ -89,74 +59,51 @@ 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() - for w in pool: - if active_id is not None and 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()).""" + """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 - return not _tree_and_title(out)[0].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]: - """``(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 ─────────────────────────────────────────── @contextmanager def _disarming(self) -> Iterator[None]: """Forget the sticky target when the wrapped capture-stage step raises.""" @@ -169,8 +116,7 @@ class _CaptureMixin: 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) + 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.""" @@ -184,77 +130,64 @@ class _CaptureMixin: 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.""" + """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 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).""" + """``_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) - if empty(out): - cli_out = self._cli_refetch(name, args, timeout, what, warning, *warning_args) - if cli_out is not None and not empty(cli_out): - out = cli_out - return out + 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 - # ── Window discovery ─────────────────────────────────────────── def _load_windows(self) -> List[Dict[str, Any]]: - """Visible windows frontmost-first, re-fetching over the CLI transport - when MCP returns nothing.""" - args = {"on_screen_only": True, "session": self._session_id} + """Visible windows frontmost-first, re-fetching over the CLI transport when MCP returns nothing.""" return _sorted_windows(self._fetch_or_refetch( - "list_windows", args, 20.0, "list_windows", lambda out: not _sorted_windows(out), + "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) - # 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. + 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], @@ -265,154 +198,117 @@ class _CaptureMixin: 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, "") - target_pid, target_window_id = _positive_int(pid), _positive_int(window_id) - if target_pid is None or target_window_id is None: + if (target_pid := _positive_int(pid)) is None or (target_window_id := _positive_int(window_id)) is None: return self._failed_capture(mode, "") - return [{"app_name": app or "", "pid": target_pid, "window_id": target_window_id, - "off_screen": False, "title": "", "z_index": 0}] - + 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: not _is_desktop_window(w, _DESKTOP_BACKDROP_NAMES)) - # 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))) + # 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)) - # ── 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: 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) - if cli_out is not None and cli_out.get("images"): + 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)``.""" - # 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". + # 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) + "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: - sc = self._session.call_tool("get_config", {"session": self._session_id}, timeout=10.0).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") @@ -423,78 +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, 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)), - ) + # 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: - apps = candidate() - if isinstance(apps, list) and apps: + 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.""" + """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(): - windows = self._load_windows() - matched = self._match_windows_for_app(windows, app) - # No silent fallback to the frontmost window: that hides the real - # failure (often a localized macOS app-name mismatch). + 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 diff --git a/tools/computer_use/cua_backend_input.py b/tools/computer_use/cua_backend_input.py index 9f11c2de1f..a8b5b036f4 100644 --- a/tools/computer_use/cua_backend_input.py +++ b/tools/computer_use/cua_backend_input.py @@ -1,24 +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, Callable, Dict, List, Optional, Sequence, 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." -) -# (what, lazy extra-args) pointer variants: ``extra`` is None when the caller -# did not supply that addressing form. -_Variant = Tuple[str, Optional[Callable[[], Dict[str, Any]]]] +_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) @@ -27,55 +24,44 @@ def _refuse(action: str, message: str, **fields: Any) -> ActionResult: class _InputMixin: """Pointer / keyboard / value-setter actions against the sticky target.""" - # ── Target resolution ────────────────────────────────────────── 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), {} - args: Dict[str, Any] = {"pid": self._active_pid} - if need_window: - args["window_id"] = self._active_window_id - return None, args + return None, {"pid": self._active_pid, **({"window_id": self._active_window_id} if need_window else {})} 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 = proceed).""" + """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()) - args["window_id"] = self._active_window_id + 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).""" + 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 @@ -84,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") @@ -97,30 +82,25 @@ 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: 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. + # 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["button"] = button_norm + tool, args["button"] = ("double_click" if click_count == 2 else "click"), button_norm refusal = self._pointer_args(tool, args, ( - ("element_index click", (lambda: {"element_index": element}) if element is not None else None), - ("coordinate click", (lambda: {"x": x, "y": y}) if x is not None and y is not None else None), + ("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 refusal is not None: - return refusal 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, @@ -129,15 +109,13 @@ class _InputMixin: refusal, args = self._target_args("drag") if refusal is None: refusal = self._pointer_args("drag", args, ( - ("element-based drag", (lambda: {"from_element": from_element, "to_element": to_element}) + ("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", (lambda: {"from_x": int(from_xy[0]), "from_y": int(from_xy[1]), - "to_x": int(to_xy[0]), "to_y": int(to_xy[1])}) + ("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.") - if refusal is not None: - return refusal - return self._run_input_action("drag", args, delivery_mode, bring_to_front) + 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, @@ -146,37 +124,25 @@ class _InputMixin: if refusal is not None: return refusal args.update(direction=direction, amount=max(1, min(50, amount))) - - def _xy() -> Dict[str, Any]: - # 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"): - return {"x": x, "y": y} - return {} - - # 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. + # 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", (lambda: {"element_index": element}) + ("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), + ("coordinate scroll", xy if x is not None and y is not None else None), ), None) - if refusal is not None: - return refusal - return self._run_input_action("scroll", args, delivery_mode, bring_to_front) + return refusal if refusal is not None else self._run_input_action("scroll", args, delivery_mode, bring_to_front) - # ── Keyboard ─────────────────────────────────────────────────── - def type_text(self, text: str, *, delivery_mode: Optional[str] = None, - bring_to_front: bool = False) -> ActionResult: + 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) - if refusal is not None: - return refusal - args["text"] = text - return self._run_input_action("type_text", args, delivery_mode, bring_to_front) + return refusal if refusal is not None else self._run_input_action("type_text", {**args, "text": text}, + delivery_mode, bring_to_front) - def key(self, keys: str, *, delivery_mode: Optional[str] = None, - bring_to_front: bool = False) -> ActionResult: + 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 @@ -184,12 +150,9 @@ class _InputMixin: if not key_name: return _refuse("key", f"Could not parse key from '{keys}'.") 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.""" refusal, args = self._target_args("set_value", need_window=True) @@ -197,5 +160,4 @@ class _InputMixin: return refusal if element is None: return _refuse("set_value", "set_value requires element= (element index).") - args.update(element_index=element, value=value) - return self._action("set_value", args) + return self._action("set_value", {**args, "element_index": element, "value": value})