macos-harness 给 LLM 一台真实 Mac

21def _namespace() -> dict[str, Any]:
22    return {
23        "__name__": "__macos_harness__",
24        "browser": BrowserHarness(),
25        "mac": MacOS(),
26        "Path": Path,
27        "subprocess": subprocess,
28    }
29
30
31def _execute(code: str) -> int:
32    if not code.strip():
33        print("No Python code received on stdin", file=sys.stderr)
34        return 2
35    namespace = _namespace()
36    exec(compile(code, "<macos-harness>", "exec"), namespace, namespace)  # noqa: S102
37    return 0

85def main(argv: list[str] | None = None) -> int:
86    parser = _build_parser()
87    args = parser.parse_args(argv)
88    if args.command == "telemetry":
89        return run_telemetry_cli([args.action] if args.action else [])
90
91    started = time.monotonic()
92    result: int | None = None
93    try:
94        if args.command == "doctor":
95            mac = MacOS()
96            if args.request:
97                mac.request_permissions()
98            print(json.dumps(mac.doctor(), indent=2))
99            result = 0
100            return result
101        if args.command == "apps":
102            print(json.dumps(MacOS().list_apps(), indent=2))
103            result = 0
104            return result
105        if args.command == "skill":
106            print(_skill_text(), end="")
107            result = 0
108            return result
109        if args.command == "repl":
110            code.interact(
111                banner=(
112                    "macos-harness: mac.see/key/type/click/ax/script, browser, "
113                    "Path, and subprocess are ready"
114                ),
115                local=_namespace(),
116                exitmsg="",
117            )
118            result = 0
119            return result
120        if args.command == "see":
121            result = MacOS().see(
122                args.app,
123                max_width=args.max_width,
124                max_height=args.max_height,
125                show_pointer=not args.no_pointer,
126            )
127            print(json.dumps(result, indent=2, ensure_ascii=False))
128            result = 0
129            return result
130        if args.command == "state":
131            state = MacOS().get_app_state(
132                args.app,
133                screenshot=args.screenshot,
134                max_depth=args.max_depth,
135                max_nodes=args.max_nodes,
136                include_menu_bar=args.include_menu_bar,
137            )
138            print(json.dumps(state, indent=2, ensure_ascii=False))
139            result = 0
140            return result
141        if args.command is None:
142            if sys.stdin.isatty():
143                parser.print_help()
144                result = 2
145                return result
146            result = _execute(sys.stdin.read())
147            return result
148        parser.error(f"unknown command: {args.command}")
149    except (MacOSError, RuntimeError) as exc:
150        print(f"macos-harness: {exc}", file=sys.stderr)
151        result = 1
152        return result
153    finally:
154        capture_cli(args.command or "python", result == 0, time.monotonic() - started)

405    def _ensure_accessibility(self) -> None:
406        if not self.is_accessibility_trusted():
407            raise AccessibilityPermissionError(
408                "Accessibility permission is required. Grant it to the terminal or "
409                "agent host in System Settings → Privacy & Security → Accessibility, "
410                "or call mac.request_accessibility_permission() to show Apple's prompt."
411            )
412
413    def _ensure_screen_recording(self) -> None:
414        if self._preflight_permission("CGPreflightScreenCaptureAccess") is False:
415            raise AccessibilityPermissionError(
416                "Screen Recording permission is required. Grant it to the terminal "
417                "or agent host in System Settings → Privacy & Security → Screen & "
418                "System Audio Recording, or call mac.request_permissions()."
419            )
420
421    def _ensure_post_events(self) -> None:
422        if self._preflight_permission("CGPreflightPostEventAccess") is False:
423            raise AccessibilityPermissionError(
424                "Permission to post input events is required. Grant control access "
425                "to the terminal or agent host, or call mac.request_permissions()."
426            )
427

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

968    def capture_screenshot(
969        self,
970        app: str | None = None,
971        *,
972        window_index: int = 0,
973        path: str | Path | None = None,
974    ) -> dict[str, Any]:
975        self._ensure_screen_recording()
976        _, info = self._resolve_app(app)
977        windows = self.windows(str(info["pid"]))
978        if not windows:
979            raise MacOSError(f"No capturable windows found for {app or self._last_app}")
980        try:
981            window = windows[window_index]
982        except IndexError as exc:
983            raise MacOSError(
984                f"Window index {window_index} is out of range; found {len(windows)} windows"
985            ) from exc
986
987        if path is None:
988            with tempfile.NamedTemporaryFile(
989                prefix="macos-harness-", suffix=".png", delete=False
990            ) as handle:
991                output = Path(handle.name)
992            output.unlink(missing_ok=True)
993        else:
994            output = Path(path).expanduser().resolve()
995            output.parent.mkdir(parents=True, exist_ok=True)
996
997        result = subprocess.run(
998            [
999                "/usr/sbin/screencapture",
1000                "-x",
1001                "-o",
1002                "-l",
1003                str(window["window_id"]),
1004                str(output),
1005            ],
1006            capture_output=True,
1007            text=True,
1008            check=False,
1009        )
1010        if result.returncode != 0 or not output.exists():
1011            output.unlink(missing_ok=True)
1012            detail = (result.stderr or result.stdout or "no image returned").strip()
1013            raise MacOSError(f"Window screenshot failed: {detail}")
1014
1015        width, height = _png_size(output)
1016        bounds = window["bounds"]
1017        screenshot = {
1018            "path": str(output),
1019            "app": info,
1020            "pid": info["pid"],
1021            "window_id": window["window_id"],
1022            "width": width,
1023            "height": height,
1024            "bounds": bounds,
1025            "scale_x": width / bounds["width"],
1026            "scale_y": height / bounds["height"],
1027        }
1028        self._last_app = info
1029        self._last_windows = windows
1030        self._last_screenshot = screenshot
1031        return screenshot

1130    def _pid(self, app: str | None) -> int | None:
1131        if app is None and self._last_app is None:
1132            return None
1133        _, info = self._resolve_app(app)
1134        self._last_app = info
1135        return int(info["pid"])
1136
1137    @staticmethod
1138    def _post(event: Any, pid: int | None) -> None:
1139        if pid is None:
1140            raise MacOSError("Input requires an app or prior app snapshot")
1141        AS.CGEventPostToPid(pid, event)
1142
1143    def _screen_point(
1144        self, x: float, y: float, coordinate_space: str
1145    ) -> tuple[float, float]:
1146        if coordinate_space == "screen":
1147            return float(x), float(y)
1148        if self._last_screenshot is None:
1149            raise MacOSError(
1150                "Take a screenshot before using window or screenshot coordinates, "
1151                "or pass coordinate_space='screen'"
1152            )
1153        shot = self._last_screenshot
1154        bounds = shot["bounds"]
1155        if coordinate_space == "screenshot":
1156            x = float(x) / float(shot["scale_x"])
1157            y = float(y) / float(shot["scale_y"])
1158        elif coordinate_space != "window":
1159            raise MacOSError(
1160                "coordinate_space must be 'screenshot', 'window', or 'screen'"
1161            )
1162        return bounds["x"] + float(x), bounds["y"] + float(y)

1206    def click(
1207        self,
1208        x: float,
1209        y: float,
1210        *,
1211        app: str | None = None,
1212        button: str = "left",
1213        clicks: int = 1,
1214        coordinate_space: str = "screenshot",
1215    ) -> dict[str, Any]:
1216        """Send one raw coordinate click to an app PID; never guess an AX action."""
1217        self._ensure_accessibility()
1218        self._ensure_post_events()
1219        button = button.casefold()
1220        if button not in _BUTTONS:
1221            raise MacOSError(f"Unknown mouse button {button!r}")
1222        point = self._screen_point(x, y, coordinate_space)
1223        pid = self._pid(app)
1224        if pid is None:
1225            raise MacOSError("Pointer input requires an app or prior app snapshot")
1226        focus_before = self._frontmost_app()
1227        self._pointer_position = point
1228        self._overlay.move(*point)
1229        down_type, up_type, _ = _MOUSE_EVENTS[button]
1230        for click_count in range(1, max(1, int(clicks)) + 1):
1231            down = AS.CGEventCreateMouseEvent(
1232                self._event_source, down_type, point, _BUTTONS[button]
1233            )
1234            up = AS.CGEventCreateMouseEvent(
1235                self._event_source, up_type, point, _BUTTONS[button]
1236            )
1237            AS.CGEventSetIntegerValueField(
1238                down, AS.kCGMouseEventClickState, click_count
1239            )
1240            AS.CGEventSetIntegerValueField(up, AS.kCGMouseEventClickState, click_count)
1241            self._post(down, pid)
1242            time.sleep(0.03)
1243            self._post(up, pid)
1244            time.sleep(0.03)
1245            self._guard_focus(focus_before, pid, "click")
1246        self._overlay.click()
1247        pointer = self._pointer_info()
1248        assert pointer is not None
1249        return pointer

1358    def type(self, text: str, *, app: str | None = None) -> None:
1359        self._ensure_accessibility()
1360        self._ensure_post_events()
1361        pid = self._pid(app)
1362        if pid is None:
1363            raise MacOSError("Keyboard input requires an app or prior app snapshot")
1364        focus_before = self._frontmost_app()
1365        aliases = {" ": "space", "\n": "return", "\r": "return", "\t": "tab"}
1366        for character in text:
1367            base = _SHIFTED_CHARACTERS.get(
1368                character, aliases.get(character, character.casefold())
1369            )
1370            flags = (
1371                AS.kCGEventFlagMaskShift
1372                if character.isupper() or character in _SHIFTED_CHARACTERS
1373                else 0
1374            )
1375            keycode = _KEYCODES.get(base, 0)
1376            down = AS.CGEventCreateKeyboardEvent(self._event_source, keycode, True)
1377            up = AS.CGEventCreateKeyboardEvent(self._event_source, keycode, False)
1378            length = len(character.encode("utf-16-le")) // 2
1379            AS.CGEventKeyboardSetUnicodeString(down, length, character)
1380            AS.CGEventKeyboardSetUnicodeString(up, length, character)
1381            AS.CGEventSetFlags(down, flags)
1382            AS.CGEventSetFlags(up, flags)
1383            self._post(down, pid)
1384            self._post(up, pid)
1385            time.sleep(0.01)
1386            self._guard_focus(focus_before, pid, "typing")
1387
1388    def key(self, key: str, *, app: str | None = None) -> None:
1389        self._ensure_accessibility()
1390        self._ensure_post_events()
1391        parts = [part.casefold() for part in re.split(r"[+-]", key) if part]
1392        if not parts:
1393            raise MacOSError("Key must not be empty")
1394        base = parts[-1]
1395        modifiers = parts[:-1]
1396        try:
1397            keycode = _KEYCODES[base]
1398        except KeyError as exc:
1399            raise MacOSError(
1400                f"Unsupported key {base!r}; use mac.type() for arbitrary text"
1401            ) from exc
1402        parsed_modifiers: list[tuple[int, int]] = []
1403        seen_keycodes: set[int] = set()
1404        for modifier in modifiers:
1405            try:
1406                modifier_keycode = _MODIFIER_KEYCODES[modifier]
1407                modifier_flag = _MODIFIER_FLAGS[modifier]
1408            except KeyError as exc:
1409                raise MacOSError(f"Unsupported modifier {modifier!r}") from exc
1410            if modifier_keycode not in seen_keycodes:
1411                seen_keycodes.add(modifier_keycode)
1412                parsed_modifiers.append((modifier_keycode, modifier_flag))
1413        pid = self._pid(app)
1414        if pid is None:
1415            raise MacOSError("Keyboard input requires an app or prior app snapshot")
1416        focus_before = self._frontmost_app()
1417        active_flags = 0
1418        pressed: list[tuple[int, int]] = []
1419        try:
1420            for modifier_keycode, modifier_flag in parsed_modifiers:
1421                active_flags |= modifier_flag
1422                event = AS.CGEventCreateKeyboardEvent(
1423                    self._event_source, modifier_keycode, True
1424                )
1425                AS.CGEventSetFlags(event, active_flags)
1426                self._post(event, pid)
1427                pressed.append((modifier_keycode, modifier_flag))
1428                time.sleep(0.005)
1429
1430            down = AS.CGEventCreateKeyboardEvent(self._event_source, keycode, True)
1431            up = AS.CGEventCreateKeyboardEvent(self._event_source, keycode, False)
1432            AS.CGEventSetFlags(down, active_flags)
1433            AS.CGEventSetFlags(up, active_flags)
1434            self._post(down, pid)
1435            self._post(up, pid)
1436        finally:
1437            for modifier_keycode, modifier_flag in reversed(pressed):
1438                active_flags &= ~modifier_flag
1439                event = AS.CGEventCreateKeyboardEvent(
1440                    self._event_source, modifier_keycode, False
1441                )
1442                AS.CGEventSetFlags(event, active_flags)
1443                self._post(event, pid)
1444        time.sleep(0.01)
1445        self._guard_focus(focus_before, pid, f"key {key!r}")

106class BrowserHarness:
107    """Expose Browser Harness helpers without connecting until first use."""
108
109    def __init__(self, *, name: str | None = None, wait: float = 60.0) -> None:
110        self.name = name
111        self.wait = wait
112        self._helpers: ModuleType | None = None
113
114    def connect(self) -> BrowserHarness:
115        if self._helpers is not None:
116            return self
117
118        if self.name:
119            os.environ["BU_NAME"] = self.name
120
121        try:
122            from browser_harness import admin, helpers
123        except ImportError as exc:  # pragma: no cover - packaging failure
124            raise RuntimeError(
125                "Browser Harness is unavailable. Install the project dependencies "
126                "with `uv sync`."
127            ) from exc
128
129        approver = subprocess.Popen(
130            [
131                sys.executable,
132                "-c",
133                (
134                    "from macos_harness.browser import _watch_chrome_prompts; "
135                    f"_watch_chrome_prompts({self.wait!r})"
136                ),
137            ],
138            stdout=subprocess.DEVNULL,
139            stderr=subprocess.DEVNULL,
140        )
141        try:
142            admin.ensure_daemon(wait=self.wait, name=self.name)
143        finally:
144            if approver.poll() is None:
145                approver.terminate()
146                try:
147                    approver.wait(timeout=0.2)
148                except subprocess.TimeoutExpired:
149                    approver.kill()
150        if self.name:
151            helpers.NAME = self.name
152        self._helpers = helpers
153        return self
154
155    def __getattr__(self, name: str) -> Any:
156        if name.startswith("_"):
157            raise AttributeError(name)
158        self.connect()
159        assert self._helpers is not None
160        try:
161            return getattr(self._helpers, name)
162        except AttributeError as exc:
163            raise AttributeError(f"Browser Harness has no helper {name!r}") from exc
164
165    def __repr__(self) -> str:
166        status = "connected" if self._helpers is not None else "lazy"
167        suffix = f", name={self.name!r}" if self.name else ""
168        return f"BrowserHarness({status}{suffix})"

21class LivePointerOverlay:
22    """Send pointer updates to a disposable AppKit helper process."""
23
24    def __init__(self) -> None:
25        self._process: subprocess.Popen[str] | None = None
26        self._visible = True
27        atexit.register(self.close)
28
29    @property
30    def visible(self) -> bool:
31        return self._visible
32
33    @property
34    def running(self) -> bool:
35        return self._process is not None and self._process.poll() is None
36
37    def move(self, x: float, y: float, *, duration: float = 0.16) -> None:
38        self._visible = True
39        self._send(
40            {
41                "cmd": "move",
42                "x": float(x),
43                "y": float(y),
44                "duration": max(0.0, float(duration)),
45            }
46        )
47
48    def show(self, x: float, y: float) -> None:
49        self._visible = True
50        self._send({"cmd": "show", "x": float(x), "y": float(y)})
51
52    def hide(self) -> None:
53        self._visible = False
54        if self.running:
55            self._send({"cmd": "hide"}, start=False)
56
57    def click(self) -> None:
58        if self._visible:
59            self._send({"cmd": "click"})
60
61    def close(self) -> None:
62        process = self._process
63        self._process = None
64        if process is None:
65            return
66        if process.poll() is None and process.stdin is not None:
67            try:
68                process.stdin.write('{"cmd":"quit"}\n')
69                process.stdin.flush()
70                process.stdin.close()
71            except (BrokenPipeError, OSError):
72                pass
73
74    def _start(self) -> subprocess.Popen[str]:
75        process = subprocess.Popen(
76            [sys.executable, "-m", "macos_harness.overlay", "--helper"],
77            stdin=subprocess.PIPE,
78            stdout=subprocess.DEVNULL,
79            stderr=subprocess.DEVNULL,
80            text=True,
81            bufsize=1,
82        )
83        self._process = process
84        return process
85
86    def _send(self, payload: dict[str, Any], *, start: bool = True) -> None:
87        process = self._process
88        if process is None or process.poll() is not None:
89            if not start:
90                return
91            process = self._start()
92        stream = process.stdin
93        if stream is None:
94            return
95        try:
96            stream.write(json.dumps(payload, separators=(",", ":")) + "\n")
97            stream.flush()
98        except (BrokenPipeError, OSError):
99            if not start:
100                return
101            self._process = None
102            retry = self._start()
103            assert retry.stdin is not None
104            retry.stdin.write(json.dumps(payload, separators=(",", ":")) + "\n")
105            retry.stdin.flush()

157def capture_cli(command: str | None, success: bool, duration_seconds: float) -> None:
158    """Capture only a known command and coarse runtime metadata—never user data."""
159    if not is_enabled():
160        return
161    try:
162        payload = {
163            "api_key": POSTHOG_KEY,
164            "distinct_id": _install_id(),
165            "event": "macos_harness_cli",
166            "properties": {
167                "$process_person_profile": False,
168                "$geoip_disable": True,
169                "command": command if command in COMMANDS else "python",
170                "success": success,
171                "duration_seconds": round(max(duration_seconds, 0), 2),
172                "macos_harness_version": _version(),
173                "python_version": f"{sys.version_info.major}.{sys.version_info.minor}",
174                "os": platform.system() or "unknown",
175                "machine": platform.machine() or "unknown",
176                "agent": _agent(),
177            },
178        }
179        _send(payload)
180    except (OSError, TypeError, ValueError, subprocess.SubprocessError):

179def test_background_click_posts_to_pid_without_warp_or_activate(monkeypatch) -> None:
180    mac = MacOS()
181    posted = []
182    monkeypatch.setattr(mac, "_ensure_accessibility", lambda: None)
183    monkeypatch.setattr(mac, "_ensure_post_events", lambda: None)
184    monkeypatch.setattr(mac, "_pid", lambda app: 42)
185    monkeypatch.setattr(mac, "_post", lambda event, pid: posted.append(pid))
186    monkeypatch.setattr(
187        macos_module.AS,
188        "CGWarpMouseCursorPosition",
189        lambda point: pytest.fail("background click must not warp the cursor"),
190    )
191
192    mac.click(
193        10,
194        20,
195        app="Slack",
196        coordinate_space="screen",
197    )
198
199    assert posted == [42, 42]
200
201
202def test_input_without_target_is_refused(monkeypatch) -> None:
203    mac = MacOS()
204    monkeypatch.setattr(mac, "_ensure_accessibility", lambda: None)
205    monkeypatch.setattr(mac, "_ensure_post_events", lambda: None)
206    monkeypatch.setattr(
207        macos_module.AS,
208        "AXUIElementCopyElementAtPosition",
209        lambda *args: pytest.fail("untargeted input must stop before AX hit testing"),
210    )
211
212    with pytest.raises(MacOSError, match="requires an app"):
213        mac.type("hello")
214    with pytest.raises(MacOSError, match="requires an app"):
215        mac.click(10, 20, coordinate_space="screen")

8def test_chrome_approver_uses_systemwide_ax_without_input(monkeypatch) -> None:
9    root = object()
10    button = object()
11    pressed = []
12    stop = threading.Event()
13    bounds = {"X": 100, "Y": 200, "Width": 448, "Height": 240}
14
15    class FakeAS:
16        kCGWindowListOptionOnScreenOnly = 1
17        kCGNullWindowID = 0
18
19        @staticmethod
20        def AXUIElementCreateSystemWide():
21            return root
22
23        @staticmethod
24        def CGWindowListCopyWindowInfo(options, window_id):
25            return [
26                {
27                    "kCGWindowOwnerName": "Google Chrome",
28                    "kCGWindowName": "Allow remote debugging?",
29                    "kCGWindowBounds": bounds,
30                }
31            ]
32
33        @staticmethod
34        def AXUIElementCopyElementAtPosition(element, x, y, error):
35            assert element is root
36            assert (x, y) == (490.0, 402.0)
37            return 0, button
38
39        @staticmethod
40        def AXUIElementCopyAttributeValue(element, attribute, error):
41            assert element is button
42            return 0, {"AXRole": "AXButton", "AXDescription": "Allow"}[attribute]
43
44        @staticmethod
45        def AXUIElementPerformAction(element, action):
46            pressed.append((element, action))
47            stop.set()
48
49    monkeypatch.setattr(browser_module.platform, "system", lambda: "Darwin")
50    monkeypatch.setitem(__import__("sys").modules, "ApplicationServices", FakeAS)
51
52    browser_module._approve_chrome_prompt(stop, 1)
53
54    assert pressed == [(button, "AXPress")]