sdaduanbilei/cc-haha
1
0
Fork 0
mirror of https://github.com/NanmiCoder/cc-haha synced 2026-07-16 13:03:31 +08:00
Code Issues Packages Projects Releases Wiki Activity
cc-haha/runtime/win_helper.py
程序员阿江(Relakkes) 5cd6b5d07b Prevent desktop Computer Use from stalling on missing approvals and unstable text input 
Desktop sessions were missing a visible request_access approval path and could
mis-detect their own app window as an unapproved frontmost target, which caused
Computer Use clicks to fail even after opening the intended app. On macOS, text
entry was also split across inconsistent clipboard and keystroke paths, making
Electron inputs unreliable for Chinese and short strings.

This change adds a desktop approval bridge over the existing session websocket,
renders a dedicated desktop approval modal, threads the real desktop bundle id
into the Computer Use executor, and switches macOS clipboard typing onto the
native pasteboard plus system paste shortcut path. It also makes tool error
results expandable in the desktop chat UI so frontmost-gate failures are fully
visible during debugging.

Constraint: Desktop sessions run the CLI over the SDK websocket path, so Ink tool JSX dialogs are not visible there
Constraint: macOS IME and Electron text inputs are unreliable with pyautogui.write and generic hotkey synthesis
Rejected: Reuse CLI setToolJSX dialogs in desktop mode | no transport for mid-call Ink UI over the SDK bridge
Rejected: Keep shell pbcopy/pbpaste for clipboard typing | inconsistent with NSPasteboard path and less reliable for Chinese text
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: Keep desktop Computer Use approvals and macOS text-entry behavior on a single bridge/path; avoid reintroducing separate CLI-only and desktop-only codepaths for the same action
Tested: python3 -m unittest runtime/test_helpers.py
Tested: bun test src/utils/computerUse/permissions.test.ts src/server/__tests__/conversation-service.test.ts
Tested: cd desktop && bun run test ComputerUsePermissionModal chatStore
Tested: cd desktop && bun run test chatBlocks
Tested: cd desktop && bun run lint
Not-tested: End-to-end manual Computer Use interaction against a live Electron target app on macOS
2026-04-20 12:12:02 +08:00

724 lines
25 KiB
Python
Raw Blame History

#!/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)
def paste_clipboard() -> None:
pyautogui.hotkey("ctrl", "v", interval=0.02)
# ---------------------------------------------------------------------------
# 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
if command == "paste_clipboard":
paste_clipboard()
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())
Reference in New Issue View Git Blame Copy Permalink