观察与 AX 树 深度解析
这一页只看 macOS Harness 如何把运行中 App 转成模型可读状态:应用解析、权限预检、窗口选择、AX 树快照和截图缩放。
导入链与契约
观察层入口来自 src/macos_harness/cli.py:120 到 src/macos_harness/cli.py:139:CLI 的 see/state 子命令最终都落到 MacOS 实例。MacOS 初始化在 src/macos_harness/macos.py:271 到 src/macos_harness/macos.py:288 创建私有 event source、overlay 和 Accessibility facade;这意味着“观察”和“输入”共享同一个对象状态,最近截图、最近 app 和 element cache 都能连续使用。
核心实现,逐段拆
▸App 解析先消除歧义
App 匹配在 src/macos_harness/macos.py:375 到 src/macos_harness/macos.py:401,它同时比较 PID、localizedName、bundleIdentifier 和 path。关键是 src/macos_harness/macos.py:393 到 src/macos_harness/macos.py:400:多匹配直接抛错,而不是随便选第一个。这对 Agent 很重要,因为错误点击到同名窗口比失败更危险。
375 def _resolve_app(self, query: str | None) -> tuple[Any, dict[str, Any]]: 376 if not query and self._last_app: 377 query = str(self._last_app["pid"]) 378 if not query: 379 raise MacOSError("Specify an app name, bundle ID, path, or PID") 380 381 needle = str(query).casefold() 382 candidates: list[tuple[Any, dict[str, Any]]] = [] 383 exact: list[tuple[Any, dict[str, Any]]] = [] 384 for app in NSWorkspace.sharedWorkspace().runningApplications(): 385 info = self._app_info(app) 386 values = [str(info["pid"]), info["name"], info["bundle_id"], info["path"]] 387 lowered = [str(value).casefold() for value in values if value] 388 if needle in lowered: 389 exact.append((app, info)) 390 elif any(needle in value for value in lowered): 391 candidates.append((app, info)) 392 393 matches = exact or candidates 394 if not matches: 395 raise MacOSError(f"No running application matches {query!r}") 396 if len(matches) > 1: 397 names = ", ".join( 398 f"{item[1]['name']} ({item[1]['pid']})" for item in matches[:8] 399 ) 400 raise MacOSError(f"Application query {query!r} is ambiguous: {names}") 401 return matches[0]
▸AX 快照有深度和节点预算
快照实现 src/macos_harness/macos.py:536 到 src/macos_harness/macos.py:613 每次重置 cache,并在 src/macos_harness/macos.py:555 到 src/macos_harness/macos.py:564 控制深度、数量和重复元素。src/macos_harness/macos.py:566 到 src/macos_harness/macos.py:600 说明它不是盲目 dump,而是选择 JSON-safe 属性、actions 和 settable 字段。
536 def _snapshot_tree( 537 self, 538 root: Any, 539 *, 540 max_depth: int, 541 max_nodes: int, 542 include_menu_bar: bool, 543 extra_attributes: Iterable[str] = (), 544 include_actions: bool = True, 545 include_settable: bool = True, 546 ) -> list[dict[str, Any]]: 547 self._elements = {} 548 nodes: list[dict[str, Any]] = [] 549 seen: set[int] = set() 550 requested_attributes = tuple( 551 dict.fromkeys((*_AX_ATTRIBUTES, *(str(item) for item in extra_attributes))) 552 ) 553 standard_attributes = set(_AX_ATTRIBUTES) 554 555 def visit(element: Any, depth: int) -> None: 556 if depth > max_depth or len(nodes) >= max_nodes: 557 return 558 try: 559 identity = hash(element) 560 except TypeError: 561 identity = id(element) 562 if identity in seen: 563 return 564 seen.add(identity) 565 566 raw = self._copy_attributes(element, requested_attributes) 567 if raw["AXRole"] == "AXMenuBar" and not include_menu_bar: 568 return 569 index = len(nodes) 570 self._elements[index] = element 571 node: dict[str, Any] = { 572 "element_index": index, 573 "depth": depth, 574 "role": self._jsonable(raw["AXRole"]), 575 } 576 for source, target in _AX_NODE_MAPPING.items(): 577 value = self._jsonable(raw[source]) 578 if value not in (None, "", [], {}): 579 node[target] = value 580 extra = { 581 name: self._jsonable(raw[name]) 582 for name in requested_attributes 583 if name not in standard_attributes 584 and self._jsonable(raw[name]) not in (None, "", [], {}) 585 } 586 if extra: 587 node["attributes"] = extra 588 if include_actions: 589 actions = self._actions(element) 590 if actions: 591 node["actions"] = actions 592 if include_settable: 593 settable = [ 594 name 595 for name in _SETTABLE_CANDIDATES 596 if self._settable(element, name) 597 ] 598 if settable: 599 node["settable"] = settable 600 nodes.append(node) 601 602 children = raw["AXChildren"] 603 if not children and depth == 0: 604 children = raw["AXWindows"] 605 if isinstance(children, Iterable) and not isinstance( 606 children, (str, bytes, dict) 607 ): 608 for child in children: 609 if self._is_ax_element(child): 610 visit(child, depth + 1) 611 612 visit(root, 0) 613 return nodes
▸截图与指针分两阶段
截图阶段 src/macos_harness/macos.py:968 到 src/macos_harness/macos.py:1031 负责拿真实窗口图和 scale;see 阶段 src/macos_harness/macos.py:1033 到 src/macos_harness/macos.py:1126 才决定是否缩放和绘制虚拟指针。这个拆分让坐标不会因为图片压缩而失真,测试 tests/test_macos.py:442 到 tests/test_macos.py:502 专门覆盖了绘制和隐藏指针两种情况。
1033 def see( 1034 self, 1035 app: str | None = None, 1036 *, 1037 window_index: int = 0, 1038 path: str | Path | None = None, 1039 max_width: int = 1280, 1040 max_height: int = 1280, 1041 show_pointer: bool = True, 1042 ) -> dict[str, Any]: 1043 """Capture a bounded window image and draw the harness pointer onto it.""" 1044 if max_width <= 0 or max_height <= 0: 1045 raise MacOSError("max_width and max_height must be positive") 1046 screenshot = self.capture_screenshot(app, window_index=window_index, path=path) 1047 output = Path(screenshot["path"]) 1048 raw_width = int(screenshot["width"]) 1049 raw_height = int(screenshot["height"]) 1050 1051 with Image.open(output) as source: 1052 image = source.convert("RGBA") 1053 ratio = min(1.0, max_width / image.width, max_height / image.height) 1054 if ratio < 1.0: 1055 size = ( 1056 max(1, round(image.width * ratio)), 1057 max(1, round(image.height * ratio)), 1058 ) 1059 image = image.resize(size, Image.Resampling.LANCZOS) 1060 1061 bounds = screenshot["bounds"] 1062 scale_x = image.width / float(bounds["width"]) 1063 scale_y = image.height / float(bounds["height"]) 1064 pointer = self._pointer_position 1065 pointer_info: dict[str, Any] | None = None 1066 if pointer is not None: 1067 image_x = (pointer[0] - float(bounds["x"])) * scale_x 1068 image_y = (pointer[1] - float(bounds["y"])) * scale_y 1069 inside = 0 <= image_x < image.width and 0 <= image_y < image.height 1070 pointer_info = { 1071 "screen": {"x": pointer[0], "y": pointer[1]}, 1072 "image": {"x": image_x, "y": image_y}, 1073 "inside": inside, 1074 "visible": self._overlay.visible, 1075 } 1076 if show_pointer and self._overlay.visible and inside: 1077 draw = ImageDraw.Draw(image) 1078 hot_x, hot_y = POINTER_HOTSPOT 1079 points = [ 1080 ( 1081 round(image_x + (x - hot_x) * scale_x), 1082 round(image_y + (y - hot_y) * scale_y), 1083 ) 1084 for x, y in pointer_points() 1085 ] 1086 shadow = [ 1087 (x + max(1, round(scale_x)), y + max(1, round(scale_y))) 1088 for x, y in points 1089 ] 1090 draw.polygon(shadow, fill=(0, 0, 0, 72)) 1091 outline_width = max(2, round(2.5 * (scale_x + scale_y) / 2)) 1092 draw.polygon( 1093 points, 1094 fill=(0, 0, 0, 255), 1095 outline=(255, 255, 255, 240), 1096 width=outline_width, 1097 ) 1098 draw.line( 1099 [*points, points[0]], 1100 fill=(0, 0, 0, 230), 1101 width=max(1, round(0.65 * (scale_x + scale_y) / 2)), 1102 joint="curve", 1103 ) 1104 1105 image.save(output, format="PNG", optimize=True) 1106 frontmost = self._frontmost_app() 1107 screenshot.update( 1108 { 1109 "raw_width": raw_width, 1110 "raw_height": raw_height, 1111 "width": image.width, 1112 "height": image.height, 1113 "scale_x": scale_x, 1114 "scale_y": scale_y, 1115 "virtual_pointer": pointer_info, 1116 "focus": { 1117 "frontmost": frontmost, 1118 "target_is_frontmost": ( 1119 frontmost is not None 1120 and int(frontmost["pid"]) == int(screenshot["pid"]) 1121 ), 1122 }, 1123 } 1124 ) 1125 self._last_screenshot = screenshot 1126 return screenshot
关键流程
这条流程解释了为什么 src/macos_harness/macos.py:642 到 src/macos_harness/macos.py:687 返回的是 app、nodes、text、windows 和 screenshot,而不是单张图片。模型可以先读语义树,再需要时看图;如果 screenshot 失败,src/macos_harness/macos.py:676 到 src/macos_harness/macos.py:684 也会保留错误而不是丢掉 AX 状态。
边界 / 失败模式
src/macos_harness/macos.py:394到src/macos_harness/macos.py:400遇到模糊 App 名会停止,调用方必须给 PID 或更精确名称。src/macos_harness/macos.py:978到src/macos_harness/macos.py:985没有可捕获窗口或窗口索引越界会直接失败。src/macos_harness/macos.py:1044到src/macos_harness/macos.py:1045防止无效缩放尺寸进入 Pillow。
关联模块
src/macos_harness/controls.py:28 到 src/macos_harness/controls.py:57 提供坐标命中元素的 AX escape hatch。tests/test_macos.py:27 到 tests/test_macos.py:44 验证 AX tree 文本输出,便于模型直接阅读。