mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 15:55:43 +00:00
fix(computer-use): target Linux app windows reliably (#63725)
* fix(computer-use): target Linux app windows reliably Resolve app filters through the canonical cua-driver MCP app metadata and join running app PIDs back to windows. Preserve an exact selected window across capture_after, support direct capture by pid/window_id, and send the active window ID for coordinate pointer actions on Linux. Co-authored-by: annguyenNous <annguyenNous@users.noreply.github.com> Co-authored-by: grimmjoww578 <willies578@gmail.com> Co-authored-by: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> * fix(computer-use): address review on PR #63725 Address three review comments from @f-trycua: 1. type_text, press_key, and hotkey now carry _active_window_id and fail closed when it is missing. Previously they sent only the PID, so CUA Driver fell back to the first window for that PID — input could reach the wrong window in multi-window apps. 2. Coordinate scroll x/y are now capability-gated behind input.scroll.coordinates. CUA Driver 0.7.1 Linux schema rejects x/y on scroll; omitting them when the driver doesn't advertise support avoids the schema rejection while still routing via window_id. 3. Windows are sorted by z_index descending (higher = front, per CUA Driver semantics) instead of ascending. Null z_index (Wayland) is coerced to 0 in _ingest_windows so it doesn't crash the sort and sorts to the back instead of being selected as the capture target. --------- Co-authored-by: LeonSGP43 <cine.dreamer.one@gmail.com> Co-authored-by: annguyenNous <annguyenNous@users.noreply.github.com> Co-authored-by: grimmjoww578 <willies578@gmail.com> Co-authored-by: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com>
This commit is contained in:
kshitij
GitHub
parent
58033baba2
commit
2b6897f982
6 changed files with 1085 additions and 101 deletions
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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, "<capture targeting requires both pid and window_id>",
|
||||
)
|
||||
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, "<capture targeting requires positive integer pid and window_id>",
|
||||
)
|
||||
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"<no desktop/shell window found for app={app!r}; "
|
||||
f"cua-driver captures one window at a time and exposes "
|
||||
f"no whole-virtual-desktop or per-monitor capture. "
|
||||
|
|
@ -1324,7 +1513,6 @@ class CuaDriverBackend(ComputerUseBackend):
|
|||
f"specific window instead. On Windows the taskbar is "
|
||||
f"'Shell_TrayWnd' and the desktop is 'Progman'.>"
|
||||
),
|
||||
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"<no on-screen window matched app={app!r}; "
|
||||
f"call list_apps to see available app names "
|
||||
f"call list_apps to see available app names or bundle IDs "
|
||||
f"(macOS reports localized names, e.g. '計算機' "
|
||||
f"instead of 'Calculator')>"
|
||||
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}'.")
|
||||
|
||||
|
|
|
|||
|
|
@ -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": {
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue