diff --git a/scripts/release.py b/scripts/release.py index 476fa82001ac..5fa023bef2c9 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -1485,6 +1485,7 @@ AUTHOR_MAP = { "2114364329@qq.com": "cuyua9", "2557058999@qq.com": "Disaster-Terminator", "cine.dreamer.one@gmail.com": "LeonSGP43", + "willies578@gmail.com": "grimmjoww578", # PR #52142 co-author (direct CUA window targeting) "zyprothh@gmail.com": "Zyproth", "amitgaur@gmail.com": "amitgaur", "albuquerque.abner@gmail.com": "mrbob-git", diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index 9f4a6a5ca051..44b5f1a5dbd1 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -6,7 +6,7 @@ import base64 import json import os import sys -from typing import Any, Dict, List +from typing import Any, Dict, List, cast from unittest.mock import MagicMock, patch import pytest @@ -69,9 +69,17 @@ class TestSchema: actions = set(COMPUTER_USE_SCHEMA["parameters"]["properties"]["action"]["enum"]) assert actions >= { "capture", "click", "double_click", "right_click", "middle_click", - "drag", "scroll", "type", "key", "wait", "list_apps", "focus_app", + "drag", "scroll", "type", "key", "wait", "list_apps", "list_windows", + "focus_app", } + def test_schema_exposes_exact_capture_targeting(self): + from tools.computer_use.schema import COMPUTER_USE_SCHEMA + + props = COMPUTER_USE_SCHEMA["parameters"]["properties"] + assert props["pid"]["type"] == "integer" + assert props["window_id"]["type"] == "integer" + def test_capture_mode_enum_has_som_vision_ax(self): from tools.computer_use.schema import COMPUTER_USE_SCHEMA modes = set(COMPUTER_USE_SCHEMA["parameters"]["properties"]["mode"]["enum"]) @@ -165,6 +173,12 @@ class TestDispatch: assert "apps" in parsed assert parsed["count"] == 0 + def test_list_windows_returns_json(self, noop_backend): + from tools.computer_use.tool import handle_computer_use + out = handle_computer_use({"action": "list_windows"}) + parsed = json.loads(out) + assert parsed == {"windows": [], "count": 0} + def test_wait_clamps_long_waits(self, noop_backend): from tools.computer_use.tool import handle_computer_use # The backend's default wait() uses time.sleep with clamping. @@ -265,6 +279,19 @@ class TestDispatch: out = handle_computer_use({"action": "set_value"}) parsed = json.loads(out) assert "error" in parsed + + def test_capture_forwards_exact_pid_window_target(self, noop_backend): + from tools.computer_use.tool import handle_computer_use + + handle_computer_use({ + "action": "capture", "mode": "ax", "pid": 23502, "window_id": 58720504, + }) + + capture_kw = next(c[1] for c in noop_backend.calls if c[0] == "capture") + assert capture_kw == { + "mode": "ax", "app": None, "pid": 23502, "window_id": 58720504, + } + def test_capture_after_skipped_when_action_failed(self, noop_backend): """capture_after must not fire when res.ok=False (regression guard). @@ -1421,6 +1448,45 @@ def _make_cua_backend_with_windows(windows: List[Dict[str, Any]]): return backend +def _make_cua_backend_with_windows_and_apps( + windows: List[Dict[str, Any]], apps: List[Dict[str, Any]] +): + """Construct a backend whose mocked session serves list_windows/list_apps.""" + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + backend._session = MagicMock() + + def _call_tool(name, args): + if name == "list_windows": + return { + "data": "", + "images": [], + "structuredContent": {"windows": windows}, + "isError": False, + } + if name == "list_apps": + # cua-driver MCP puts the canonical app objects in + # structuredContent; `data` is only a human-readable summary. + return { + "data": f"✅ Found {len(apps)} app(s)", + "images": [], + "structuredContent": {"apps": apps}, + "isError": False, + } + if name == "get_window_state": + return { + "data": '✅ FreeCAD — 0 elements\n', + "images": [], + "structuredContent": None, + "isError": False, + } + raise AssertionError(f"unexpected tool call: {name}") + + backend._session.call_tool.side_effect = _call_tool + return backend + + class TestCuaDriverSessionReconnect: """Verify reconnect-once on a closed-resource error. After the lifecycle-owner refactor (Sun Jun 21 2026) the session no longer goes @@ -1595,6 +1661,32 @@ class TestCuaDriverSessionReconnect: assert "AXButton" in out["data"] assert "7 elements" in out["data"] + def test_cli_fallback_preserves_logical_error(self, monkeypatch): + from typing import Any, cast + from tools.computer_use.cua_backend import _CuaDriverSession + + session = cast(Any, _CuaDriverSession.__new__(_CuaDriverSession)) + + class FakeProc: + stdout = '{"isError": true, "message": "bad target"}' + stderr = "" + returncode = 0 + + monkeypatch.setattr("subprocess.run", lambda *a, **kw: FakeProc()) + + out = session._call_tool_via_cli("list_windows", {}, 30.0) + + assert out["isError"] is True + + def test_extract_tool_result_missing_mock_error_flag_is_not_error(self): + from tools.computer_use.cua_backend import _extract_tool_result + + result = MagicMock() + result.content = [] + result.structuredContent = None + + assert _extract_tool_result(result)["isError"] is False + class TestCaptureEmptyResultClipFallback: """When the MCP bridge returns a degenerate/empty get_window_state result @@ -1740,6 +1832,231 @@ class TestCaptureAppFilterNoMatch: assert backend._active_pid == 200 assert backend._active_window_id == 2 + def test_app_filter_falls_back_to_list_apps_metadata(self): + windows = [ + {"app_name": "Qt6Application", "pid": 7675, "window_id": 42, + "is_on_screen": True, "title": "FreeCAD 1.1.1", "z_index": 0}, + ] + apps = [ + {"name": "FreeCAD", "bundle_id": "org.freecad.FreeCAD", "pid": 7675}, + ] + backend = _make_cua_backend_with_windows_and_apps(windows, apps) + + cap = backend.capture(mode="ax", app="org.freecad.FreeCAD") + + assert cap.app == "Qt6Application" + assert backend._active_pid == 7675 + assert backend._active_window_id == 42 + + def test_exact_metadata_alias_beats_broader_direct_window_name(self): + windows = [ + {"app_name": "Visual Studio Code", "pid": 100, "window_id": 1, + "is_on_screen": True, "title": "Visual Studio Code", "z_index": 0}, + {"app_name": "Qt6Application", "pid": 200, "window_id": 2, + "is_on_screen": True, "title": "Code", "z_index": 1}, + ] + apps = [{"name": "Code", "bundle_id": "org.example.Code", "pid": 200}] + backend = _make_cua_backend_with_windows_and_apps(windows, apps) + + cap = backend.capture(mode="ax", app="Code") + + assert cap.app == "Qt6Application" + assert backend._active_pid == 200 + assert backend._active_window_id == 2 + + def test_exact_pid_window_capture_bypasses_window_discovery(self): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + session = MagicMock() + + def _call_tool(name, args): + assert name != "list_windows", "exact target must bypass discovery" + assert name == "get_window_state" + assert args["pid"] == 7675 + assert args["window_id"] == 42 + return { + "data": "✅ FreeCAD — 0 elements", "images": [], + "structuredContent": None, "isError": False, + } + + session.call_tool.side_effect = _call_tool + backend._session = session + + cap = backend.capture(mode="ax", pid=7675, window_id=42) + + assert cap.app == "" + assert backend._active_pid == 7675 + assert backend._active_window_id == 42 + + @pytest.mark.parametrize( + ("pid", "window_id", "diagnostic"), + [ + (None, 2, "both"), + ("bad", 2, "positive integer"), + (True, 2, "positive integer"), + (1, False, "positive integer"), + (-1, 2, "positive integer"), + (1, 0, "positive integer"), + ], + ) + def test_failed_exact_capture_clears_prior_target_and_tokens( + self, pid, window_id, diagnostic, + ): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + session = MagicMock() + session.call_tool.return_value = { + "data": "ok", "images": [], "structuredContent": None, "isError": False, + } + backend._session = session + backend._active_pid = 111 + backend._active_window_id = 222 + backend._last_app = "Previous" + backend._last_target = {"pid": 111, "window_id": 222} + backend._snapshot_tokens = {1: "stale-token"} + + cap = backend.capture(mode="ax", pid=pid, window_id=window_id) + + assert cap.width == 0 and cap.height == 0 + assert diagnostic in cap.window_title + assert backend._active_pid is None + assert backend._active_window_id is None + assert backend._last_target is None + assert backend._snapshot_tokens == {} + assert backend.click(x=1, y=2).ok is False + session.call_tool.assert_not_called() + + def test_list_windows_drops_nonpositive_and_boolean_identifiers(self): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + session = MagicMock() + session.call_tool.return_value = { + "data": "", "images": [], "isError": False, + "structuredContent": {"windows": [ + {"app_name": "Good", "pid": 12, "window_id": 34, + "is_on_screen": True, "z_index": 0}, + {"app_name": "Bool", "pid": True, "window_id": 2, + "is_on_screen": True, "z_index": 1}, + {"app_name": "Zero", "pid": 0, "window_id": 3, + "is_on_screen": True, "z_index": 2}, + {"app_name": "Negative", "pid": 4, "window_id": -1, + "is_on_screen": True, "z_index": 3}, + ]}, + } + backend._session = session + + assert backend.list_windows() == [{ + "app_name": "Good", "pid": 12, "window_id": 34, + "off_screen": False, "title": "", "z_index": 0, + }] + + def test_capture_transport_exception_disarms_prior_target(self): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + session = MagicMock() + session.call_tool.side_effect = RuntimeError("list_windows failed") + backend._session = session + backend._active_pid = 111 + backend._active_window_id = 222 + backend._last_target = {"pid": 111, "window_id": 222} + backend._snapshot_tokens = {1: "stale-token"} + + with pytest.raises(RuntimeError, match="list_windows failed"): + backend.capture(mode="ax") + + assert backend._active_pid is None + assert backend._active_window_id is None + assert backend._last_target is None + assert backend._snapshot_tokens == {} + + def test_focus_changes_target_and_clears_snapshot_tokens(self): + windows = [ + {"app_name": "New", "pid": 333, "window_id": 444, + "is_on_screen": True, "z_index": 0}, + ] + backend = _make_cua_backend_with_windows(windows) + backend._active_pid = 111 + backend._active_window_id = 222 + backend._snapshot_tokens = {1: "stale-token"} + + res = backend.focus_app("New") + + assert res.ok is True + assert backend._active_pid == 333 + assert backend._active_window_id == 444 + assert backend._snapshot_tokens == {} + + def test_get_window_state_exception_disarms_selected_target_and_tokens(self): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + session = MagicMock() + session.call_tool.side_effect = [ + {"data": "", "images": [], "isError": False, + "structuredContent": {"windows": [{ + "app_name": "New", "pid": 333, "window_id": 444, + "is_on_screen": True, "z_index": 0, + }]}}, + RuntimeError("get_window_state failed"), + ] + backend._session = session + backend._active_pid = 111 + backend._active_window_id = 222 + backend._snapshot_tokens = {1: "stale-token"} + + with pytest.raises(RuntimeError, match="get_window_state failed"): + backend.capture(mode="ax", app="New") + + assert backend._active_pid is None + assert backend._active_window_id is None + assert backend._last_target is None + assert backend._snapshot_tokens == {} + + @pytest.mark.parametrize("tool_name", ["list_windows", "get_window_state"]) + def test_capture_logical_driver_error_disarms_target_and_tokens(self, tool_name): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + session = MagicMock() + window_result = { + "data": "", "images": [], "isError": False, + "structuredContent": {"windows": [{ + "app_name": "New", "pid": 333, "window_id": 444, + "is_on_screen": True, "z_index": 0, + }]}, + } + logical_error = { + "data": f"{tool_name} rejected target", "images": [], + "structuredContent": None, "isError": True, + } + session.call_tool.side_effect = ( + [logical_error] + if tool_name == "list_windows" + else [window_result, logical_error] + ) + backend._session = session + backend._active_pid = 111 + backend._active_window_id = 222 + backend._last_app = "Previous" + backend._last_target = {"pid": 111, "window_id": 222} + backend._snapshot_tokens = {1: "stale-token"} + + with pytest.raises(RuntimeError, match=f"cua-driver {tool_name} failed"): + backend.capture(mode="ax", app="New") + + assert backend._active_pid is None + assert backend._active_window_id is None + assert backend._last_app is None + assert backend._last_target is None + assert backend._snapshot_tokens == {} + calls_before = session.call_tool.call_count + assert backend.click(x=1, y=2).ok is False + assert session.call_tool.call_count == calls_before + def test_no_app_filter_still_picks_frontmost(self): """When no app= is given, capture continues to pick the frontmost window — the no-match early-return must not fire on the empty case.""" @@ -1798,6 +2115,181 @@ class TestFocusAppFilterNoMatch: assert backend._active_pid == 200 assert backend._active_window_id == 2 + def test_focus_app_falls_back_to_list_apps_metadata(self): + windows = [ + {"app_name": "Qt6Application", "pid": 7675, "window_id": 42, + "is_on_screen": True, "title": "FreeCAD 1.1.1", "z_index": 0}, + ] + apps = [ + {"name": "FreeCAD", "bundle_id": "org.freecad.FreeCAD", "pid": 7675}, + ] + backend = _make_cua_backend_with_windows_and_apps(windows, apps) + + res = backend.focus_app("FreeCAD") + + assert res.ok is True + assert backend._active_pid == 7675 + assert backend._active_window_id == 42 + + def test_focus_exact_metadata_alias_beats_broader_direct_window_name(self): + windows = [ + {"app_name": "Visual Studio Code", "pid": 100, "window_id": 1, + "is_on_screen": True, "title": "Visual Studio Code", "z_index": 0}, + {"app_name": "Qt6Application", "pid": 200, "window_id": 2, + "is_on_screen": True, "title": "Code", "z_index": 1}, + ] + apps = [{"name": "Code", "bundle_id": "org.example.Code", "pid": 200}] + backend = _make_cua_backend_with_windows_and_apps(windows, apps) + + res = backend.focus_app("Code") + + assert res.ok is True + assert backend._active_pid == 200 + assert backend._active_window_id == 2 + + def test_title_fallback_only_targets_nameless_windows(self): + windows = [ + {"app_name": "計算機", "pid": 200, "window_id": 2, + "is_on_screen": True, "title": "Calculator", "z_index": 0}, + {"app_name": "", "pid": 300, "window_id": 3, + "is_on_screen": True, "title": "Calculator", "z_index": 1}, + ] + backend = _make_cua_backend_with_windows_and_apps(windows, []) + + cap = backend.capture(mode="ax", app="Calculator") + + assert backend._active_pid == 300 + assert backend._active_window_id == 3 + assert cap.app == "" + + def test_title_fallback_survives_list_apps_failure(self): + windows = [ + {"app_name": "", "pid": 300, "window_id": 3, + "is_on_screen": True, "title": "Mousepad", "z_index": 0}, + ] + backend = _make_cua_backend_with_windows(windows) + session = cast(Any, backend._session) + session.call_tool.side_effect = [ + { + "data": "", "images": [], "isError": False, + "structuredContent": {"windows": windows}, + }, + RuntimeError("list_apps unavailable"), + { + "data": "✅ Mousepad — 0 elements", "images": [], "isError": False, + "structuredContent": None, + }, + ] + + cap = backend.capture(mode="ax", app="Mousepad") + + assert cap.app == "" + assert backend._active_pid == 300 + assert backend._active_window_id == 3 + + def test_focus_app_title_fallback_targets_a_nameless_window(self): + windows = [ + {"app_name": "", "pid": 300, "window_id": 3, + "is_on_screen": True, "title": "Mousepad", "z_index": 0}, + ] + backend = _make_cua_backend_with_windows_and_apps(windows, []) + + res = backend.focus_app("Mousepad") + + assert res.ok is True + assert backend._active_pid == 300 + assert backend._active_window_id == 3 + + def test_installed_only_metadata_cannot_target_a_pid_zero_window(self): + windows = [ + {"app_name": "", "pid": 0, "window_id": 7, + "is_on_screen": True, "title": "Desktop", "z_index": 0}, + ] + apps = [ + {"name": "FreeCAD", "bundle_id": "org.freecad.FreeCAD", + "pid": 0, "running": False}, + ] + backend = _make_cua_backend_with_windows_and_apps(windows, apps) + + cap = backend.capture(mode="ax", app="org.freecad.FreeCAD") + + assert cap.app == "" + assert backend._active_pid is None + assert backend._active_window_id is None + + +class TestCaptureAfterExactTarget: + def test_followup_capture_reuses_exact_window_identity(self): + from tools.computer_use.backend import ActionResult, CaptureResult + from tools.computer_use.tool import _maybe_follow_capture + + class GenericWindowBackend: + _last_app = "Qt6Application" + _last_target = {"pid": 7675, "window_id": 42} + + def __init__(self): + self.capture_calls = [] + + def capture(self, mode="som", app=None, pid=None, window_id=None): + self.capture_calls.append({ + "mode": mode, "app": app, "pid": pid, "window_id": window_id, + }) + return CaptureResult( + mode=mode, width=0, height=0, png_b64=None, + elements=[], app="Qt6Application", window_title="FreeCAD", + ) + + backend = GenericWindowBackend() + _maybe_follow_capture(cast(Any, backend), ActionResult(ok=True, action="click"), True) + + assert backend.capture_calls == [{ + "mode": "som", "app": None, "pid": 7675, "window_id": 42, + }] + + def test_dispatch_capture_after_reuses_metadata_resolved_identity(self, monkeypatch): + from tools.computer_use import tool as computer_use_tool + + windows = [ + {"app_name": "Qt6Application", "pid": 100, "window_id": 1, + "is_on_screen": True, "title": "Other Qt App", "z_index": 0}, + {"app_name": "Qt6Application", "pid": 7675, "window_id": 42, + "is_on_screen": True, "title": "FreeCAD", "z_index": 1}, + ] + apps = [ + {"name": "FreeCAD", "bundle_id": "org.freecad.FreeCAD", "pid": 7675}, + ] + backend = _make_cua_backend_with_windows_and_apps(windows, apps) + session = cast(Any, backend._session) + original_call_tool = session.call_tool.side_effect + + def _call_tool(name, args): + if name == "click": + return { + "data": "clicked", "images": [], + "structuredContent": None, "isError": False, + } + return original_call_tool(name, args) + + session.call_tool.side_effect = _call_tool + monkeypatch.setattr(computer_use_tool, "_backend", backend) + + capture_out = computer_use_tool.handle_computer_use({ + "action": "capture", "mode": "ax", "app": "org.freecad.FreeCAD", + }) + assert "error" not in json.loads(capture_out) + click_out = computer_use_tool.handle_computer_use({ + "action": "click", "coordinate": [10, 20], "capture_after": True, + }) + assert "error" not in json.loads(click_out) + + tool_calls = [call.args for call in session.call_tool.call_args_list] + gws_calls = [args for name, args in tool_calls if name == "get_window_state"] + assert len(gws_calls) == 2 + assert all(call["pid"] == 7675 and call["window_id"] == 42 for call in gws_calls) + action_calls = [args for name, args in tool_calls if name == "click"] + assert len(action_calls) == 1 + assert action_calls[0]["window_id"] == 42 + class TestCuaEnvironmentScrubbing: """Verify that cua-driver subprocess environment is sanitized (issue #37878).""" @@ -1985,6 +2477,206 @@ class TestClickButtonPassthrough: assert name == "click" assert args["button"] == "right" assert args["x"] == 10 and args["y"] == 20 + assert args["window_id"] == 222 + + def test_coordinate_drag_and_scroll_keep_the_captured_window(self): + backend = self._backend_with_active_target() + # Mock the capability check so x/y are included (they're gated + # behind the input.scroll.coordinates capability). + backend._session.supports_capability.return_value = True + + backend.drag(from_xy=(10, 20), to_xy=(30, 40)) + drag_name, drag_args = backend._session.call_tool.call_args.args + assert drag_name == "drag" + assert drag_args == { + "pid": 111, + "from_x": 10, + "from_y": 20, + "to_x": 30, + "to_y": 40, + "window_id": 222, + "session": backend._session_id, + } + + backend.scroll(direction="down", x=50, y=60) + scroll_name, scroll_args = backend._session.call_tool.call_args.args + assert scroll_name == "scroll" + assert scroll_args["window_id"] == 222 + assert scroll_args["x"] == 50 and scroll_args["y"] == 60 + + def test_scroll_xy_omitted_without_coordinate_capability(self): + """CUA Driver 0.7.1 Linux schema rejects x/y on scroll. When the + driver does not advertise input.scroll.coordinates, x/y must be + omitted so the call is not rejected.""" + backend = self._backend_with_active_target() + backend._session.supports_capability.return_value = False + + backend.scroll(direction="down", x=50, y=60) + _, scroll_args = backend._session.call_tool.call_args.args + assert "x" not in scroll_args + assert "y" not in scroll_args + assert scroll_args["window_id"] == 222 + + def test_scroll_xy_present_with_coordinate_capability(self): + """When the driver advertises input.scroll.coordinates, x/y are + included in the scroll call.""" + backend = self._backend_with_active_target() + backend._session.supports_capability.return_value = True + + backend.scroll(direction="up", x=10, y=20) + _, scroll_args = backend._session.call_tool.call_args.args + assert scroll_args["x"] == 10 + assert scroll_args["y"] == 20 + assert scroll_args["window_id"] == 222 + + def test_coordinate_actions_without_window_id_fail_closed(self): + backend = self._backend_with_active_target() + backend._active_window_id = None + + assert backend.click(x=10, y=20).ok is False + assert backend.drag(from_xy=(10, 20), to_xy=(30, 40)).ok is False + assert backend.scroll(direction="down", x=10, y=20).ok is False + backend._session.call_tool.assert_not_called() + + +class TestKeyboardWindowIdRouting: + """Review comment #1 on PR #63725: type_text, press_key, and hotkey + must carry window_id so CUA Driver routes input to the correct window + in multi-window apps. Without window_id the driver falls back to the + first window for that PID, which can be the wrong one. + + These tests also verify fail-closed: when _active_window_id is None, + keyboard actions return an error rather than sending PID-only input. + """ + + def _backend_with_active_target(self): + from unittest.mock import MagicMock + from tools.computer_use.cua_backend import CuaDriverBackend + backend = CuaDriverBackend() + backend._session = MagicMock() + backend._session.call_tool.return_value = { + "data": "ok", + "images": [], + "structuredContent": None, + "isError": False, + } + backend._active_pid = 111 + backend._active_window_id = 222 + return backend + + def test_type_text_carries_window_id(self): + backend = self._backend_with_active_target() + backend.type_text("hello world") + name, args = backend._session.call_tool.call_args.args + assert name == "type_text" + assert args["pid"] == 111 + assert args["window_id"] == 222 + assert args["text"] == "hello world" + + def test_press_key_carries_window_id(self): + backend = self._backend_with_active_target() + backend.key("a") + name, args = backend._session.call_tool.call_args.args + assert name == "press_key" + assert args["pid"] == 111 + assert args["window_id"] == 222 + assert args["key"] == "a" + + def test_hotkey_carries_window_id(self): + backend = self._backend_with_active_target() + backend.key("cmd+s") + name, args = backend._session.call_tool.call_args.args + assert name == "hotkey" + assert args["pid"] == 111 + assert args["window_id"] == 222 + assert "cmd" in args["keys"] + assert "s" in args["keys"] + + def test_type_text_fails_closed_without_window_id(self): + backend = self._backend_with_active_target() + backend._active_window_id = None + res = backend.type_text("hello") + assert res.ok is False + backend._session.call_tool.assert_not_called() + + def test_press_key_fails_closed_without_window_id(self): + backend = self._backend_with_active_target() + backend._active_window_id = None + res = backend.key("a") + assert res.ok is False + backend._session.call_tool.assert_not_called() + + def test_hotkey_fails_closed_without_window_id(self): + backend = self._backend_with_active_target() + backend._active_window_id = None + res = backend.key("cmd+s") + assert res.ok is False + backend._session.call_tool.assert_not_called() + + +class TestZIndexSorting: + """Review comment #3 on PR #63725: CUA Driver defines higher z_index + values as closer to the front (top of the stack). The wrapper must sort + descending so the frontmost window is selected. Wayland may return + z_index: null, which must be handled without crashing. + """ + + def test_frontmost_window_selected_by_higher_z_index(self): + """The frontmost window (highest z_index) should be the one + capture() selects as its target.""" + from tools.computer_use.cua_backend import CuaDriverBackend + + windows = [ + {"app_name": "Terminal", "pid": 100, "window_id": 1, + "is_on_screen": True, "title": "term", "z_index": 5}, + {"app_name": "Firefox", "pid": 200, "window_id": 2, + "is_on_screen": True, "title": "browser", "z_index": 10}, + {"app_name": "Desktop", "pid": 300, "window_id": 3, + "is_on_screen": True, "title": "desktop", "z_index": 0}, + ] + backend = _make_cua_backend_with_windows(windows) + + cap = backend.capture(mode="ax") + + # Firefox has z_index=10 (frontmost) — must be selected. + assert backend._active_pid == 200 + assert backend._active_window_id == 2 + + def test_null_z_index_treated_as_lowest(self): + """Wayland may return z_index: null. _ingest_windows must coerce + it to 0 (backmost) so it doesn't crash the sort or get selected + over real foreground windows.""" + from tools.computer_use.cua_backend import _ingest_windows + + raw = [ + {"app_name": "Desktop", "pid": 300, "window_id": 3, + "is_on_screen": True, "title": "desktop", "z_index": None}, + {"app_name": "Firefox", "pid": 200, "window_id": 2, + "is_on_screen": True, "title": "browser", "z_index": 5}, + ] + out = _ingest_windows(raw) + # Both windows survive (null z_index doesn't drop the window). + assert len(out) == 2 + # Null z_index was normalised to 0. + desktop = next(w for w in out if w["app_name"] == "Desktop") + assert desktop["z_index"] == 0 + + def test_null_z_index_does_not_crash_capture(self): + """End-to-end: a null z_index window in the list must not crash + capture(), and a real window (with z_index) must be selected over + the null-z_index desktop.""" + windows = [ + {"app_name": "Desktop", "pid": 300, "window_id": 3, + "is_on_screen": True, "title": "desktop", "z_index": None}, + {"app_name": "Firefox", "pid": 200, "window_id": 2, + "is_on_screen": True, "title": "browser", "z_index": 5}, + ] + backend = _make_cua_backend_with_windows(windows) + + cap = backend.capture(mode="ax") + + assert backend._active_pid == 200 + assert backend._active_window_id == 2 class TestImageMimeTypePropagation: diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index 0537f47b246b..ba109cf2e42e 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -99,7 +99,13 @@ class ComputerUseBackend(ABC): # ── Capture ───────────────────────────────────────────────────── @abstractmethod - def capture(self, mode: str = "som", app: Optional[str] = None) -> CaptureResult: ... + def capture( + self, + mode: str = "som", + app: Optional[str] = None, + pid: Optional[int] = None, + window_id: Optional[int] = None, + ) -> CaptureResult: ... # ── Pointer actions ───────────────────────────────────────────── @abstractmethod @@ -151,6 +157,14 @@ class ComputerUseBackend(ABC): def list_apps(self) -> List[Dict[str, Any]]: """Return running apps with bundle IDs, PIDs, window counts.""" + def list_windows(self) -> List[Dict[str, Any]]: + """Return visible native windows with PID and window identifiers. + + Optional compatibility hook: backends that predate window discovery + remain instantiable and simply report no windows. + """ + 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.""" diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index 32dae8b7b861..1481d09fbb36 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -960,7 +960,12 @@ class _CuaDriverSession: images: List[str] = [] data: Any = None structured: Optional[Dict] = parsed if isinstance(parsed, dict) else None + is_error = False if isinstance(parsed, dict): + # Current cua-driver CLI responses may report logical failures + # in-band even when the subprocess itself exits successfully. + # Preserve that bit so stateful callers can fail closed. + is_error = parsed.get("isError") is True or parsed.get("is_error") is True shot = parsed.get("screenshot_png_b64") if not shot: # Screenshot was routed to a file (ours or the daemon's choice). @@ -978,7 +983,12 @@ class _CuaDriverSession: ec = parsed.get("element_count") summary = f"{ec} elements" if ec is not None else "" data = f"{summary}\n{tree}" if summary else tree - return {"data": data, "images": images, "structuredContent": structured, "isError": False} + return { + "data": data, + "images": images, + "structuredContent": structured, + "isError": is_error, + } finally: if shot_file and os.path.exists(shot_file): try: @@ -1042,7 +1052,9 @@ def _extract_tool_result(mcp_result: Any) -> Dict[str, Any]: data: Any = None images: List[str] = [] image_mime_types: List[str] = [] - is_error = bool(getattr(mcp_result, "isError", False)) + # Use identity, not truthiness: unittest mocks and proxy objects commonly + # synthesize truthy attributes that were never present in the real result. + is_error = getattr(mcp_result, "isError", False) is True structured: Optional[Dict] = getattr(mcp_result, "structuredContent", None) or None text_chunks: List[str] = [] for part in getattr(mcp_result, "content", []) or []: @@ -1109,6 +1121,17 @@ def _image_from_tool_result(out: Dict[str, Any]) -> tuple[Optional[str], Optiona return None, None +def _positive_int(value: Any) -> Optional[int]: + """Return a positive integer, rejecting booleans and malformed values.""" + if isinstance(value, bool) or not isinstance(value, (int, str)): + return None + try: + parsed = int(value) + except ValueError: + return None + return parsed if parsed > 0 else None + + def _ingest_windows(raw_windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Normalise cua-driver ``list_windows`` entries, dropping unusable ones. @@ -1123,23 +1146,28 @@ def _ingest_windows(raw_windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: (``int(w["pid"])``) let one such window abort enumeration of the real, targetable windows. We skip the unusable entries instead so capture() and focus_app() still find the windows that matter. + + ``z_index`` follows CUA Driver semantics: higher = closer to front. + Wayland may return ``z_index: null`` (undefined stacking order); we + treat null as the lowest priority so real windows still sort above + desktop/root windows, and the backmost never ends up selected as the + capture target. """ windows: List[Dict[str, Any]] = [] for w in raw_windows: - pid, window_id = w.get("pid"), w.get("window_id") - if pid is None or window_id is None: - continue - try: - pid_int, window_id_int = int(pid), int(window_id) - except (TypeError, ValueError): + pid_int = _positive_int(w.get("pid")) + window_id_int = _positive_int(w.get("window_id")) + if pid_int is None or window_id_int is None: continue + z_raw = w.get("z_index") + z_index = z_raw if isinstance(z_raw, (int, float)) and not isinstance(z_raw, bool) else 0 windows.append({ "app_name": w.get("app_name", ""), "pid": pid_int, "window_id": window_id_int, "off_screen": not w.get("is_on_screen", True), "title": w.get("title", ""), - "z_index": w.get("z_index", 0), + "z_index": z_index, }) return windows @@ -1158,6 +1186,9 @@ class CuaDriverBackend(ComputerUseBackend): self._active_pid: Optional[int] = None self._active_window_id: Optional[int] = None self._last_app: Optional[str] = None # last app name targeted via capture/focus_app + # Exact identity for capture_after. App names may be generic on Linux + # (for example, multiple unrelated Qt windows can say Qt6Application). + self._last_target: Optional[Dict[str, Optional[int]]] = None # Surface 6 of NousResearch/hermes-agent#47072: per-snapshot # `element_index -> element_token` map populated on capture(). # Action tools (click/scroll/set_value/...) attach the matching @@ -1243,9 +1274,174 @@ class CuaDriverBackend(ComputerUseBackend): return False return cua_driver_binary_available() + def _clear_active_target(self) -> None: + """Forget a capture/focus target so a failed lookup cannot misroute input.""" + self._active_pid = None + self._active_window_id = None + self._last_app = None + self._last_target = None + self._snapshot_tokens = {} + + 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.""" + try: + 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 "") + ) + return out + + def _load_windows(self) -> List[Dict[str, Any]]: + """Load normalized visible windows, with the shared CLI recovery path. + + Windows are sorted by ``z_index`` **descending**: CUA Driver + defines higher values as closer to the front, so the frontmost + window ends up at index 0 — which is what ``capture()`` and + ``focus_app()`` pick as the default target. ``_ingest_windows`` + already normalised null ``z_index`` (Wayland) to 0, so those + windows sort to the back. + """ + out = self._call_capture_tool( + "list_windows", + {"on_screen_only": True, "session": self._session_id}, + ) + raw_windows = (out.get("structuredContent") or {}).get("windows") or [] + windows = _ingest_windows(raw_windows) + windows.sort(key=lambda w: w["z_index"], reverse=True) + if windows: + return windows + + logger.warning( + "cua-driver list_windows returned no windows over MCP; " + "re-fetching via CLI transport", + ) + try: + cli_out = self._session._call_tool_via_cli( + "list_windows", + {"on_screen_only": True, "session": self._session_id}, + 20.0, + ) + except Exception as exc: + logger.error("cua-driver CLI re-fetch for list_windows failed: %s", exc) + return [] + if cli_out.get("isError") is True: + logger.error("cua-driver CLI re-fetch for list_windows returned an error") + self._clear_active_target() + return [] + raw_windows = (cli_out.get("structuredContent") or {}).get("windows") or [] + windows = _ingest_windows(raw_windows) + windows.sort(key=lambda w: w["z_index"], reverse=True) + return windows + + def _match_windows_for_app( + self, windows: List[Dict[str, Any]], app: str + ) -> List[Dict[str, Any]]: + """Resolve ``app=`` through exact names before convenience substrings. + + Linux ``list_windows`` can omit an app name while ``list_apps`` retains + name/bundle-ID metadata. Exact direct names and exact metadata aliases + must win over substring matches: querying ``Code`` must not silently + select ``Visual Studio Code`` merely because it is frontmost. + """ + app_lower = app.strip().lower() + if not app_lower: + return [] + + direct_exact = [ + w for w in windows + if app_lower == str(w.get("app_name", "")).strip().lower() + ] + if 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 retain the constrained title + # fallback below instead of treating this as a hard no-match. + 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() + for raw_app in running_apps: + if not isinstance(raw_app, dict) or raw_app.get("running") is False: + continue + raw_pid = raw_app.get("pid") + if isinstance(raw_pid, bool) or not isinstance(raw_pid, (int, str)): + continue + try: + pid = int(raw_pid) + except ValueError: + continue + if pid <= 0: + continue + + 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) + + metadata_exact = [w for w in windows if w.get("pid") in exact_pids] + if metadata_exact: + return metadata_exact + + direct_partial = [ + w for w in windows + if app_lower in str(w.get("app_name", "")).lower() + ] + if direct_partial: + return direct_partial + + metadata_partial = [w for w in windows if w.get("pid") in partial_pids] + if metadata_partial: + return metadata_partial + + # 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 str(w.get("app_name", "")).strip() + and app_lower in str(w.get("title", "")).lower() + ] + # ── Capture ──────────────────────────────────────────────────── - def capture(self, mode: str = "som", app: Optional[str] = None) -> CaptureResult: - """Capture the frontmost on-screen window (optionally filtered by app name). + 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. Maps hermes `capture(mode, app)` → cua-driver `list_windows` + `get_window_state` (ax/som) or `screenshot` (vision). @@ -1258,41 +1454,35 @@ class CuaDriverBackend(ComputerUseBackend): # PR's effective minimum (trycua/cua#1961 + #1908) is well past # that, so the fallback is gone — the wrapper now treats the # structured shape as the only contract. - lw_out = self._session.call_tool( - "list_windows", - {"on_screen_only": True, "session": self._session_id}, - ) - - def _windows_from(out: Dict[str, Any]) -> List[Dict[str, Any]]: - raw_ = (out.get("structuredContent") or {}).get("windows") or [] - wins_ = _ingest_windows(raw_) - # Sort by z_index descending (lowest z_index = frontmost on macOS). - wins_.sort(key=lambda w: w["z_index"]) - return wins_ - - windows = _windows_from(lw_out) - - # If the MCP bridge returned an empty/degenerate window list (flaky - # session), re-fetch over the CLI transport before giving up — otherwise - # the caller sees a silent 0x0 capture even though windows exist. - if not windows: - logger.warning( - "cua-driver list_windows returned no windows over MCP; " - "re-fetching via CLI transport", - ) - try: - cli_lw = self._session._call_tool_via_cli( - "list_windows", - {"on_screen_only": True, "session": self._session_id}, - 20.0, + # An exact pid/window pair is both the stable capture_after target and + # the escape hatch when app/window discovery is unavailable on X11. + if pid is not None or window_id is not None: + if pid is None or window_id is None: + return self._failed_capture( + mode, "", ) - windows = _windows_from(cli_lw) - except Exception as cli_exc: - logger.error("cua-driver CLI re-fetch for list_windows failed: %s", cli_exc) - - if not windows: - return CaptureResult(mode=mode, width=0, height=0, png_b64=None, - elements=[], app="", window_title="", png_bytes_len=0) + target_pid = _positive_int(pid) + target_window_id = _positive_int(window_id) + if target_pid is None or target_window_id is None: + return self._failed_capture( + mode, "", + ) + windows = [{ + "app_name": app or "", + "pid": target_pid, + "window_id": target_window_id, + "off_screen": False, + "title": "", + "z_index": 0, + }] + else: + try: + windows = self._load_windows() + except Exception: + self._clear_active_target() + raise + if not windows: + return self._failed_capture(mode) # Filter by app name (case-insensitive substring) if requested. # When the filter matches nothing, surface that explicitly instead of @@ -1300,7 +1490,7 @@ class CuaDriverBackend(ComputerUseBackend): # returned by list_windows is the localized name (e.g. "計算機"), so # `app="Calculator"` legitimately matches no windows on a non-English # system and the caller needs to retry with the localized name. - if app and app.strip().lower() in _SCREEN_CAPTURE_SENTINELS: + if pid is None and window_id is None and app and app.strip().lower() in _SCREEN_CAPTURE_SENTINELS: # Whole-screen / desktop request. cua-driver has no virtual-desktop # capture tool, so resolve to the OS shell/desktop window (the # desktop backdrop or the taskbar/menu-bar), which list_windows @@ -1313,10 +1503,9 @@ class CuaDriverBackend(ComputerUseBackend): desktop = [w for w in windows if _is_desktop_window(w)] if not desktop: - return CaptureResult( - mode=mode, width=0, height=0, png_b64=None, - elements=[], app="", - window_title=( + return self._failed_capture( + mode, + ( f"" ), - png_bytes_len=0, ) # Prefer the desktop backdrop (Progman/WorkerW/Finder) over the # taskbar when both are present, so a bare "screen" capture shows @@ -1336,20 +1524,18 @@ class CuaDriverBackend(ComputerUseBackend): for n in ("progman", "workerw", "program manager", "finder", "desktop") ) else 1, ) - elif app: - app_lower = app.lower() - filtered = [w for w in windows if app_lower in w["app_name"].lower()] + elif pid is None and window_id is None and app: + filtered = self._match_windows_for_app(windows, app) if not filtered: - return CaptureResult( - mode=mode, width=0, height=0, png_b64=None, - elements=[], app="", - window_title=( + return self._failed_capture( + mode, + ( f"" + f"instead of 'Calculator'; some Linux/Qt apps only " + f"resolve via list_apps metadata)>" ), - png_bytes_len=0, ) windows = filtered @@ -1357,11 +1543,18 @@ class CuaDriverBackend(ComputerUseBackend): target = next((w for w in windows if not w["off_screen"]), windows[0]) self._active_pid = target["pid"] self._active_window_id = target["window_id"] + # Tokens belong to the prior window snapshot. Disarm them before any + # capture call so an exception cannot pair old tokens with this target. + self._snapshot_tokens = {} app_name = target["app_name"] # Record the resolved app name so capture_after= follow-ups can 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 + self._last_target = { + "pid": self._active_pid, + "window_id": self._active_window_id, + } # Step 2: capture. png_b64: Optional[str] = None @@ -1390,7 +1583,7 @@ class CuaDriverBackend(ComputerUseBackend): ) sc_out: Optional[Dict[str, Any]] = None if use_screenshot: - sc_out = self._session.call_tool( + sc_out = self._call_capture_tool( "screenshot", { "window_id": self._active_window_id, @@ -1407,7 +1600,7 @@ class CuaDriverBackend(ComputerUseBackend): sc_out = None if sc_out is None: - gws_out = self._session.call_tool( + gws_out = self._call_capture_tool( "get_window_state", { "pid": self._active_pid, @@ -1444,7 +1637,9 @@ class CuaDriverBackend(ComputerUseBackend): }, 30.0, ) - if cli_out.get("images"): + if cli_out.get("isError") is True: + self._clear_active_target() + elif cli_out.get("images"): png_b64 = cli_out["images"][0] image_mime_type = "image/png" except Exception as cli_exc: @@ -1453,7 +1648,7 @@ class CuaDriverBackend(ComputerUseBackend): ) else: # get_window_state: AX tree + screenshot. - gws_out = self._session.call_tool( + gws_out = self._call_capture_tool( "get_window_state", { "pid": self._active_pid, @@ -1499,7 +1694,9 @@ class CuaDriverBackend(ComputerUseBackend): }, 30.0, ) - if not _gws_is_empty(cli_out): + if cli_out.get("isError") is True: + self._clear_active_target() + elif not _gws_is_empty(cli_out): gws_out = cli_out except Exception as cli_exc: logger.error( @@ -1606,8 +1803,12 @@ class CuaDriverBackend(ComputerUseBackend): args["element_index"] = element args["window_id"] = self._active_window_id elif x is not None and y is not None: + if self._active_window_id is None: + return ActionResult(ok=False, action=tool, + message="No active window_id for coordinate click.") args["x"] = x args["y"] = y + args["window_id"] = self._active_window_id else: return ActionResult(ok=False, action=tool, message="click requires element= or x/y.") @@ -1639,8 +1840,12 @@ class CuaDriverBackend(ComputerUseBackend): args["to_element"] = to_element args["window_id"] = self._active_window_id elif from_xy is not None and to_xy is not None: + if self._active_window_id is None: + return ActionResult(ok=False, action="drag", + message="No active window_id for coordinate drag.") args["from_x"], args["from_y"] = int(from_xy[0]), int(from_xy[1]) args["to_x"], args["to_y"] = int(to_xy[0]), int(to_xy[1]) + args["window_id"] = self._active_window_id else: return ActionResult(ok=False, action="drag", message="drag requires from_element/to_element or from_coordinate/to_coordinate.") @@ -1669,21 +1874,36 @@ class CuaDriverBackend(ComputerUseBackend): args["element_index"] = element args["window_id"] = self._active_window_id elif x is not None and y is not None: - args["x"] = x - args["y"] = y + if self._active_window_id is None: + return ActionResult(ok=False, action="scroll", + message="No active window_id for coordinate scroll.") + # CUA Driver 0.7.1 Linux schema rejects x/y on scroll. Only + # include them when the driver explicitly advertises support + # for coordinate scrolling; otherwise omit and let the driver + # scroll the targeted window (window_id is still sent for + # routing). This is the safe default when capabilities + # haven't been discovered yet (older drivers). + if self._session.supports_capability( + "input.scroll.coordinates", tool="scroll" + ): + args["x"] = x + args["y"] = y + args["window_id"] = self._active_window_id return self._action("scroll", args) # ── Keyboard ─────────────────────────────────────────────────── def type_text(self, text: str) -> ActionResult: pid = self._active_pid - if pid is None: + window_id = self._active_window_id + if pid is None or window_id is None: return ActionResult(ok=False, action="type_text", message="No active window — call capture() first.") - return self._action("type_text", {"pid": pid, "text": text}) + return self._action("type_text", {"pid": pid, "window_id": window_id, "text": text}) def key(self, keys: str) -> ActionResult: pid = self._active_pid - if pid is None: + window_id = self._active_window_id + if pid is None or window_id is None: return ActionResult(ok=False, action="key", message="No active window — call capture() first.") @@ -1694,9 +1914,11 @@ class CuaDriverBackend(ComputerUseBackend): if modifiers: # hotkey requires at least one modifier + one key. - return self._action("hotkey", {"pid": pid, "keys": modifiers + [key_name]}) + return self._action("hotkey", {"pid": pid, "window_id": window_id, + "keys": modifiers + [key_name]}) else: - return self._action("press_key", {"pid": pid, "key": key_name}) + return self._action("press_key", {"pid": pid, "window_id": window_id, + "key": key_name}) # ── Value setter ──────────────────────────────────────────────── def set_value(self, value: str, element: Optional[int] = None) -> ActionResult: @@ -1720,12 +1942,17 @@ class CuaDriverBackend(ComputerUseBackend): # ── Introspection ────────────────────────────────────────────── def list_apps(self) -> List[Dict[str, Any]]: out = self._session.call_tool("list_apps", {"session": self._session_id}) - data = out["data"] + structured = out.get("structuredContent") + if isinstance(structured, dict) and isinstance(structured.get("apps"), list): + return structured["apps"] + + # Older drivers and direct CLI fallbacks may put apps in data instead. + data = out.get("data") if isinstance(data, list): return data - if isinstance(data, dict): - return data.get("apps", []) - # list_apps returns plain text — parse app lines. + if isinstance(data, dict) and isinstance(data.get("apps"), list): + return data["apps"] + # Old text-only drivers retain a small, name/PID-only fallback. if isinstance(data, str): apps = [] for line in data.splitlines(): @@ -1735,6 +1962,9 @@ class CuaDriverBackend(ComputerUseBackend): return apps return [] + def list_windows(self) -> List[Dict[str, Any]]: + return self._load_windows() + def focus_app(self, app: str, raise_window: bool = False) -> ActionResult: """Target an app for subsequent actions without stealing system focus. @@ -1748,16 +1978,13 @@ class CuaDriverBackend(ComputerUseBackend): raise_window=True is intentionally ignored: stealing the user's focus is exactly what this backend is designed to avoid. """ - lw_out = self._session.call_tool( - "list_windows", - {"on_screen_only": True, "session": self._session_id}, - ) - raw_windows = (lw_out.get("structuredContent") or {}).get("windows") or [] - windows = _ingest_windows(raw_windows) - windows.sort(key=lambda w: w["z_index"]) + try: + windows = self._load_windows() + except Exception: + self._clear_active_target() + raise - app_lower = app.lower() - matched = [w for w in windows if app_lower in w["app_name"].lower()] + matched = self._match_windows_for_app(windows, app) # Don't silently fall back to the frontmost window when the filter # matches nothing — that hides the real failure (often a localized # macOS app name mismatch, e.g. caller passed "Calculator" but @@ -1766,12 +1993,18 @@ class CuaDriverBackend(ComputerUseBackend): if target: self._active_pid = target["pid"] self._active_window_id = target["window_id"] - self._last_app = target["app_name"] # preserve for capture_after= follow-ups + self._snapshot_tokens = {} + self._last_app = target["app_name"] # retained for back-compat diagnostics + self._last_target = { + "pid": self._active_pid, + "window_id": self._active_window_id, + } 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.", ) + self._clear_active_target() return ActionResult(ok=False, action="focus_app", message=f"No on-screen window found for app '{app}'.") diff --git a/tools/computer_use/schema.py b/tools/computer_use/schema.py index a3394d23276e..613185d83a2c 100644 --- a/tools/computer_use/schema.py +++ b/tools/computer_use/schema.py @@ -44,6 +44,7 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "set_value", "wait", "list_apps", + "list_windows", "focus_app", ], "description": ( @@ -81,6 +82,21 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "one window or display at a time." ), }, + "pid": { + "type": "integer", + "description": ( + "Optional exact process target for action='capture'. Pair " + "with window_id when discovery cannot resolve an X11 app." + ), + }, + "window_id": { + "type": "integer", + "description": ( + "Optional exact native window target for action='capture'. " + "Pair with pid when an external cua-driver list_windows " + "lookup has already identified the window." + ), + }, "max_elements": { "type": "integer", "description": ( @@ -118,9 +134,9 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = { "minItems": 2, "maxItems": 2, "description": ( - "Pixel coordinates [x, y] in logical screen space (as " - "returned by capture width/height). Only use this if " - "no element index is available." + "Pixel coordinates [x, y] relative to the captured window " + "screenshot (top-left origin). Only use this if no element " + "index is available." ), }, "button": { diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index 6d6902169161..fec551b6075b 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -193,8 +193,17 @@ class _NoopBackend(ComputerUseBackend): # pragma: no cover def stop(self) -> None: self._started = False def is_available(self) -> bool: return True - def capture(self, mode: str = "som", app: Optional[str] = None) -> CaptureResult: - self.calls.append(("capture", {"mode": mode, "app": app})) + def capture( + self, + mode: str = "som", + app: Optional[str] = None, + pid: Optional[int] = None, + window_id: Optional[int] = None, + ) -> CaptureResult: + self.calls.append(( + "capture", + {"mode": mode, "app": app, "pid": pid, "window_id": window_id}, + )) return CaptureResult(mode=mode, width=1024, height=768, png_b64=None, elements=[], app=app or "", window_title="") @@ -222,6 +231,10 @@ class _NoopBackend(ComputerUseBackend): # pragma: no cover self.calls.append(("list_apps", {})) return [] + def list_windows(self) -> List[Dict[str, Any]]: + self.calls.append(("list_windows", {})) + return [] + def focus_app(self, app: str, raise_window: bool = False) -> ActionResult: self.calls.append(("focus_app", {"app": app, "raise": raise_window})) return ActionResult(ok=True, action="focus_app") @@ -347,7 +360,13 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> mode = str(args.get("mode", "som")) if mode not in {"som", "vision", "ax"}: return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"}) - cap = backend.capture(mode=mode, app=args.get("app")) + capture_kwargs: Dict[str, Any] = {"mode": mode, "app": args.get("app")} + if args.get("pid") is not None or args.get("window_id") is not None: + capture_kwargs.update({ + "pid": args.get("pid"), + "window_id": args.get("window_id"), + }) + cap = backend.capture(**capture_kwargs) return _capture_response(cap, max_elements=_coerce_max_elements(args.get("max_elements"))) if action == "wait": @@ -359,6 +378,10 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> apps = backend.list_apps() return json.dumps({"apps": apps, "count": len(apps)}) + if action == "list_windows": + windows = backend.list_windows() + return json.dumps({"windows": windows, "count": len(windows)}) + if action == "focus_app": app = args.get("app") if not app: @@ -844,11 +867,16 @@ def _maybe_follow_capture( if not res.ok: return _text_response(res) try: - # Preserve the app context established by the preceding capture/focus_app so - # that capture_after=True re-captures the same app rather than the frontmost - # window (which may have changed if the action caused a focus shift). - last_app = getattr(backend, "_last_app", None) - cap = backend.capture(mode="som", app=last_app) + # Preserve the exact selected window when possible. Linux may expose a + # generic app name for several unrelated windows, so app-only recapture + # can silently switch targets after a successful action. + target = getattr(backend, "_last_target", None) or {} + pid = target.get("pid") + window_id = target.get("window_id") + if pid is not None and window_id is not None: + cap = backend.capture(mode="som", pid=pid, window_id=window_id) + else: + cap = backend.capture(mode="som", app=getattr(backend, "_last_app", None)) except Exception as e: logger.warning("follow-up capture failed: %s", e) return _text_response(res)