fix(computer-use): normalize cua-driver result envelopes

cua-driver 0.7.x can return list_windows/list_apps payloads under
structuredContent.windows, data.windows, data._legacy_windows, or
top-level windows/_legacy_windows (direct CLI responses). The wrapper
only read structuredContent.windows, so discovery came back empty
(capture 0x0, list_apps []) while raw cua-driver calls worked.

- add _windows_from_tool_result(): walks the known envelope shapes in
  priority order, skipping empty higher-priority envelopes
- route _load_windows() (MCP + CLI re-fetch paths) through the helper,
  covering capture() and focus_app()
- harden _ingest_windows(): skip non-dict members, normalize untrusted
  app_name/title/z_index fields
- list_apps(): prefer structuredContent.apps, fall through populated
  data/top-level envelopes, derive unique apps from window-shaped
  payloads via _apps_from_windows(), keep the text-line fallback last
- tests for every envelope shape, precedence, malformed records, and
  app derivation

Salvaged from #63037 (Reaper-Legion), which itself preserved the
original implementation from #57961 (kohoj); #73007 (umi008)
independently proposed the same normalization later.

Fixes #57905

Co-authored-by: Reaper <248977840+Reaper-Forge@users.noreply.github.com>
Co-authored-by: Ulises Millan Guerrero <ulises.millanguerrero@gmail.com>

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

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]]: