diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index 0e05a830a1f..d6e98c691c2 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -1532,6 +1532,195 @@ def _make_cua_backend_with_windows_and_apps( return backend +def _make_cua_backend_with_tool_result(result: Dict[str, Any]): + from tools.computer_use.cua_backend import CuaDriverBackend + + backend = CuaDriverBackend() + backend._session = MagicMock() + backend._session.call_tool.return_value = result + return backend + + +class TestCuaDriverWindowResultShapes: + def test_extracts_windows_from_structured_content(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}] + + assert _windows_from_tool_result({ + "structuredContent": {"windows": windows}, + "data": {}, + }) == windows + + def test_extracts_windows_from_data_windows(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}] + + assert _windows_from_tool_result({ + "structuredContent": None, + "data": {"windows": windows}, + }) == windows + + def test_extracts_windows_from_legacy_data_windows(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}] + + assert _windows_from_tool_result({ + "structuredContent": None, + "data": {"_legacy_windows": windows}, + }) == windows + + def test_empty_structured_windows_falls_through_to_data_windows(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}] + + assert _windows_from_tool_result({ + "structuredContent": {"windows": []}, + "data": {"windows": windows}, + }) == windows + + def test_extracts_windows_from_top_level_windows(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}] + + assert _windows_from_tool_result({"windows": windows}) == windows + + def test_extracts_windows_from_top_level_legacy_windows(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + windows = [{"app_name": "Terminal", "pid": 1, "window_id": 2}] + + assert _windows_from_tool_result({"_legacy_windows": windows}) == windows + + def test_extract_windows_missing_fields_returns_empty(self): + from tools.computer_use.cua_backend import _windows_from_tool_result + + assert _windows_from_tool_result({ + "structuredContent": None, + "data": {}, + }) == [] + + def test_ingest_windows_skips_malformed_members(self): + from tools.computer_use.cua_backend import _ingest_windows + + valid = {"app_name": "Terminal", "pid": 100, "window_id": 7} + + assert _ingest_windows([None, "bad", [], valid]) == [ # type: ignore[list-item] + {"app_name": "Terminal", "pid": 100, "window_id": 7, + "off_screen": False, "title": "", "z_index": 0}, + ] + + def test_ingest_windows_normalizes_untrusted_display_fields(self): + from tools.computer_use.cua_backend import _ingest_windows + + assert _ingest_windows([{ + "app_name": None, "pid": "100", "window_id": "7", + "title": ["bad"], "z_index": "bad", + }]) == [{ + "app_name": "", "pid": 100, "window_id": 7, + "off_screen": False, "title": "", "z_index": 0, + }] + + def test_capture_uses_data_windows_shape(self): + windows = [ + {"app_name": "Terminal", "pid": 100, "window_id": 7, + "is_on_screen": True, "title": "shell", "z_index": 0}, + ] + backend = _make_cua_backend_with_tool_result({ + "data": {"windows": windows}, + "images": [], + "isError": False, + "structuredContent": None, + }) + backend._session.call_tool.side_effect = [ + {"data": {"windows": windows}, "images": [], "isError": False, + "structuredContent": None}, + {"data": "ok", "images": [], "isError": False, "structuredContent": None}, + ] + + cap = backend.capture(mode="ax", app="Terminal") + + assert cap.app == "Terminal" + assert backend._active_pid == 100 + assert backend._active_window_id == 7 + + def test_focus_app_uses_legacy_data_windows_shape(self): + windows = [ + {"app_name": "Terminal", "pid": 100, "window_id": 7, + "is_on_screen": True, "title": "shell", "z_index": 0}, + ] + backend = _make_cua_backend_with_tool_result({ + "data": {"_legacy_windows": windows}, + "images": [], + "isError": False, + "structuredContent": None, + }) + + res = backend.focus_app("Terminal") + + assert res.ok is True + assert backend._active_pid == 100 + assert backend._active_window_id == 7 + + def test_list_apps_accepts_top_level_windows_without_data(self): + windows = [ + {"app_name": "Terminal", "pid": 100, "window_id": 7}, + {"app_name": "Notes", "pid": 200, "window_id": 9}, + ] + backend = _make_cua_backend_with_tool_result({ + "windows": windows, + "images": [], + "isError": False, + }) + + assert backend.list_apps() == [ + {"name": "Terminal", "pid": 100}, + {"name": "Notes", "pid": 200}, + ] + + def test_list_apps_accepts_top_level_legacy_windows_without_data(self): + windows = [{"app_name": "Terminal", "pid": 100, "window_id": 7}] + backend = _make_cua_backend_with_tool_result({ + "_legacy_windows": windows, + "images": [], + "isError": False, + }) + + assert backend.list_apps() == [{"name": "Terminal", "pid": 100}] + + def test_list_apps_prefers_structured_apps_over_data_apps(self): + backend = _make_cua_backend_with_tool_result({ + "structuredContent": {"apps": [{"name": "Canonical", "pid": 1}]}, + "data": {"apps": [{"name": "Stale", "pid": 2}]}, + "images": [], + "isError": False, + }) + + assert backend.list_apps() == [{"name": "Canonical", "pid": 1}] + + def test_list_apps_derives_apps_from_data_windows_shape(self): + windows = [ + {"app_name": "Terminal", "pid": 100, "window_id": 7}, + {"app_name": "Terminal", "pid": 100, "window_id": 8}, + {"app_name": "Notes", "pid": 200, "window_id": 9}, + ] + backend = _make_cua_backend_with_tool_result({ + "data": {"windows": windows}, + "images": [], + "isError": False, + "structuredContent": None, + }) + + assert backend.list_apps() == [ + {"name": "Terminal", "pid": 100}, + {"name": "Notes", "pid": 200}, + ] + + class TestCuaDriverSessionReconnect: """Verify reconnect-once on a closed-resource error. After the lifecycle-owner refactor (Sun Jun 21 2026) the session no longer goes diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index dc084d56351..9d09a4e2657 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -1590,25 +1590,72 @@ def _ingest_windows(raw_windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ windows: List[Dict[str, Any]] = [] for w in raw_windows: + # Compatibility envelopes are untrusted input: skip non-dict members + # instead of raising AttributeError on one malformed record. + if not isinstance(w, dict): + continue 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 + app_name = w.get("app_name", "") + title = w.get("title", "") windows.append({ - "app_name": w.get("app_name", ""), + "app_name": app_name if isinstance(app_name, str) else "", "pid": pid_int, "window_id": window_id_int, # cua-driver 0.6.x on Linux may return JSON null here. # Only explicit False means off-screen; null means unknown. "off_screen": w.get("is_on_screen") is False, - "title": w.get("title", ""), + "title": title if isinstance(title, str) else "", "z_index": z_index, }) return windows +def _windows_from_tool_result(out: Dict[str, Any]) -> List[Dict[str, Any]]: + """Return list_windows payloads across cua-driver result shapes.""" + structured = out.get("structuredContent") + if isinstance(structured, dict): + windows = structured.get("windows") + if isinstance(windows, list) and windows: + return windows + + data = out.get("data") + if isinstance(data, dict): + windows = data.get("windows") + if isinstance(windows, list) and windows: + return windows + legacy_windows = data.get("_legacy_windows") + if isinstance(legacy_windows, list) and legacy_windows: + return legacy_windows + + windows = out.get("windows") + if isinstance(windows, list) and windows: + return windows + legacy_windows = out.get("_legacy_windows") + if isinstance(legacy_windows, list) and legacy_windows: + return legacy_windows + return [] + + +def _apps_from_windows(windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + apps: List[Dict[str, Any]] = [] + seen: set[tuple[str, int]] = set() + for summary in _ingest_windows(windows): + name = summary["app_name"] + if not name: + continue + key = (name, summary["pid"]) + if key in seen: + continue + seen.add(key) + apps.append({"name": name, "pid": summary["pid"]}) + return apps + + # --------------------------------------------------------------------------- # The backend itself # --------------------------------------------------------------------------- @@ -1784,8 +1831,7 @@ class CuaDriverBackend(ComputerUseBackend): "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 = _ingest_windows(_windows_from_tool_result(out)) windows.sort(key=lambda w: w["z_index"], reverse=True) if windows: return windows @@ -1807,8 +1853,7 @@ class CuaDriverBackend(ComputerUseBackend): 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 = _ingest_windows(_windows_from_tool_result(cli_out)) windows.sort(key=lambda w: w["z_index"], reverse=True) return windows @@ -2478,23 +2523,37 @@ class CuaDriverBackend(ComputerUseBackend): def list_apps(self) -> List[Dict[str, Any]]: out = self._session.call_tool("list_apps", {"session": self._session_id}) 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): + + # structuredContent is the canonical MCP payload. Empty lists fall + # through so a populated compatibility envelope can still recover. + if isinstance(structured, dict): + apps = structured.get("apps") + if isinstance(apps, list) and apps: + return apps + # Older drivers and direct CLI fallbacks may put apps in data instead. + if isinstance(data, list) and data: return data - if isinstance(data, dict) and isinstance(data.get("apps"), list): - return data["apps"] + if isinstance(data, dict): + apps = data.get("apps") + if isinstance(apps, list) and apps: + return apps + apps = out.get("apps") + if isinstance(apps, list) and apps: + return apps + + derived = _apps_from_windows(_windows_from_tool_result(out)) + if derived: + return derived + # Old text-only drivers retain a small, name/PID-only fallback. if isinstance(data, str): - apps = [] + parsed_apps = [] for line in data.splitlines(): m = re.search(r'(.+?)\s+\(pid\s+(\d+)\)', line) if m: - apps.append({"name": m.group(1).strip(), "pid": int(m.group(2))}) - return apps + parsed_apps.append({"name": m.group(1).strip(), "pid": int(m.group(2))}) + return parsed_apps return [] def list_windows(self) -> List[Dict[str, Any]]: