diff --git a/native/computer-use-linux/runtime.py b/native/computer-use-linux/runtime.py index 903289b2696..ef217185cf3 100644 --- a/native/computer-use-linux/runtime.py +++ b/native/computer-use-linux/runtime.py @@ -53,6 +53,10 @@ BLOCKED_APP_FRAGMENTS = ( CLIPBOARD_COMMAND_TIMEOUT_SECONDS = 2 CLIPBOARD_OWNER_SETTLE_SECONDS = 0.05 CLIPBOARD_PASTE_SETTLE_SECONDS = 0.15 +ATSPI_METHOD_TIMEOUT_MS = 800 +ATSPI_STARTUP_TIMEOUT_MS = 15_000 +GEOMETRY_BUDGET_SECONDS = 3 +MIN_GEOMETRY_TIMEOUT_MS = 50 @dataclass @@ -113,23 +117,34 @@ def has_state(node, state): return bool(state_set is not None and attempt(lambda: state_set.contains(state), False)) -def screen_rect(node): - component = attempt(node.get_component_iface) +class GeometryReader: + def __init__(self): + self.deadline = time.monotonic() + GEOMETRY_BUDGET_SECONDS + + def read(self, fn): + remaining_ms = int((self.deadline - time.monotonic()) * 1000) + if remaining_ms < MIN_GEOMETRY_TIMEOUT_MS: + return None + Atspi.set_timeout(min(ATSPI_METHOD_TIMEOUT_MS, remaining_ms), -1) + try: + return attempt(fn) + finally: + Atspi.set_timeout(ATSPI_METHOD_TIMEOUT_MS, ATSPI_STARTUP_TIMEOUT_MS) + + def rect(self, node): + return screen_rect(node, self.read) + + +def screen_rect(node, read=attempt): + component = read(node.get_component_iface) if component is None: return None - rect = attempt(lambda: Atspi.Component.get_extents(component, Atspi.CoordType.SCREEN)) + rect = read(lambda: Atspi.Component.get_extents(component, Atspi.CoordType.SCREEN)) if rect is None or rect.width <= 0 or rect.height <= 0: return None return Rect(float(rect.x), float(rect.y), float(rect.width), float(rect.height)) -def relative_rect(node, window_rect): - rect = screen_rect(node) - if rect is None or window_rect is None: - return rect - return Rect(rect.x - window_rect.x, rect.y - window_rect.y, rect.width, rect.height) - - def desktop_apps(): for _, app in children(desktop_root()): if name_of(app): @@ -140,8 +155,8 @@ def windows_for(app): result = [] for index, child in children(app): role = role_of(child).lower() - rect = screen_rect(child) - if rect is not None or role in {"frame", "window", "dialog", "alert"}: + standard_role = role in {"frame", "window", "dialog", "alert"} + if standard_role or screen_rect(child) is not None: result.append((index, child)) return result @@ -332,9 +347,8 @@ def accessible_id(node): return str(attempt(node.get_accessible_id, "") or "") -def record(node, index, path, window_rect): +def record(node, index, path): role = role_of(node) - rect = relative_rect(node, window_rect) return { "index": index, "runtimeId": path, @@ -346,12 +360,12 @@ def record(node, index, path, window_rect): "value": string_value(node), "isSelected": has_state(node, Atspi.StateType.SELECTED), "nativeWindowHandle": 0, - "frame": rect.to_json() if rect else None, + "frame": None, "actions": action_labels(node), } -def render_accessibility_tree(root, window_rect, root_path, compact_browser_tabs=False): +def render_accessibility_tree(root, root_path, compact_browser_tabs=False, geometry_nodes=None): records = [] lines = [] truncation = {"truncated": False, "maxNodes": MAX_NODES, "maxDepth": MAX_DEPTH, "maxDepthReached": False} @@ -410,7 +424,7 @@ def render_accessibility_tree(root, window_rect, root_path, compact_browser_tabs if depth > MAX_DEPTH: truncation["maxDepthReached"] = True return - item = record(node, len(records), path, window_rect) + item = record(node, len(records), path) role_key = (item["controlType"] or "").lower() title = item["name"] or item["automationId"] or "" suppressed = suppress_children(role_key, title, item["value"], None) @@ -425,6 +439,8 @@ def render_accessibility_tree(root, window_rect, root_path, compact_browser_tabs walk(child, depth, path + [child_index]) return records.append(item) + if geometry_nodes is not None: + geometry_nodes[id(item)] = node role_label = item["localizedControlType"] or item["controlType"] line = f'{item["index"]} {role_label} {sanitize_text(title)}'.rstrip() line += formatted_value(role_key, title, item["value"]) @@ -662,17 +678,29 @@ def make_snapshot(query, include_screenshot, window_id=None, window_index=None, restore_window(app, window) window_index, window = choose_window(app, window_id, window_index) bounds = screen_rect(window) + geometry_nodes = {} records, lines, truncation = render_accessibility_tree( window, - bounds, [window_index], compact_browser_tabs=is_browser_app(query, app), + geometry_nodes=geometry_nodes, ) + focused = focused_summary(window) + selection = selected_text(window) + app_info = app_json(app) + window_title = name_of(window) screenshot = capture_png(bounds) if include_screenshot else None + geometry = GeometryReader() + for item in records: + node = geometry_nodes[id(item)] + rect = bounds if node is window else geometry.rect(node) + if rect is not None and bounds is not None: + rect = Rect(rect.x - bounds.x, rect.y - bounds.y, rect.width, rect.height) + item["frame"] = rect.to_json() if rect else None return { "snapshotId": str(uuid.uuid4()), - "app": app_json(app), - "windowTitle": name_of(window), + "app": app_info, + "windowTitle": window_title, "windowId": None, "windowIndex": window_index, "windowBounds": bounds.to_json() if bounds else None, @@ -684,8 +712,8 @@ def make_snapshot(query, include_screenshot, window_id=None, window_index=None, "coordinateSpace": "window", "truncation": truncation, "treeLines": lines, - "focusedSummary": focused_summary(window), - "selectedText": selected_text(window), + "focusedSummary": focused, + "selectedText": selection, "elements": records, } diff --git a/native/computer-use-linux/runtime_render_test.py b/native/computer-use-linux/runtime_render_test.py index d3672912ca6..37c1935fa2f 100644 --- a/native/computer-use-linux/runtime_render_test.py +++ b/native/computer-use-linux/runtime_render_test.py @@ -1,13 +1,19 @@ +from contextlib import ExitStack import importlib.util +import json import sys import types import unittest from pathlib import Path +from unittest.mock import patch class FakeStateType: PROTECTED = 1 SELECTED = 2 + ACTIVE = 3 + SHOWING = 4 + FOCUSED = 5 class FakeText: @@ -103,10 +109,175 @@ class FailingChildrenAccessible(FakeAccessible): def render(root): - return runtime.render_accessibility_tree(root, None, [0]) + return runtime.render_accessibility_tree(root, [0]) + + +class SnapshotFixture(ExitStack): + def __init__(self, root, app_name="Fixture"): + super().__init__() + self.root = root + self.app = FakeAccessible("application", app_name, children=[root]) + self.app.get_process_id = lambda: 42 + self.clock = 0.0 + self.timeout = (800, 15000) + self.delays = {} + self.stalled = [] + self.captured_at = None + self.restore = lambda: None + self.bounds = {node: runtime.Rect(100 + i * 10, 200 + i * 20, 300, 400) + for i, node in enumerate(runtime.all_nodes(root))} + for node in self.bounds: + node.get_component_iface = lambda node=node: node + + def __enter__(self): + for obj, name, value in ( + (runtime, "desktop_apps", lambda: iter([self.app])), + (runtime.time, "monotonic", lambda: self.clock), + (runtime.Atspi, "set_timeout", self.set_timeout), + (runtime.Atspi, "Component", types.SimpleNamespace(get_extents=self.extents)), + (runtime.Atspi, "CoordType", types.SimpleNamespace(SCREEN=0)), + (runtime, "capture_png", self.capture), + (runtime, "restore_window", lambda *_: self.restore()), + ): + self.enter_context(patch.object(obj, name, value, create=True)) + return self + + def set_timeout(self, method, startup): + self.timeout = (method, startup) + + def extents(self, node, _): + method, startup = self.timeout + allowance = max(method, startup - self.clock * 1000) / 1000 + delay = self.delays.get(node, 0) + self.clock += min(delay, allowance) + if delay > allowance: + self.stalled.append(node.name) + raise TimeoutError("unresponsive geometry") + return self.bounds.get(node) + + def capture(self, bounds): + self.clock += 1 + self.captured_at = self.clock + return {"base64": json.dumps(bounds.to_json())} if bounds else None class RuntimeRenderTest(unittest.TestCase): + def test_snapshot_keeps_semantics_and_image_when_child_geometry_stalls(self): + late = FakeAccessible("button", "Continue") + root = FakeAccessible("frame", "Editor", children=[ + FakeAccessible("button", "Healthy"), + *[FakeAccessible("button", f"Slow {i}") for i in range(4)], late, + ]) + for semantic_delay, stall_children in ((3.1, False), (16, True)): + with self.subTest(semantic_delay=semantic_delay), SnapshotFixture(root) as fixture: + late.read = False + fixture.delays = {root: 1, **{node: 20 if stall_children else 0 for node in root.children[1:-1]}} + def delayed_name(): + if not getattr(late, "read", False): + fixture.clock += semantic_delay + late.read = True + return "" if fixture.stalled else "Continue" + late.get_name = delayed_name + late.get_n_actions = lambda: 1 + late.get_action_name = lambda _: "click" + snapshot = runtime.make_snapshot("Fixture", True, window_index=0) + self.assertLessEqual(fixture.clock - fixture.captured_at, 3) + self.assertEqual(fixture.timeout, (800, 15000)) + elements = snapshot["elements"] + self.assertEqual([item["name"] for item in elements], [root.name, *[n.name for n in root.children]]) + self.assertEqual([item["index"] for item in elements], list(range(7))) + self.assertEqual(elements[-1]["runtimeId"], [0, 5]) + self.assertEqual(elements[-1]["actions"], ["click"]) + self.assertEqual(snapshot["treeLines"][-1], "\t6 button Continue") + self.assertFalse(snapshot["truncation"]["truncated"]) + self.assertEqual(snapshot["windowBounds"], fixture.bounds[root].to_json()) + self.assertEqual(json.loads(snapshot["screenshotPngBase64"]), fixture.bounds[root].to_json()) + self.assertEqual(elements[0]["frame"], runtime.Rect(0, 0, 300, 400).to_json()) + self.assertEqual(elements[1]["frame"], runtime.Rect(10, 20, 300, 400).to_json()) + self.assertTrue(all((item["frame"] is None) == stall_children for item in elements[2:])) + + def test_nonstandard_window_selection_and_fresh_restored_capture(self): + for query, index, delay, restore in (("Fixture", None, 1, False), + ("pid:42", 1, 1, False), ("Editor", 1, 0, True)): + with self.subTest(query=query), SnapshotFixture( + FakeAccessible("panel", "Editor", children=[FakeAccessible("button", "Save")]) + ) as fixture: + root = fixture.root + fixture.app.children.insert(0, FakeAccessible("label", "Unrelated")) + fixture.delays[root] = delay + def move(): + fixture.bounds[root] = runtime.Rect(500, 200, 300, 400) + fixture.bounds[root.children[0]] = runtime.Rect(510, 220, 300, 400) + fixture.restore = move + snapshot = runtime.make_snapshot(query, True, window_index=index, restore=restore) + self.assertEqual(snapshot["windowIndex"], 1) + self.assertEqual(snapshot["windowBounds"]["x"], 500 if restore else 100) + self.assertEqual(snapshot["elements"][1]["frame"]["x"], 10) + self.assertEqual(json.loads(snapshot["screenshotPngBase64"]), fixture.bounds[root].to_json()) + + def test_window_only_protection_preserves_completed_semantic_action(self): + button = FakeAccessible("button", "Continue") + effects = [] + button.get_n_actions = lambda: 1 + button.get_action_name = lambda _: "click" + def click(_): + effects.append("clicked") + button.value = "Completed" + return True + button.do_action = click + with SnapshotFixture(FakeAccessible("frame", "Editor", children=[button])) as fixture: + fixture.delays[button] = 20 + snapshot = runtime.make_snapshot("Fixture", False) + self.assertIsNone(snapshot["elements"][1]["frame"]) + fixture.app.children.append(FakeAccessible("label", "Bitwarden integration")) + result = runtime.run_operation({"tool": "click", "app": "Fixture", "windowIndex": 0, + "element": snapshot["elements"][1], "noScreenshot": True}) + self.assertEqual(effects, ["clicked"]) + self.assertTrue(result["ok"]) + self.assertEqual(result["action"]["path"], "accessibility") + self.assertEqual(result["snapshot"]["elements"][1]["value"], "Completed") + fixture.root.name = "Bitwarden vault" + fixture.bounds[fixture.root] = None + with self.assertRaisesRegex(RuntimeError, "appBlocked"): + runtime.run_operation({"tool": "click", "app": "Fixture"}) + self.assertEqual(effects, ["clicked"]) + + def test_compacted_browser_records_keep_their_own_frames(self): + tabs = [FakeAccessible("page tab", f"Tab {i}", value="1" if i == 9 else "") for i in range(10)] + buttons = [FakeAccessible("button", f"Other {i}") for i in range(10)] + root = FakeAccessible("frame", "Browser", children=[ + FakeAccessible("page tab list", "Tabs", children=tabs), *buttons, + ]) + with SnapshotFixture(root, "chrome") as fixture: + fixture.delays = {tab: 20 for tab in tabs[:-1]} + snapshot = runtime.make_snapshot("chrome", False) + by_name = {item["name"]: item for item in snapshot["elements"]} + for node in (tabs[-1], buttons[8]): + self.assertEqual(by_name[node.name]["frame"]["x"], fixture.bounds[node].x - 100) + self.assertNotIn("Tab 0", by_name) + + def test_geometry_skips_insufficient_budget_and_restores_action_timeout(self): + clock = [0.0] + timeouts = [] + + def fail(): + raise TimeoutError("unresponsive geometry") + + with patch.object(runtime.time, "monotonic", side_effect=lambda: clock[0]), \ + patch.object(runtime.Atspi, "set_timeout", side_effect=lambda *args: timeouts.append(args), create=True): + reader = runtime.GeometryReader() + clock[0] = 2.96 + calls = [] + self.assertIsNone(reader.read(lambda: calls.append("called"))) + self.assertEqual(calls, []) + self.assertEqual(timeouts, []) + clock[0] = 2.9 + self.assertIsNone(reader.read(fail)) + self.assertGreaterEqual(timeouts[0][0], 50) + self.assertLessEqual(timeouts[0][0], 100) + self.assertEqual(timeouts[0][1], -1) + self.assertEqual(timeouts[-1], (800, 15000)) + def test_named_non_generic_node_skips_unused_summary_walk(self): root = FakeAccessible("button", "Save", children=[FakeAccessible("text", "unused")])