diff --git a/desktop/src/i18n/locales/en.ts b/desktop/src/i18n/locales/en.ts index 0e2ef2ca..f59cb6c5 100644 --- a/desktop/src/i18n/locales/en.ts +++ b/desktop/src/i18n/locales/en.ts @@ -236,8 +236,8 @@ export const en = { // Settings > Computer Use 'settings.tab.computerUse': 'Computer Use', 'settings.computerUse.title': 'Computer Use', - 'settings.computerUse.description': 'Allow Claude to take screenshots, click, type, and control your computer. Requires Python 3 and macOS accessibility permissions.', - 'settings.computerUse.notSupported': 'Computer Use is only supported on macOS.', + 'settings.computerUse.description': 'Allow Claude to take screenshots, click, type, and control your computer. Requires Python 3. On macOS, accessibility permissions are also needed.', + 'settings.computerUse.notSupported': 'Computer Use is only supported on macOS and Windows.', 'settings.computerUse.python': 'Python 3', 'settings.computerUse.pythonNotFound': 'Not installed. Please install Python 3 first.', 'settings.computerUse.pythonFound': 'Installed', diff --git a/desktop/src/i18n/locales/zh.ts b/desktop/src/i18n/locales/zh.ts index 866508ee..2563db46 100644 --- a/desktop/src/i18n/locales/zh.ts +++ b/desktop/src/i18n/locales/zh.ts @@ -238,8 +238,8 @@ export const zh: Record = { // Settings > Computer Use 'settings.tab.computerUse': 'Computer Use', 'settings.computerUse.title': 'Computer Use', - 'settings.computerUse.description': '允许 Claude 截屏、点击、打字并控制你的电脑。需要 Python 3 和 macOS 辅助功能权限。', - 'settings.computerUse.notSupported': 'Computer Use 仅支持 macOS。', + 'settings.computerUse.description': '允许 Claude 截屏、点击、打字并控制你的电脑。需要 Python 3，macOS 上还需要辅助功能权限。', + 'settings.computerUse.notSupported': 'Computer Use 仅支持 macOS 和 Windows。', 'settings.computerUse.python': 'Python 3', 'settings.computerUse.pythonNotFound': '未安装，请先安装 Python 3。', 'settings.computerUse.pythonFound': '已安装', diff --git a/desktop/src/pages/ComputerUseSettings.tsx b/desktop/src/pages/ComputerUseSettings.tsx index fda18e60..9ff7b4fe 100644 --- a/desktop/src/pages/ComputerUseSettings.tsx +++ b/desktop/src/pages/ComputerUseSettings.tsx @@ -225,8 +225,8 @@ export function ComputerUseSettings() { /> - {/* macOS Permissions */} - {envReady && ( + {/* macOS Permissions — only shown on macOS (darwin) */} + {envReady && status.platform === 'darwin' && ( <> )} - {allReady && status.permissions.accessibility && screenRecordingReady && ( + {allReady && (status.platform !== 'darwin' || (status.permissions.accessibility && screenRecordingReady)) && (
verified {t('settings.computerUse.allReady')} diff --git a/runtime/requirements-win.txt b/runtime/requirements-win.txt new file mode 100644 index 00000000..32c7a94c --- /dev/null +++ b/runtime/requirements-win.txt @@ -0,0 +1,7 @@ +mss>=10.1.0 +Pillow>=11.3.0 +pyautogui>=0.9.54 +pywin32>=306 +psutil>=5.9.0 +pyperclip>=1.8.2 +screeninfo>=0.8.1 diff --git a/runtime/test_helpers.py b/runtime/test_helpers.py new file mode 100644 index 00000000..4c5e8db1 --- /dev/null +++ b/runtime/test_helpers.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""Cross-platform tests for mac_helper.py and win_helper.py. + +Tests the platform-independent parts (JSON protocol, key mapping, capture logic) +without requiring platform-specific dependencies. Can run on any OS with pytest. + +Usage: + python -m pytest runtime/test_helpers.py -v + # or simply: + python runtime/test_helpers.py +""" +from __future__ import annotations + +import json +import subprocess +import sys +import unittest +from pathlib import Path +from unittest.mock import patch, MagicMock + +# Determine which helper to test based on current platform +IS_WINDOWS = sys.platform == "win32" +IS_MACOS = sys.platform == "darwin" + +RUNTIME_DIR = Path(__file__).parent +MAC_HELPER = RUNTIME_DIR / "mac_helper.py" +WIN_HELPER = RUNTIME_DIR / "win_helper.py" + + +class TestKeyMap(unittest.TestCase): + """Test the KEY_MAP and normalize_key function — platform-independent logic.""" + + def _load_key_map(self, helper_path: Path) -> dict[str, str]: + """Extract KEY_MAP from a helper by importing it with mocked deps.""" + # Read the file and extract just the KEY_MAP dict + source = helper_path.read_text() + # Find KEY_MAP definition + start = source.index("KEY_MAP = {") + # Find the matching closing brace + depth = 0 + for i, ch in enumerate(source[start:], start): + if ch == "{": + depth += 1 + elif ch == "}": + depth -= 1 + if depth == 0: + end = i + 1 + break + key_map_source = source[start:end] + ns: dict = {} + exec(key_map_source, ns) + return ns["KEY_MAP"] + + def test_mac_key_map_exists(self): + if not MAC_HELPER.exists(): + self.skipTest("mac_helper.py not found") + km = self._load_key_map(MAC_HELPER) + self.assertIn("cmd", km) + self.assertIn("ctrl", km) + self.assertEqual(km["cmd"], "command") + self.assertEqual(km["alt"], "option") + + def test_win_key_map_exists(self): + if not WIN_HELPER.exists(): + self.skipTest("win_helper.py not found") + km = self._load_key_map(WIN_HELPER) + self.assertIn("cmd", km) + self.assertIn("ctrl", km) + # Windows maps cmd/command/meta to 'win' key + self.assertEqual(km["cmd"], "win") + self.assertEqual(km["command"], "win") + self.assertEqual(km["meta"], "win") + # Windows maps alt/option to 'alt' + self.assertEqual(km["alt"], "alt") + self.assertEqual(km["option"], "alt") + + def test_common_keys_present_in_both(self): + """Both helpers must have the same set of key names.""" + if not MAC_HELPER.exists() or not WIN_HELPER.exists(): + self.skipTest("Both helpers required") + mac_km = self._load_key_map(MAC_HELPER) + win_km = self._load_key_map(WIN_HELPER) + # All keys in mac should be in win and vice versa + self.assertEqual(set(mac_km.keys()), set(win_km.keys()), + "KEY_MAP keys must be identical across platforms") + + def test_all_alphabet_keys(self): + """All a-z keys should map to themselves.""" + for helper in [MAC_HELPER, WIN_HELPER]: + if not helper.exists(): + continue + km = self._load_key_map(helper) + for char in "abcdefghijklmnopqrstuvwxyz": + self.assertEqual(km[char], char, f"{helper.name}: {char} should map to itself") + + def test_all_digit_keys(self): + """All 0-9 keys should map to themselves.""" + for helper in [MAC_HELPER, WIN_HELPER]: + if not helper.exists(): + continue + km = self._load_key_map(helper) + for digit in "0123456789": + self.assertEqual(km[digit], digit, f"{helper.name}: {digit} should map to itself") + + def test_function_keys(self): + """F1-F12 should map to themselves.""" + for helper in [MAC_HELPER, WIN_HELPER]: + if not helper.exists(): + continue + km = self._load_key_map(helper) + for i in range(1, 13): + key = f"f{i}" + self.assertEqual(km[key], key, f"{helper.name}: {key} should map to itself") + + +class TestJSONProtocol(unittest.TestCase): + """Test that both helpers follow the same JSON command protocol.""" + + def _get_helper(self) -> Path: + """Get the appropriate helper for the current platform.""" + if IS_WINDOWS and WIN_HELPER.exists(): + return WIN_HELPER + if IS_MACOS and MAC_HELPER.exists(): + return MAC_HELPER + return MAC_HELPER if MAC_HELPER.exists() else WIN_HELPER + + def _parse_main_commands(self, helper_path: Path) -> list[str]: + """Extract all command names from the main() dispatcher.""" + source = helper_path.read_text() + commands = [] + for line in source.splitlines(): + stripped = line.strip() + if stripped.startswith('if command == "'): + cmd = stripped.split('"')[1] + commands.append(cmd) + return commands + + def test_both_helpers_same_commands(self): + """Both helpers must support the exact same set of commands.""" + if not MAC_HELPER.exists() or not WIN_HELPER.exists(): + self.skipTest("Both helpers required") + mac_cmds = set(self._parse_main_commands(MAC_HELPER)) + win_cmds = set(self._parse_main_commands(WIN_HELPER)) + self.assertEqual(mac_cmds, win_cmds, + f"Command sets differ.\nOnly in mac: {mac_cmds - win_cmds}\nOnly in win: {win_cmds - mac_cmds}") + + def test_expected_commands_exist(self): + """Core commands should be present in each helper.""" + expected = { + "check_permissions", "list_displays", "get_display_size", + "screenshot", "resolve_prepare_capture", "zoom", + "prepare_for_action", "preview_hide_set", "find_window_displays", + "key", "hold_key", "type", "click", "drag", + "move_mouse", "scroll", "mouse_down", "mouse_up", + "cursor_position", "frontmost_app", "app_under_point", + "list_installed_apps", "list_running_apps", "open_app", + "read_clipboard", "write_clipboard", + } + for helper in [MAC_HELPER, WIN_HELPER]: + if not helper.exists(): + continue + cmds = set(self._parse_main_commands(helper)) + missing = expected - cmds + self.assertFalse(missing, + f"{helper.name} missing commands: {missing}") + + def test_unknown_command_returns_error(self): + """Running a non-existent command should return a JSON error.""" + helper = self._get_helper() + if not helper.exists(): + self.skipTest("No helper found") + # On macOS without venv, mac_helper.py may fail at import (AppKit); + # on Windows without venv, win_helper.py may fail at import (win32gui). + # Only test if the helper can actually import. + check = subprocess.run( + [sys.executable, "-c", f"import importlib.util; " + f"spec = importlib.util.spec_from_file_location('h', '{helper}')"], + capture_output=True, text=True + ) + result = subprocess.run( + [sys.executable, str(helper), "nonexistent_command_xyz"], + capture_output=True, text=True + ) + if result.returncode == 1 and not result.stdout.strip(): + # Import failed — platform deps missing, skip this test + self.skipTest(f"Cannot run {helper.name} on this platform (missing deps)") + # Should exit with code 2 + self.assertEqual(result.returncode, 2) + parsed = json.loads(result.stdout.strip()) + self.assertFalse(parsed["ok"]) + self.assertEqual(parsed["error"]["code"], "bad_command") + + +class TestHelperOutputFormat(unittest.TestCase): + """Test the JSON output helpers are consistent.""" + + def test_json_output_function_exists(self): + """Both helpers should define json_output and error_output.""" + for helper in [MAC_HELPER, WIN_HELPER]: + if not helper.exists(): + continue + source = helper.read_text() + self.assertIn("def json_output(", source, + f"{helper.name} missing json_output function") + self.assertIn("def error_output(", source, + f"{helper.name} missing error_output function") + + def test_main_entry_point(self): + """Both helpers should have the standard main entry point.""" + for helper in [MAC_HELPER, WIN_HELPER]: + if not helper.exists(): + continue + source = helper.read_text() + self.assertIn('if __name__ == "__main__":', source, + f"{helper.name} missing __main__ guard") + self.assertIn("def main()", source, + f"{helper.name} missing main() function") + + +class TestWinHelperPermissions(unittest.TestCase): + """Windows-specific: permissions should always return True.""" + + def test_check_permissions_always_granted(self): + """On Windows, permissions are not needed — should always be True.""" + if not WIN_HELPER.exists(): + self.skipTest("win_helper.py not found") + + # Extract and exec just the check_permissions function + source = WIN_HELPER.read_text() + + # Find the function + self.assertIn("def check_permissions()", source) + + # The function should return both as True + # We can verify by reading the source + start = source.index("def check_permissions()") + # Find next def or end + rest = source[start:] + lines = rest.split("\n") + func_lines = [lines[0]] + for line in lines[1:]: + if line and not line[0].isspace() and not line.startswith("#"): + break + func_lines.append(line) + func_source = "\n".join(func_lines) + self.assertIn('"accessibility": True', func_source) + self.assertIn('"screenRecording": True', func_source) + + +class TestCrossPlatformFunctions(unittest.TestCase): + """Test functions that are identical between both helpers.""" + + def _get_function_body(self, helper_path: Path, func_name: str) -> str: + """Extract a function's body (code lines only, no comments/blanks).""" + source = helper_path.read_text() + marker = f"def {func_name}(" + if marker not in source: + return "" + start = source.index(marker) + rest = source[start:] + lines = rest.split("\n") + func_lines = [lines[0]] + for line in lines[1:]: + # Stop at next top-level def/class or non-indented non-empty line + stripped = line.strip() + if line and not line[0].isspace() and stripped and not stripped.startswith("#"): + break + # Skip comments and blank lines for comparison + if stripped.startswith("#") or not stripped: + continue + func_lines.append(line) + return " ".join(" ".join(func_lines).split()) + + def test_input_functions_identical(self): + """Input action functions (click, scroll, etc.) should be identical.""" + if not MAC_HELPER.exists() or not WIN_HELPER.exists(): + self.skipTest("Both helpers required") + for func in ["click", "scroll", "key_action", "hold_keys", "type_text"]: + mac_src = self._get_function_body(MAC_HELPER, func) + win_src = self._get_function_body(WIN_HELPER, func) + self.assertEqual(mac_src, win_src, + f"{func} should be identical across platforms") + + +if __name__ == "__main__": + unittest.main() diff --git a/runtime/win_helper.py b/runtime/win_helper.py new file mode 100644 index 00000000..f6ff848e --- /dev/null +++ b/runtime/win_helper.py @@ -0,0 +1,715 @@ +#!/usr/bin/env python3 +"""Windows Computer Use helper — same JSON protocol as mac_helper.py. + +Uses win32gui / win32api / win32process / psutil / pyperclip / screeninfo +to replicate macOS-specific Quartz/AppKit functionality on Windows. +""" +from __future__ import annotations + +import argparse +import base64 +import json +import os +import subprocess +import sys +import time +from io import BytesIO +from pathlib import Path +from typing import Any + +import mss +from PIL import Image + +os.environ.setdefault("PYTHONDONTWRITEBYTECODE", "1") +os.environ.setdefault("PYAUTOGUI_HIDE_SUPPORT_PROMPT", "1") + +import pyautogui # noqa: E402 + +pyautogui.FAILSAFE = False +pyautogui.PAUSE = 0 + +# --------------------------------------------------------------------------- +# Key mapping — Windows uses 'win' instead of 'command' +# --------------------------------------------------------------------------- +KEY_MAP = { + "a": "a", "b": "b", "c": "c", "d": "d", "e": "e", + "f": "f", "g": "g", "h": "h", "i": "i", "j": "j", + "k": "k", "l": "l", "m": "m", "n": "n", "o": "o", + "p": "p", "q": "q", "r": "r", "s": "s", "t": "t", + "u": "u", "v": "v", "w": "w", "x": "x", "y": "y", + "z": "z", + "0": "0", "1": "1", "2": "2", "3": "3", "4": "4", + "5": "5", "6": "6", "7": "7", "8": "8", "9": "9", + # Modifier keys — map macOS names to Windows equivalents + "cmd": "win", + "command": "win", + "meta": "win", + "super": "win", + "ctrl": "ctrl", + "control": "ctrl", + "shift": "shift", + "alt": "alt", + "option": "alt", + "opt": "alt", + "fn": "fn", + # Navigation / editing + "escape": "esc", + "esc": "esc", + "enter": "enter", + "return": "enter", + "tab": "tab", + "space": "space", + "backspace": "backspace", + "delete": "delete", + "forwarddelete": "delete", + "up": "up", + "down": "down", + "left": "left", + "right": "right", + "home": "home", + "end": "end", + "pageup": "pageup", + "pagedown": "pagedown", + "capslock": "capslock", + # Function keys + "f1": "f1", "f2": "f2", "f3": "f3", "f4": "f4", + "f5": "f5", "f6": "f6", "f7": "f7", "f8": "f8", + "f9": "f9", "f10": "f10", "f11": "f11", "f12": "f12", + # Symbols + "-": "-", "=": "=", "[": "[", "]": "]", "\\": "\\", + ";": ";", "'": "'", ",": ",", ".": ".", "/": "/", "`": "`", +} + + +def normalize_key(name: str) -> str: + key = name.strip().lower() + if key not in KEY_MAP: + raise ValueError(f"Unsupported key: {name}") + return KEY_MAP[key] + + +# --------------------------------------------------------------------------- +# JSON output helpers +# --------------------------------------------------------------------------- + +def json_output(payload: dict[str, Any]) -> None: + sys.stdout.write(json.dumps(payload, ensure_ascii=False)) + sys.stdout.write("\n") + sys.stdout.flush() + + +def error_output(message: str, code: str = "runtime_error") -> None: + json_output({"ok": False, "error": {"code": code, "message": message}}) + + +def bool_env(name: str, default: bool = False) -> bool: + value = os.environ.get(name) + if value is None: + return default + return value not in {"0", "false", "False", ""} + + +# --------------------------------------------------------------------------- +# Display / Monitor helpers (via screeninfo + ctypes) +# --------------------------------------------------------------------------- + +def get_displays() -> list[dict[str, Any]]: + """Enumerate monitors via screeninfo, with DPI scale from ctypes.""" + from screeninfo import get_monitors + + displays: list[dict[str, Any]] = [] + for idx, m in enumerate(get_monitors()): + scale_factor = _get_monitor_scale(m) + name = m.name or f"Display {idx + 1}" + displays.append({ + "id": idx, + "displayId": idx, + "width": m.width, + "height": m.height, + "scaleFactor": scale_factor, + "originX": m.x, + "originY": m.y, + "isPrimary": m.is_primary if hasattr(m, "is_primary") else (idx == 0), + "name": name, + "label": name, + }) + return displays + + +def _get_monitor_scale(monitor: Any) -> float: + """Get the DPI scale factor for a monitor. Returns 1.0 on failure.""" + try: + import ctypes + # SetProcessDPIAware so we get real pixel values + ctypes.windll.user32.SetProcessDPIAware() + # Get DPI for the primary — simplified; per-monitor DPI is complex + hdc = ctypes.windll.user32.GetDC(0) + dpi = ctypes.windll.gdi32.GetDeviceCaps(hdc, 88) # LOGPIXELSX + ctypes.windll.user32.ReleaseDC(0, hdc) + return dpi / 96.0 + except Exception: + return 1.0 + + +def choose_display(display_id: int | None) -> dict[str, Any]: + displays = get_displays() + if not displays: + raise RuntimeError("No active displays found") + if display_id is None: + for display in displays: + if display["isPrimary"]: + return display + return displays[0] + for display in displays: + if display["displayId"] == display_id or display["id"] == display_id: + return display + raise RuntimeError(f"Unknown display: {display_id}") + + +# --------------------------------------------------------------------------- +# Screen capture (mss — cross-platform, identical to mac_helper) +# --------------------------------------------------------------------------- + +def capture_display(display_id: int | None, resize: tuple[int, int] | None = None) -> dict[str, Any]: + display = choose_display(display_id) + monitor = { + "left": display["originX"], + "top": display["originY"], + "width": display["width"], + "height": display["height"], + } + with mss.mss() as sct: + raw = sct.grab(monitor) + image = Image.frombytes("RGB", raw.size, raw.rgb) + if resize: + image = image.resize(resize, Image.Resampling.LANCZOS) + buffer = BytesIO() + image.save(buffer, format="JPEG", quality=75, optimize=True) + base64_data = base64.b64encode(buffer.getvalue()).decode("ascii") + return { + "base64": base64_data, + "width": image.width, + "height": image.height, + "displayWidth": display["width"], + "displayHeight": display["height"], + "displayId": display["displayId"], + "originX": display["originX"], + "originY": display["originY"], + "display": display, + } + + +def capture_region(region: dict[str, int], resize: tuple[int, int] | None = None) -> dict[str, Any]: + with mss.mss() as sct: + raw = sct.grab(region) + image = Image.frombytes("RGB", raw.size, raw.rgb) + if resize: + image = image.resize(resize, Image.Resampling.LANCZOS) + buffer = BytesIO() + image.save(buffer, format="JPEG", quality=75, optimize=True) + base64_data = base64.b64encode(buffer.getvalue()).decode("ascii") + return {"base64": base64_data, "width": image.width, "height": image.height} + + +# --------------------------------------------------------------------------- +# Window management (win32gui) +# --------------------------------------------------------------------------- + +def list_windows() -> list[dict[str, Any]]: + """List visible on-screen windows with their bounds.""" + import win32gui + + results: list[dict[str, Any]] = [] + + def _enum_cb(hwnd: int, _: Any) -> None: + if not win32gui.IsWindowVisible(hwnd): + return + title = win32gui.GetWindowText(hwnd) + try: + left, top, right, bottom = win32gui.GetWindowRect(hwnd) + except Exception: + return + width = right - left + height = bottom - top + if width <= 1 or height <= 1: + return + # Get the process name as owner + owner = _get_window_process_name(hwnd) + results.append({ + "ownerName": owner, + "title": title, + "bounds": {"x": left, "y": top, "width": width, "height": height}, + }) + + win32gui.EnumWindows(_enum_cb, None) + return results + + +def _get_window_process_name(hwnd: int) -> str: + """Get the exe name of the process owning a window handle.""" + try: + import win32process + import psutil + _, pid = win32process.GetWindowThreadProcessId(hwnd) + proc = psutil.Process(pid) + return proc.name() + except Exception: + return "" + + +# --------------------------------------------------------------------------- +# Application management +# --------------------------------------------------------------------------- + +def _get_exe_path_for_pid(pid: int) -> str | None: + try: + import psutil + return psutil.Process(pid).exe() + except Exception: + return None + + +def installed_apps() -> list[dict[str, Any]]: + """List installed programs from Windows registry and Start Menu shortcuts.""" + import winreg + + results: dict[str, dict[str, Any]] = {} + reg_paths = [ + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall"), + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"), + ] + + for hive, sub_key in reg_paths: + try: + key = winreg.OpenKey(hive, sub_key) + except OSError: + continue + try: + i = 0 + while True: + try: + name = winreg.EnumKey(key, i) + i += 1 + except OSError: + break + try: + app_key = winreg.OpenKey(key, name) + except OSError: + continue + try: + display_name = winreg.QueryValueEx(app_key, "DisplayName")[0] + except OSError: + winreg.CloseKey(app_key) + continue + # Use the registry key name as a stable identifier (like bundleId) + try: + install_location = winreg.QueryValueEx(app_key, "InstallLocation")[0] + except OSError: + install_location = "" + try: + display_icon = winreg.QueryValueEx(app_key, "DisplayIcon")[0] + except OSError: + display_icon = "" + # Use registry key name as bundleId equivalent + bundle_id = name + if bundle_id not in results: + results[bundle_id] = { + "bundleId": bundle_id, + "displayName": str(display_name), + "path": str(install_location or display_icon or ""), + } + winreg.CloseKey(app_key) + finally: + winreg.CloseKey(key) + + return sorted(results.values(), key=lambda item: item["displayName"].lower()) + + +def running_apps() -> list[dict[str, Any]]: + """List running GUI applications.""" + import psutil + + apps: list[dict[str, Any]] = [] + seen: set[str] = set() + + for proc in psutil.process_iter(["pid", "name", "exe"]): + try: + name = proc.info["name"] or "" + exe_path = proc.info["exe"] or "" + if not name or name in seen: + continue + # Skip system/background processes (no window) + if not exe_path: + continue + seen.add(name) + # Use exe name (without .exe) as bundleId + bundle_id = Path(exe_path).stem if exe_path else name + apps.append({"bundleId": bundle_id, "displayName": name}) + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + return sorted(apps, key=lambda item: item["displayName"].lower()) + + +def app_display_name(bundle_id: str) -> str | None: + """Find display name for a given bundleId (exe stem or registry key).""" + import psutil + for proc in psutil.process_iter(["name", "exe"]): + try: + exe = proc.info["exe"] or "" + if exe and Path(exe).stem == bundle_id: + return proc.info["name"] + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + return None + + +def frontmost_app() -> dict[str, str] | None: + """Get the currently focused (foreground) application.""" + import win32gui + import win32process + import psutil + + hwnd = win32gui.GetForegroundWindow() + if not hwnd: + return None + try: + _, pid = win32process.GetWindowThreadProcessId(hwnd) + proc = psutil.Process(pid) + exe_path = proc.exe() + return { + "bundleId": Path(exe_path).stem, + "displayName": proc.name(), + } + except Exception: + return None + + +def app_under_point(x: int, y: int) -> dict[str, str] | None: + """Find the app whose window is under the given screen coordinate.""" + import win32gui + import win32process + import psutil + + hwnd = win32gui.WindowFromPoint((x, y)) + if not hwnd: + return frontmost_app() + # Walk up to the top-level owner + root = win32gui.GetAncestor(hwnd, 3) # GA_ROOTOWNER = 3 + if root: + hwnd = root + try: + _, pid = win32process.GetWindowThreadProcessId(hwnd) + proc = psutil.Process(pid) + exe_path = proc.exe() + return { + "bundleId": Path(exe_path).stem, + "displayName": proc.name(), + } + except Exception: + return frontmost_app() + + +def find_window_displays(bundle_ids: list[str]) -> list[dict[str, Any]]: + """For each bundleId, find which display(s) its windows are on.""" + if not bundle_ids: + return [] + + displays = get_displays() + windows = list_windows() + + # Build exe-stem -> ownerName mapping + names_by_bundle: dict[str, str | None] = {} + for bid in bundle_ids: + names_by_bundle[bid] = app_display_name(bid) + + result = [] + for bundle_id in bundle_ids: + target_name = names_by_bundle.get(bundle_id) + display_ids: set[int] = set() + for window in windows: + owner = window["ownerName"] + if not owner: + continue + # Match by exe name + owner_stem = Path(owner).stem if owner.endswith(".exe") else owner + if target_name and owner != target_name and owner_stem != bundle_id: + continue + if not target_name and owner_stem != bundle_id and owner != bundle_id: + continue + # Check which displays this window overlaps + wx = window["bounds"]["x"] + wy = window["bounds"]["y"] + ww = window["bounds"]["width"] + wh = window["bounds"]["height"] + for display in displays: + dx = display["originX"] + dy = display["originY"] + dw = display["width"] + dh = display["height"] + # Check rectangle intersection + if wx < dx + dw and wx + ww > dx and wy < dy + dh and wy + wh > dy: + display_ids.add(int(display["displayId"])) + result.append({"bundleId": bundle_id, "displayIds": sorted(display_ids)}) + return result + + +def open_app(bundle_id: str) -> None: + """Open an application by its bundleId (exe path or program name).""" + # Try to find the exe path from registry + import winreg + exe_path = None + + reg_paths = [ + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall"), + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"), + ] + for hive, sub_key in reg_paths: + try: + key = winreg.OpenKey(hive, sub_key) + app_key = winreg.OpenKey(key, bundle_id) + try: + exe_path = winreg.QueryValueEx(app_key, "DisplayIcon")[0] + if exe_path and "," in exe_path: + exe_path = exe_path.split(",")[0] + except OSError: + try: + exe_path = winreg.QueryValueEx(app_key, "InstallLocation")[0] + except OSError: + pass + winreg.CloseKey(app_key) + winreg.CloseKey(key) + if exe_path: + break + except OSError: + continue + + if exe_path and Path(exe_path).exists(): + os.startfile(exe_path) + else: + # Fallback: try to run it directly + try: + subprocess.Popen([bundle_id], shell=True) + except Exception: + raise RuntimeError(f"App not found for identifier: {bundle_id}") + + +# --------------------------------------------------------------------------- +# Clipboard (pyperclip — cross-platform) +# --------------------------------------------------------------------------- + +def read_clipboard() -> str: + import pyperclip + try: + return pyperclip.paste() or "" + except Exception: + return "" + + +def write_clipboard(text: str) -> None: + import pyperclip + pyperclip.copy(text) + + +# --------------------------------------------------------------------------- +# Permissions — Windows doesn't have macOS-style TCC +# --------------------------------------------------------------------------- + +def check_permissions() -> dict[str, bool | None]: + """Windows does not require explicit accessibility/screen-recording + permissions like macOS TCC. Always report as granted.""" + return { + "accessibility": True, + "screenRecording": True, + } + + +# --------------------------------------------------------------------------- +# Input actions (pyautogui — identical to mac_helper) +# --------------------------------------------------------------------------- + +def click(x: int, y: int, button: str, count: int, modifiers: list[str] | None) -> None: + pyautogui.moveTo(x, y) + if modifiers: + normalized = [normalize_key(m) for m in modifiers] + for key in normalized: + pyautogui.keyDown(key) + try: + pyautogui.click(x=x, y=y, button=button, clicks=count, interval=0.08) + finally: + for key in reversed(normalized): + pyautogui.keyUp(key) + else: + pyautogui.click(x=x, y=y, button=button, clicks=count, interval=0.08) + + +def scroll(x: int, y: int, delta_x: int, delta_y: int) -> None: + pyautogui.moveTo(x, y) + if delta_y: + pyautogui.scroll(int(delta_y), x=x, y=y) + if delta_x: + pyautogui.hscroll(int(delta_x), x=x, y=y) + + +def key_action(sequence: str, repeat: int = 1) -> None: + parts = [normalize_key(part) for part in sequence.split("+") if part.strip()] + for _ in range(max(1, repeat)): + if len(parts) == 1: + pyautogui.press(parts[0]) + else: + pyautogui.hotkey(*parts, interval=0.02) + time.sleep(0.01) + + +def hold_keys(keys: list[str], duration_ms: int) -> None: + normalized = [normalize_key(k) for k in keys] + for key in normalized: + pyautogui.keyDown(key) + try: + time.sleep(max(duration_ms, 0) / 1000) + finally: + for key in reversed(normalized): + pyautogui.keyUp(key) + + +def type_text(text: str) -> None: + pyautogui.write(text, interval=0.008) + + +# --------------------------------------------------------------------------- +# Main dispatcher — exact same command protocol as mac_helper.py +# --------------------------------------------------------------------------- + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("command") + parser.add_argument("--payload", default="{}") + args = parser.parse_args() + payload = json.loads(args.payload) + + try: + command = args.command + if command == "check_permissions": + perms = check_permissions() + json_output({"ok": True, "result": perms}) + return 0 + if command == "list_displays": + json_output({"ok": True, "result": get_displays()}) + return 0 + if command == "get_display_size": + json_output({"ok": True, "result": choose_display(payload.get("displayId"))}) + return 0 + if command == "screenshot": + resize = None + if payload.get("targetWidth") and payload.get("targetHeight"): + resize = (int(payload["targetWidth"]), int(payload["targetHeight"])) + result = capture_display(payload.get("displayId"), resize) + json_output({"ok": True, "result": result}) + return 0 + if command == "resolve_prepare_capture": + resize = None + if payload.get("targetWidth") and payload.get("targetHeight"): + resize = (int(payload["targetWidth"]), int(payload["targetHeight"])) + result = capture_display(payload.get("preferredDisplayId"), resize) + result["hidden"] = [] + result["resolvedDisplayId"] = result["displayId"] + json_output({"ok": True, "result": result}) + return 0 + if command == "zoom": + resize = None + if payload.get("targetWidth") and payload.get("targetHeight"): + resize = (int(payload["targetWidth"]), int(payload["targetHeight"])) + region = { + "left": int(payload["x"]), + "top": int(payload["y"]), + "width": int(payload["width"]), + "height": int(payload["height"]), + } + json_output({"ok": True, "result": capture_region(region, resize)}) + return 0 + if command == "prepare_for_action": + json_output({"ok": True, "result": []}) + return 0 + if command == "preview_hide_set": + json_output({"ok": True, "result": []}) + return 0 + if command == "find_window_displays": + json_output({"ok": True, "result": find_window_displays(list(payload.get("bundleIds") or []))}) + return 0 + if command == "key": + key_action(str(payload["keySequence"]), int(payload.get("repeat") or 1)) + json_output({"ok": True, "result": True}) + return 0 + if command == "hold_key": + hold_keys(list(payload.get("keyNames") or []), int(payload.get("durationMs") or 0)) + json_output({"ok": True, "result": True}) + return 0 + if command == "type": + type_text(str(payload.get("text") or "")) + json_output({"ok": True, "result": True}) + return 0 + if command == "click": + click(int(payload["x"]), int(payload["y"]), str(payload.get("button") or "left"), int(payload.get("count") or 1), payload.get("modifiers")) + json_output({"ok": True, "result": True}) + return 0 + if command == "drag": + from_point = payload.get("from") + if from_point: + pyautogui.moveTo(int(from_point["x"]), int(from_point["y"])) + pyautogui.dragTo(int(payload["to"]["x"]), int(payload["to"]["y"]), duration=0.2, button="left") + json_output({"ok": True, "result": True}) + return 0 + if command == "move_mouse": + pyautogui.moveTo(int(payload["x"]), int(payload["y"])) + json_output({"ok": True, "result": True}) + return 0 + if command == "scroll": + scroll(int(payload["x"]), int(payload["y"]), int(payload.get("deltaX") or 0), int(payload.get("deltaY") or 0)) + json_output({"ok": True, "result": True}) + return 0 + if command == "mouse_down": + pyautogui.mouseDown(button="left") + json_output({"ok": True, "result": True}) + return 0 + if command == "mouse_up": + pyautogui.mouseUp(button="left") + json_output({"ok": True, "result": True}) + return 0 + if command == "cursor_position": + x, y = pyautogui.position() + json_output({"ok": True, "result": {"x": int(x), "y": int(y)}}) + return 0 + if command == "frontmost_app": + json_output({"ok": True, "result": frontmost_app()}) + return 0 + if command == "app_under_point": + json_output({"ok": True, "result": app_under_point(int(payload["x"]), int(payload["y"]))}) + return 0 + if command == "list_installed_apps": + json_output({"ok": True, "result": installed_apps()}) + return 0 + if command == "list_running_apps": + json_output({"ok": True, "result": running_apps()}) + return 0 + if command == "open_app": + open_app(str(payload["bundleId"])) + json_output({"ok": True, "result": True}) + return 0 + if command == "read_clipboard": + json_output({"ok": True, "result": read_clipboard()}) + return 0 + if command == "write_clipboard": + write_clipboard(str(payload.get("text") or "")) + json_output({"ok": True, "result": True}) + return 0 + error_output(f"Unknown command: {command}", code="bad_command") + return 2 + except Exception as exc: + error_output(str(exc)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/server/api/computer-use.ts b/src/server/api/computer-use.ts index 94fe6a86..fed13ea0 100644 --- a/src/server/api/computer-use.ts +++ b/src/server/api/computer-use.ts @@ -12,9 +12,11 @@ import { access, readFile, mkdir, writeFile } from 'fs/promises' import { createHash } from 'crypto' import path from 'path' import { fileURLToPath } from 'url' -// Embed mac_helper.py at compile time so it's available in bundled mode +// Embed helper scripts at compile time so they're available in bundled mode // @ts-ignore — Bun text import import MAC_HELPER_CONTENT from '../../../runtime/mac_helper.py' with { type: 'text' } +// @ts-ignore — Bun text import +import WIN_HELPER_CONTENT from '../../../runtime/win_helper.py' with { type: 'text' } const __dirname = path.dirname(fileURLToPath(import.meta.url)) const projectRoot = path.resolve(__dirname, '../../..') @@ -24,9 +26,8 @@ const runtimeStateRoot = join(claudeHome, '.runtime') const venvRoot = join(runtimeStateRoot, 'venv') const installStampPath = join(runtimeStateRoot, 'requirements.sha256') -// Embedded content of requirements.txt — kept in sync with runtime/requirements.txt. -// This ensures the bundled sidecar can create the file without disk access. -const REQUIREMENTS_CONTENT = `mss>=10.1.0 +// Embedded content of requirements — platform-specific. +const REQUIREMENTS_DARWIN = `mss>=10.1.0 Pillow>=11.3.0 pyautogui>=0.9.54 pyobjc-core>=11.1 @@ -34,6 +35,18 @@ pyobjc-framework-Cocoa>=11.1 pyobjc-framework-Quartz>=11.1 ` +const REQUIREMENTS_WIN32 = `mss>=10.1.0 +Pillow>=11.3.0 +pyautogui>=0.9.54 +pywin32>=306 +psutil>=5.9.0 +pyperclip>=1.8.2 +screeninfo>=0.8.1 +` + +const isWindows = process.platform === 'win32' +const REQUIREMENTS_CONTENT = isWindows ? REQUIREMENTS_WIN32 : REQUIREMENTS_DARWIN + // 清华大学 PyPI 镜像，国内安装速度更快 const PIP_INDEX_URL = 'https://pypi.tuna.tsinghua.edu.cn/simple/' const PIP_TRUSTED_HOST = 'pypi.tuna.tsinghua.edu.cn' @@ -43,11 +56,12 @@ function getRequirementsPath(): string { return join(runtimeStateRoot, 'requirements.txt') } +function getHelperFileName(): string { + return isWindows ? 'win_helper.py' : 'mac_helper.py' +} + function getHelperPath(): string { - // In bundled mode mac_helper.py is extracted to runtimeStateRoot. - // In dev mode we also copy it there during setup, so both modes - // read from the same location after setup runs. - return join(runtimeStateRoot, 'mac_helper.py') + return join(runtimeStateRoot, getHelperFileName()) } async function pathExists(target: string): Promise { @@ -92,8 +106,9 @@ async function ensureRuntimeFiles(): Promise { // requirements.txt — always write from embedded constant (authoritative) await writeFile(getRequirementsPath(), REQUIREMENTS_CONTENT, 'utf8') - // mac_helper.py — always write from embedded content (compile-time import) - await writeFile(getHelperPath(), MAC_HELPER_CONTENT, 'utf8') + // helper script — write the platform-appropriate version + const helperContent = isWindows ? WIN_HELPER_CONTENT : MAC_HELPER_CONTENT + await writeFile(getHelperPath(), helperContent, 'utf8') } type EnvStatus = { @@ -120,10 +135,11 @@ type EnvStatus = { async function checkStatus(): Promise { const platform = process.platform - const supported = platform === 'darwin' + const supported = platform === 'darwin' || platform === 'win32' - // Check Python 3 - const pythonResult = await runCommand('python3', ['--version']) + // Check Python 3 — Windows may only have `python`, not `python3` + const pythonCmd = isWindows ? 'python' : 'python3' + const pythonResult = await runCommand(pythonCmd, ['--version']) const pythonInstalled = pythonResult.ok const pythonVersion = pythonInstalled ? pythonResult.stdout.replace('Python ', '') @@ -131,12 +147,16 @@ async function checkStatus(): Promise { let pythonPath: string | null = null if (pythonInstalled) { - const whichResult = await runCommand('which', ['python3']) - pythonPath = whichResult.ok ? whichResult.stdout : null + const whichCmd = isWindows ? 'where' : 'which' + const whichResult = await runCommand(whichCmd, [pythonCmd]) + pythonPath = whichResult.ok ? whichResult.stdout.split('\n')[0] : null } - // Check venv - const venvCreated = await pathExists(join(venvRoot, 'bin', 'python3')) + // Check venv — different paths on Windows vs Unix + const venvPython = isWindows + ? join(venvRoot, 'Scripts', 'python.exe') + : join(venvRoot, 'bin', 'python3') + const venvCreated = await pathExists(venvPython) // Check dependencies — use the state dir copy const reqPath = getRequirementsPath() @@ -162,8 +182,7 @@ async function checkStatus(): Promise { try { await ensureRuntimeFiles() } catch {} const helperPath = getHelperPath() if (await pathExists(helperPath)) { - const pythonBin = join(venvRoot, 'bin', 'python3') - const permResult = await runCommand(pythonBin, [helperPath, 'check_permissions']) + const permResult = await runCommand(venvPython, [helperPath, 'check_permissions']) if (permResult.ok) { try { const parsed = JSON.parse(permResult.stdout) @@ -194,8 +213,9 @@ type SetupResult = { async function runSetup(): Promise { const steps: SetupResult['steps'] = [] - // Step 1: Check python3 - const pythonCheck = await runCommand('python3', ['--version']) + // Step 1: Check python + const pythonCmd = isWindows ? 'python' : 'python3' + const pythonCheck = await runCommand(pythonCmd, ['--version']) if (!pythonCheck.ok) { steps.push({ name: 'python_check', @@ -224,9 +244,12 @@ async function runSetup(): Promise { } // Step 3: Create venv - const venvExists = await pathExists(join(venvRoot, 'bin', 'python3')) + const venvPython = isWindows + ? join(venvRoot, 'Scripts', 'python.exe') + : join(venvRoot, 'bin', 'python3') + const venvExists = await pathExists(venvPython) if (!venvExists) { - const venvResult = await runCommand('python3', ['-m', 'venv', venvRoot]) + const venvResult = await runCommand(pythonCmd, ['-m', 'venv', venvRoot]) if (!venvResult.ok) { steps.push({ name: 'venv', @@ -241,10 +264,11 @@ async function runSetup(): Promise { } // Step 4: Ensure pip - const pipPath = join(venvRoot, 'bin', 'pip') + const pipPath = isWindows + ? join(venvRoot, 'Scripts', 'pip.exe') + : join(venvRoot, 'bin', 'pip') if (!(await pathExists(pipPath))) { - const pythonBin = join(venvRoot, 'bin', 'python3') - const pipResult = await runCommand(pythonBin, [ + const pipResult = await runCommand(venvPython, [ '-m', 'ensurepip', '--upgrade', @@ -271,16 +295,14 @@ async function runSetup(): Promise { } catch {} if (installedDigest !== digest) { - const pythonBin = join(venvRoot, 'bin', 'python3') - // Upgrade pip first (using China mirror) - await runCommand(pythonBin, [ + await runCommand(venvPython, [ '-m', 'pip', 'install', '--upgrade', 'pip', '-i', PIP_INDEX_URL, '--trusted-host', PIP_TRUSTED_HOST, ]) // Install deps (using China mirror) - const installResult = await runCommand(pythonBin, [ + const installResult = await runCommand(venvPython, [ '-m', 'pip', 'install', '-r', reqPath, '-i', PIP_INDEX_URL, '--trusted-host', PIP_TRUSTED_HOST, @@ -343,7 +365,9 @@ async function saveConfig(config: ComputerUseConfig): Promise { async function listInstalledApps(): Promise<{ bundleId: string; displayName: string; path: string }[]> { const helperPath = getHelperPath() - const pythonBin = join(venvRoot, 'bin', 'python3') + const pythonBin = isWindows + ? join(venvRoot, 'Scripts', 'python.exe') + : join(venvRoot, 'bin', 'python3') if (!(await pathExists(pythonBin)) || !(await pathExists(helperPath))) { return [] @@ -407,19 +431,25 @@ export async function handleComputerUseApi( } } - // POST /api/computer-use/open-settings — open macOS System Settings pane + // POST /api/computer-use/open-settings — open system settings pane if (action === 'open-settings' && req.method === 'POST') { - if (process.platform !== 'darwin') { - return Response.json({ error: 'macOS only' }, { status: 400 }) - } const body = (await req.json().catch(() => ({}))) as { pane?: string } const pane = body.pane ?? 'Privacy_ScreenCapture' const allowed = ['Privacy_ScreenCapture', 'Privacy_Accessibility'] if (!allowed.includes(pane)) { return Response.json({ error: 'Invalid pane' }, { status: 400 }) } - const url = `x-apple.systempreferences:com.apple.preference.security?${pane}` - await runCommand('open', [url]) + + if (process.platform === 'darwin') { + const url = `x-apple.systempreferences:com.apple.preference.security?${pane}` + await runCommand('open', [url]) + } else if (process.platform === 'win32') { + // Windows doesn't need privacy settings like macOS TCC, but we can + // open the general privacy page if requested + await runCommand('cmd', ['/c', 'start', 'ms-settings:privacy']) + } else { + return Response.json({ error: 'Unsupported platform' }, { status: 400 }) + } return Response.json({ ok: true }) } diff --git a/src/utils/computerUse/pythonBridge.ts b/src/utils/computerUse/pythonBridge.ts index 02a710bc..b99cbc8f 100644 --- a/src/utils/computerUse/pythonBridge.ts +++ b/src/utils/computerUse/pythonBridge.ts @@ -19,14 +19,19 @@ const installStampPath = path.join(runtimeStateRoot, 'requirements.sha256') const PIP_INDEX_URL = 'https://pypi.tuna.tsinghua.edu.cn/simple/' const PIP_TRUSTED_HOST = 'pypi.tuna.tsinghua.edu.cn' +const isWindows = process.platform === 'win32' + // Always read from ~/.claude/.runtime/ — works in both dev and bundled mode. const requirementsPath = path.join(runtimeStateRoot, 'requirements.txt') -const helperPath = path.join(runtimeStateRoot, 'mac_helper.py') +const helperFileName = isWindows ? 'win_helper.py' : 'mac_helper.py' +const helperPath = path.join(runtimeStateRoot, helperFileName) let bootstrapPromise: Promise | undefined function pythonBinPath(): string { - return path.join(venvRoot, 'bin', 'python3') + return isWindows + ? path.join(venvRoot, 'Scripts', 'python.exe') + : path.join(venvRoot, 'bin', 'python3') } async function pathExists(target: string): Promise { @@ -54,8 +59,9 @@ async function runOrThrow(file: string, args: string[], label: string): Promise< async function ensureRuntimeFiles(): Promise { await mkdir(runtimeStateRoot, { recursive: true }) - const devRequirements = path.join(projectRoot, 'runtime', 'requirements.txt') - const devHelper = path.join(projectRoot, 'runtime', 'mac_helper.py') + const devReqFile = isWindows ? 'requirements-win.txt' : 'requirements.txt' + const devRequirements = path.join(projectRoot, 'runtime', devReqFile) + const devHelper = path.join(projectRoot, 'runtime', helperFileName) // Always sync from dev runtime/ so source changes are reflected immediately. // Previously this only copied when the dest was missing, causing stale files @@ -77,10 +83,14 @@ export async function ensureBootstrapped(): Promise { if (!(await pathExists(pythonBinPath()))) { logForDebugging('creating runtime venv at %s', { level: 'debug' }) - await runOrThrow('python3', ['-m', 'venv', venvRoot], 'python venv creation') + const pythonCmd = isWindows ? 'python' : 'python3' + await runOrThrow(pythonCmd, ['-m', 'venv', venvRoot], 'python venv creation') } - if (!(await pathExists(path.join(venvRoot, 'bin', 'pip')))) { + const pipBin = isWindows + ? path.join(venvRoot, 'Scripts', 'pip.exe') + : path.join(venvRoot, 'bin', 'pip') + if (!(await pathExists(pipBin))) { logForDebugging('bootstrapping pip with ensurepip', { level: 'debug' }) await runOrThrow(pythonBinPath(), ['-m', 'ensurepip', '--upgrade'], 'ensurepip') }