From 2d570cf30a0441bb58f6382eaf301ccab99fb9ed Mon Sep 17 00:00:00 2001 From: Syed Abdur Rehman Ali Date: Sat, 16 May 2026 01:11:35 +0530 Subject: [PATCH] feat(tools/computer_use): native per-OS desktop control (macOS/Linux/Windows) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three OS-specific tools — `computer_use_macos`, `computer_use_linux`, `computer_use_windows` — sharing one JSON schema and one set of action semantics, but with native backends per platform. Complementary to the containerised proposal in #15876 (which targets the "Hermes-runs-in-Docker" deployment shape) and the macOS-Anthropic-protocol work in #4562 / #13308. This PR owns the "Hermes runs natively on the host desktop, control any of the three majors with consistent abstraction" shape. Architecture ============ * All three tools register at module top via `registry.register()` so the AST tool-discovery picks them up. `check_fn` returns False off the matching platform / when `HERMES_COMPUTER_USE_ENABLED` is unset / when required deps are missing — so on a given host the model only sees the one tool it can actually use. * `computer_use_common.py` — schema, `ActionRequest`, `ActionResult`, parameter validation, screen-bounds enforcement. * `computer_use_safety.py` — env gate, kill-switch flag, JSONL action log under `$HERMES_HOME/logs/computer_use.jsonl`, screenshot redaction (PIL). * `computer_use_grammar.py` — one parser, four targets. `Cmd+Shift+T` produces Quartz CGEvent flags+keycode on macOS, `xdotool key` string on X11, ydotool input event codes on Wayland, Win32 VK codes on Windows. * `computer_use_macos.py` — Quartz `CGEvent` for input, `screencapture` CLI for capture, `CGWindowListCopyWindowInfo` for active window. pyobjc-framework-Quartz is the only new dep. * `computer_use_linux.py` — runtime detection of X11 vs Wayland. X11 → `xdotool` + `scrot`/`import`. Wayland → `ydotool` + `grim` (wlroots) / `gnome-screenshot` / `spectacle`. Active-window queries via Sway IPC / hyprctl / xdotool depending on path. * `computer_use_windows.py` — `ctypes` over `user32.SendInput` (modern path; avoids legacy `keybd_event`). DPI-aware on import. Screenshot via `mss` if installed, falls back to ctypes BitBlt + PIL otherwise. Skills ====== Per-OS skill teaches the model what's actually different on each host: Cmd-vs-Ctrl, Spotlight vs Win+S, X11 vs Wayland detection, UAC/UIPI, accessibility / screen-recording perm setup, etc. The common skill covers when to reach for `computer_use_*` at all (vs `browser_tool` / `terminal`) and the screenshot-first discipline. Validation ========== * 56/56 unit tests passing (mocked Quartz / subprocess / user32 across all three backends + grammar + safety). * macOS backend integration-tested live on the author's MacBook: screen_size, cursor_position, get_active_window, screenshot (full), screenshot (region crop), screenshot (with redact), wait, off-screen click validation, type-without-text validation, unknown-action validation, env-off refusal — all 11/11 cases pass. * Linux + Windows are unit-test-only at the moment; author has no Linux or Windows host immediately available for end-to-end validation. Honest framing in the eventual PR body. Safety posture ============== * `HERMES_COMPUTER_USE_ENABLED=true` required. Default: refused. * Action allowlist + per-action validation (no off-screen, no >10K type strings, no >30s waits, no unknown actions). * Process-global kill-switch flag (`set_kill_switch()`) checked before every action — engaged once, all subsequent actions refuse until cleared. * JSONL audit log of every attempt (action, params minus image bytes, success bit, error if any). * `screenshot` action accepts `redact_regions` to blank rectangles (password manager, MFA codes) before the image reaches the model. --- skills/computer-use/common/SKILL.md | 90 +++++ skills/computer-use/linux/SKILL.md | 123 ++++++ skills/computer-use/macos/SKILL.md | 88 +++++ skills/computer-use/windows/SKILL.md | 99 +++++ tests/tools/test_computer_use.py | 561 ++++++++++++++++++++++++++- tools/computer_use_common.py | 289 ++++++++++++++ tools/computer_use_grammar.py | 322 +++++++++++++++ tools/computer_use_linux.py | 552 ++++++++++++++++++++++++++ tools/computer_use_macos.py | 375 ++++++++++++++++++ tools/computer_use_safety.py | 179 +++++++++ tools/computer_use_windows.py | 492 +++++++++++++++++++++++ 11 files changed, 3169 insertions(+), 1 deletion(-) create mode 100644 skills/computer-use/common/SKILL.md create mode 100644 skills/computer-use/linux/SKILL.md create mode 100644 skills/computer-use/macos/SKILL.md create mode 100644 skills/computer-use/windows/SKILL.md create mode 100644 tools/computer_use_common.py create mode 100644 tools/computer_use_grammar.py create mode 100644 tools/computer_use_linux.py create mode 100644 tools/computer_use_macos.py create mode 100644 tools/computer_use_safety.py create mode 100644 tools/computer_use_windows.py diff --git a/skills/computer-use/common/SKILL.md b/skills/computer-use/common/SKILL.md new file mode 100644 index 0000000000000..018bee8eb5c31 --- /dev/null +++ b/skills/computer-use/common/SKILL.md @@ -0,0 +1,90 @@ +--- +name: computer-use +description: When and how to use the native desktop computer_use_* tool — screenshot first, click by absolute pixel, never reach for it when browser_tool or terminal will do. +metadata: + hermes: + tags: [desktop, mouse, keyboard, click, screenshot, gui, automation, computer-use] +--- + +# Native desktop control (`computer_use_*`) + +You have a desktop-control tool that takes screenshots, clicks, types, and sends key combinations on the host machine. The exact tool name depends on the OS the agent is running on: + +* `computer_use_macos` — when the host is macOS +* `computer_use_linux` — when the host is Linux (X11 or Wayland) +* `computer_use_windows` — when the host is Windows + +Only one of these is registered per session — whichever matches the host. Don't try to call a different one. The tool surface (parameters, action names, return shape) is identical across the three; only the OS-specific shortcuts and idioms differ — see the per-OS skill for that. + +## When to use it + +Reach for `computer_use_*` only when **simpler tools genuinely can't do the job**: + +- `browser_tool` / `browser_camofox` — already covers any web workflow. Don't drive a browser through screenshots when CDP gives you the DOM. +- `terminal` — covers anything a CLI can do. Don't click through a GUI installer when a `brew install` / `apt install` / `winget install` line exists. +- `file_*` tools — for reading/writing files on disk. + +`computer_use_*` is the right tool when: + +- The target is a **native desktop app** with no usable CLI or web surface (Adobe apps, Office desktop, native installers, system settings panels). +- You need to **interact with a popup, modal, or system dialog** that lives outside any controllable surface. +- The user explicitly asks you to "click", "open this app", "drag", "use the GUI". +- A vision-on-dense-UI workflow benefits from screenshot grounding (proofreading a slide layout, validating a render in a 3D app). + +## When NOT to use it + +- Web tasks → use `browser_tool` instead. Faster, more reliable, no screenshot ambiguity. +- Anything scriptable via shell → use `terminal`. +- File reads / edits → use `file_*`. +- Anything touching credentials, password fields, MFA codes, or banking UIs unless the user has explicitly asked. Even then: prefer not to. Logs of synthetic clicks near sensitive UI elements are a footgun. + +## The screenshot-first discipline + +Every desktop-control session starts the same way: + +1. **Screenshot first.** Always. You don't know what's on the screen until you look. A click at coordinates you guessed from "well the button is usually in the top-right" lands somewhere wrong about half the time. +2. **Identify the target visually** in the screenshot. Note its approximate pixel position. +3. **Take the action** at those absolute coordinates. +4. **Screenshot again** after any non-trivial action (window opened, dialog appeared, focus changed) to confirm the world looks like you expected. +5. **Adjust or recover** if it doesn't. + +This is slow. That's the price. Skipping it is how computer-use agents go off the rails. + +## Coordinates + +Coordinates are **absolute screen pixels**, origin at top-left. On HiDPI / Retina displays, the tool already runs in DPI-aware mode; the numbers you see in screenshots are the numbers you click. No scaling math. + +## Cost discipline + +- A screenshot costs an action and adds a base64 PNG (often >100KB) to the next turn's context. Don't take six in a row. +- A `wait` between actions is sometimes necessary (window opens, network roundtrip, animation completes) but each `wait` costs latency too. Use `ms: 200` or `500`, not `5000`. +- If a target app needs many clicks to do something a CLI command would do in one line, **switch to terminal** mid-task. There's no shame. + +## Safety + +The tool is **opt-in by env var**: `HERMES_COMPUTER_USE_ENABLED=true` must be set on the host. If it isn't, every call returns a refusal — that's by design, not a bug. Tell the user how to enable it; don't try to enable it yourself. + +A `redact_regions` parameter on `screenshot` lets you blank rectangles (e.g. password manager popup, MFA code) before the image reaches the model. Use it when you can identify a sensitive zone in advance. + +Every action attempt is logged to `$HERMES_HOME/logs/computer_use.jsonl`. If you do something the user doesn't expect, the log is the audit trail. + +## Reading the result + +Every action returns a JSON dict. Useful fields: + +- `success` — bool. Always check this; non-zero `error` means the action didn't happen. +- `screenshot_b64` — base64 PNG (only on `screenshot`). +- `cursor` — `{x, y}` after the action. +- `screen` — `{width, height}` of the primary display. +- `active_window` — `{app, title}` (best-effort; some Wayland compositors return empty). +- `error` — a string if the action failed; surface it to the user, don't silently retry. + +## Per-OS skills + +For platform-specific shortcuts, screenshot tooling, and gotchas (Cmd-vs-Ctrl, X11-vs-Wayland, UAC dialogs, accessibility permissions), load the skill that matches the host: + +- `computer-use-macos` +- `computer-use-linux` +- `computer-use-windows` + +You typically load the per-OS skill once at the start of a desktop-control task and follow this common skill for the discipline. diff --git a/skills/computer-use/linux/SKILL.md b/skills/computer-use/linux/SKILL.md new file mode 100644 index 0000000000000..da1eec4556175 --- /dev/null +++ b/skills/computer-use/linux/SKILL.md @@ -0,0 +1,123 @@ +--- +name: computer-use-linux +description: Linux X11 vs Wayland gotchas, xdotool/ydotool/grim setup, and DE-specific shortcuts for computer_use_linux. +metadata: + hermes: + tags: [linux, x11, wayland, xdotool, ydotool, grim, gnome, kde, computer-use] +--- + +# Linux desktop control — what's different + +You have access to `computer_use_linux`. Linux is the awkward one because it's two display servers in a trench coat. Same agent-facing API, two completely different toolchains under the hood. + +## Detecting the active session + +The tool detects the display server automatically each call from `WAYLAND_DISPLAY` and `XDG_SESSION_TYPE`. You don't choose. You can confirm what the host is using with a short terminal call: + +``` +echo "$XDG_SESSION_TYPE" # x11 or wayland +``` + +If the user can choose, **X11 is more capable** for scripted automation right now. Wayland's app-isolation model deliberately makes synthetic input and full-screen capture harder. If you have the choice and the workflow is automation-heavy, suggest the user log in to an X11 session. + +## One-time host setup + +### X11 path (preferred when available) + +The host needs `xdotool` and a screenshot tool — usually `scrot`, sometimes `imagemagick` (which provides `import`). On Debian/Ubuntu/Mint: + +``` +sudo apt install xdotool scrot xdpyinfo xprop +``` + +On Arch/Manjaro: + +``` +sudo pacman -S xdotool scrot xorg-xdpyinfo xorg-xprop +``` + +After install everything works without further setup; xdotool synthesises events through XTEST. + +### Wayland path (when X11 isn't available) + +Wayland needs `ydotool` (input via `/dev/uinput`) and one of `grim` (wlroots compositors: Sway, Hyprland, labwc, river), `gnome-screenshot` (GNOME), or `spectacle` (KDE). + +``` +# Sway / Hyprland / labwc +sudo apt install ydotool grim +sudo usermod -aG input "$USER" # for /dev/uinput access +sudo systemctl enable --now ydotoold + +# GNOME on Wayland +sudo apt install ydotool gnome-screenshot +sudo systemctl enable --now ydotoold + +# KDE on Wayland +sudo apt install ydotool kde-spectacle +sudo systemctl enable --now ydotoold +``` + +The `usermod` change requires re-login to take effect. If the operator hasn't done this, every input action will fail with a permission error on `/dev/uinput`. + +## Modifier keys + +Linux uses **Ctrl** for the same shortcuts macOS uses **Cmd** for. The grammar parser accepts `Ctrl+T`, `ctrl+t`, `control+t` — all equivalent. The macOS `Cmd` token is reinterpreted as the **Super** (Windows) key on Linux, which is what the user wants in practice when porting muscle-memory. + +| Action | Combo | +|---|---| +| New tab / new window / save / close | `Ctrl+T` / `Ctrl+N` / `Ctrl+S` / `Ctrl+W` | +| Cut / copy / paste / undo | `Ctrl+X` / `Ctrl+C` / `Ctrl+V` / `Ctrl+Z` | +| Find | `Ctrl+F` | +| Switch window (most DEs) | `Alt+Tab` | +| Open terminal in many DEs | `Ctrl+Alt+T` | +| Lock screen (GNOME / KDE) | `Super+L` / `Ctrl+Alt+L` | +| Activities / overview (GNOME) | `Super` | +| App launcher (KDE) | `Alt+F1` | + +In a terminal **Ctrl+C is interrupt**, not copy. Use `Ctrl+Shift+C` / `Ctrl+Shift+V` for clipboard inside terminal emulators. + +## Active-window queries + +- **X11**: `get_active_window` returns `{id, title, app}` derived from `xdotool getactivewindow` and `xprop WM_CLASS`. Reliable. +- **Wayland (Sway)**: returns the `app_id` and window name from the i3 IPC tree. Reliable. +- **Wayland (Hyprland)**: returns from `hyprctl -j activewindow`. Reliable. +- **Wayland (GNOME)**: there is no public IPC for this. Returns empty `{}`. Don't depend on it. +- **Wayland (KDE)**: best-effort via KWin scripting; often empty. + +## Screenshot quirks + +- **X11**: `scrot` returns the full root window — works on multi-monitor setups, captures everything. +- **Wayland (wlroots / `grim`)**: full virtual desktop including all outputs. +- **Wayland (GNOME / `gnome-screenshot`)**: full screen of the focused monitor; multi-monitor capture is a known gap. +- **Wayland (KDE / `spectacle`)**: full screen of all outputs. + +`gnome-screenshot` produces a flash + shutter sound by default. There's no reliable way to suppress it from the API; warn the user once if it bothers them. + +## DE-specific things to know + +| DE | Distinctive feature | Watch out for | +|---|---|---| +| GNOME (Wayland) | Activities overview opens with `Super` | No window-position queries; `move`/`resize` programmatic control is limited | +| KDE Plasma | Most flexible; rich KWin scripting | Spectacle screenshot is async — add `wait 300` after triggering | +| Sway / Hyprland / labwc (wlroots) | Best Wayland tooling | Tiling — clicks at fixed coordinates may target wrong window if user resizes | +| Xfce / MATE / Cinnamon (X11) | Just works with xdotool | None significant | +| Unity / Pantheon | X11 — works but DE-specific shortcuts vary | Some shortcuts are DE-overridden | + +## Don't try to do these + +- **Sudo password prompts** in graphical password dialogs (polkit / pkexec) — the focus locks out synthetic input as a security measure. Use `sudo` over terminal instead, or ask the user. +- **Wayland security keyrings** (gnome-keyring unlock prompt) — same restriction. +- **VirtualKeyboard / OSK input** — these run in compositor-privileged space. + +## Example — open a terminal, run a command, screenshot + +``` +{"action": "key", "keys": "Ctrl+Alt+T"} +{"action": "wait", "ms": 500} +{"action": "type", "text": "uname -a"} +{"action": "key", "keys": "Return"} +{"action": "wait", "ms": 200} +{"action": "screenshot"} +``` + +For most CLI workflows just use the `terminal` tool — it's faster than driving a GUI terminal through screenshots. diff --git a/skills/computer-use/macos/SKILL.md b/skills/computer-use/macos/SKILL.md new file mode 100644 index 0000000000000..6009adda72e6d --- /dev/null +++ b/skills/computer-use/macos/SKILL.md @@ -0,0 +1,88 @@ +--- +name: computer-use-macos +description: macOS-specific shortcuts, screenshot tool, and accessibility/screen-recording permission setup for computer_use_macos. +metadata: + hermes: + tags: [macos, desktop, mouse, keyboard, accessibility, screen-recording, computer-use] +--- + +# macOS desktop control — what's different + +You have access to `computer_use_macos`. The action set and parameter shape are documented in the parent `computer-use` skill — load that first if you haven't. This skill covers macOS-only things you must know. + +## One-time host setup (the user does this, not you) + +Before `computer_use_macos` works at all, the operator needs to grant **two** permissions in *System Settings → Privacy & Security*: + +1. **Accessibility** — required for synthetic mouse and keyboard events to reach other apps. Without this, `CGEventPost` returns success but nothing actually happens — the events are silently dropped at the WindowServer. +2. **Screen Recording** — required for `screencapture` to include other apps' windows. Without this, screenshots show the desktop background and your own app's windows only — every other window is rendered as wallpaper, which is misleading rather than blank. + +Both prompts appear automatically the first time the tool runs. Tell the user this once at the start of a session if it looks like the permissions aren't granted (you can detect this by an action that "succeeds" but the screenshot doesn't reflect the click). + +The Hermes process needs both permissions — toggling them on is a per-binary grant, so if the user runs Hermes from a virtualenv vs system Python they'll need to grant it for whichever they're using. + +## Modifier keys + +macOS uses **Cmd** where Linux/Windows use **Ctrl** for almost every shortcut. The grammar parser accepts both `Cmd+...` (canonical for macOS) and `cmd+...`. Common combinations: + +| Action | Combo | +|---|---| +| New / open / save / close window | `Cmd+N` / `Cmd+O` / `Cmd+S` / `Cmd+W` | +| Cut / copy / paste / undo | `Cmd+X` / `Cmd+C` / `Cmd+V` / `Cmd+Z` | +| Find / find-next | `Cmd+F` / `Cmd+G` | +| Quit app | `Cmd+Q` | +| Switch app | `Cmd+Tab` | +| Switch window within app | `Cmd+~` | +| Spotlight | `Cmd+Space` | +| Mission Control | `Ctrl+Up` | +| Force quit | `Cmd+Option+Esc` | +| Full screenshot to clipboard | `Cmd+Shift+Ctrl+3` (rarely needed; use the tool's `screenshot` action) | + +Don't use `Ctrl+...` for app shortcuts on macOS unless the app is a Linux/Windows port that documented the Ctrl form (some IDEs do this, e.g. Cursor). + +## Spotlight is your friend + +To open any app reliably: + +1. `key Cmd+Space` — opens Spotlight. +2. `wait 200` — let it focus. +3. `type ` — narrow the result. +4. `wait 100` — let Spotlight resolve. +5. `key Return` — launch the top hit. + +This works whether or not the app is in the Dock and is much more reliable than clicking the Dock or hunting in Finder. + +## Active-window queries + +`get_active_window` returns `{app, title}` on macOS — the frontmost on-screen application name and the window title (when available). Some apps (especially Electron) don't expose a window title; expect empty strings sometimes. + +## Screenshot quirks + +- `screencapture` captures the full primary display. On a multi-monitor Mac you'll see only the primary; we don't currently expose multi-display capture. +- HiDPI / Retina screens return native-pixel screenshots (e.g. 3024×1964 on a 14" MacBook Pro). The pixel coordinates you click are in this same native space — no scaling. +- The first call shows a system permission prompt; subsequent calls are silent. + +## Don't try to do these + +- **Mission Control swipes** with `mouse_drag` — three-finger swipe is a trackpad-gesture-only interaction, not synthesisable through CGEvent. +- **Touch ID / Apple Watch unlock** — system-modal prompts that synthetic clicks can't pass through. +- **Quartz screen rotation / display arrangement** — those panes in System Settings have a UIPI-like restriction; ask the user instead. +- **Anything that requires admin privilege escalation** — the standard sudo/Authorization Services prompt won't accept synthetic password input. + +## Example — open Safari, navigate to a URL, screenshot + +``` +{"action": "key", "keys": "Cmd+Space"} +{"action": "wait", "ms": 200} +{"action": "type", "text": "Safari"} +{"action": "wait", "ms": 150} +{"action": "key", "keys": "Return"} +{"action": "wait", "ms": 800} +{"action": "key", "keys": "Cmd+L"} +{"action": "type", "text": "https://example.com"} +{"action": "key", "keys": "Return"} +{"action": "wait", "ms": 1500} +{"action": "screenshot"} +``` + +(For web tasks `browser_tool` is faster and more reliable — this is just an illustration of the macOS idiom.) diff --git a/skills/computer-use/windows/SKILL.md b/skills/computer-use/windows/SKILL.md new file mode 100644 index 0000000000000..b2e9a225d75da --- /dev/null +++ b/skills/computer-use/windows/SKILL.md @@ -0,0 +1,99 @@ +--- +name: computer-use-windows +description: Windows shortcuts, UAC handling, mss screenshot setup, and DPI-awareness gotchas for computer_use_windows. +metadata: + hermes: + tags: [windows, win32, sendinput, mss, uac, dpi, computer-use] +--- + +# Windows desktop control — what's different + +You have access to `computer_use_windows`. The action set and parameter shape are documented in the parent `computer-use` skill — load that first if you haven't. This skill covers Windows-only things you must know. + +## One-time host setup + +`computer_use_windows` works out of the box on a default Python install — `ctypes` is in the stdlib, and the screenshot path falls back to a built-in BitBlt routine when `mss` isn't installed. + +For best performance install `mss`: + +``` +pip install mss +``` + +`mss` is MIT-licensed, pure-Python over ctypes, and ~5-10× faster than the BitBlt fallback. Worth having if the agent will be running screenshots in a loop. + +## DPI awareness + +Modern Windows displays are usually scaled (125 %, 150 %, 200 %). The tool sets per-monitor v2 DPI awareness on import, so: + +- Click coordinates are in **physical screen pixels**, not scaled coordinates. +- Screenshots return the same physical pixel resolution. +- The numbers the screenshot shows are the numbers you click. No math. + +If you see clicks landing in the wrong place — typically off by exactly the DPI scale factor — the import-time DPI call failed (older Windows 7/8 hosts). On those hosts the fallback is to ask the user to reduce display scaling to 100 % for the session. + +## Modifier keys + +Windows uses **Ctrl** for the same shortcuts macOS uses **Cmd** for. The grammar parser accepts `Ctrl+T`, `ctrl+t`, `control+t` — all equivalent. The macOS `Cmd` token is reinterpreted as the **Win** (LWin) key on Windows, which is what the user wants in practice when porting muscle-memory. + +| Action | Combo | +|---|---| +| New tab / new window / save / close | `Ctrl+T` / `Ctrl+N` / `Ctrl+S` / `Ctrl+W` | +| Cut / copy / paste / undo | `Ctrl+X` / `Ctrl+C` / `Ctrl+V` / `Ctrl+Z` | +| Find | `Ctrl+F` | +| Switch window | `Alt+Tab` | +| Switch app (Win10/11) | `Win+Tab` | +| Start menu / search | `Win+S` (search), or just `Win` then type | +| Run dialog | `Win+R` | +| Lock workstation | `Win+L` | +| File Explorer | `Win+E` | +| Show desktop | `Win+D` | +| Window snap | `Win+Left` / `Win+Right` / `Win+Up` | +| Force-close window | `Alt+F4` | + +`Win+S` followed by typing is the analogue of macOS's `Cmd+Space` Spotlight pattern. + +## UAC (User Account Control) — the big one + +Windows enforces **UIPI** (User Interface Privilege Isolation): synthetic input from a **non-elevated** process cannot reach an **elevated** window. Specifically: + +- If a UAC prompt appears, your `key` and `click` actions hit dead air — the prompt window runs at higher integrity than the agent. +- If the user runs an admin-only program (Task Manager, Registry Editor, services.msc), same problem: the agent's clicks are silently dropped. + +Two ways to handle this: + +1. **Run Hermes elevated.** Right-click the Hermes launcher / terminal → "Run as administrator". The whole session then has integrity high, and all clicks land. This is the simplest fix when you know you'll need elevated control. +2. **Avoid elevated UI.** For most tasks (typing in apps, clicking buttons in regular programs, browser work) integrity medium is enough. UAC prompts that pop up unexpectedly should be deferred to the user. + +You can detect the UIPI failure case: a click "succeeds" (no error) but a follow-up screenshot shows nothing happened. If you see this pattern on a UAC dialog, surface to the user — don't keep retrying. + +## Active-window queries + +`get_active_window` returns `{id, title}` from `GetForegroundWindow` + `GetWindowTextW`. Reliable. The `id` is the HWND (window handle) as a decimal string, which can be passed to other Win32 calls if needed. + +## Screenshot quirks + +- Captures the **primary monitor only** by default. Multi-monitor capture is a known gap; we'd need to extend the tool to pass `monitor=N`. +- Includes the cursor by default (BitBlt path) — there's no easy way to hide it. +- Minimised windows are **not** captured (they have no client area to BitBlt). To screenshot a minimised window, restore it first with a click on the taskbar. + +## Don't try to do these + +- **UAC consent dialogs** — see UIPI above. +- **Lock screen / login screen** — different desktop session, no synthetic input access. +- **Game DirectInput** — the SendInput path injects to the standard message queue; many games (DirectX exclusive mode) read DirectInput directly and don't see synthetic events. `pydirectinput` is a Windows-game-specific library that handles this; we don't currently bundle it. +- **Driver / kernel-level UI** — same UIPI restriction as UAC. + +## Example — Win+R, run notepad, type, screenshot + +``` +{"action": "key", "keys": "Win+R"} +{"action": "wait", "ms": 300} +{"action": "type", "text": "notepad"} +{"action": "key", "keys": "Return"} +{"action": "wait", "ms": 800} +{"action": "type", "text": "hello from hermes"} +{"action": "screenshot"} +``` + +For most workflows just use `terminal` (PowerShell / cmd.exe) — it's faster than driving Notepad through the GUI. diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index 5b0359503489f..997c32f204e4e 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -678,4 +678,563 @@ def test_no_provider_gating_in_tool_registration(self): import inspect source = inspect.getsource(entry.check_fn) assert "anthropic" not in source.lower() - assert "openai" not in source.lower() + + +# --------------------------------------------------------------------------- +# Per-OS backend tests (native macOS / Linux / Windows implementations) +# --------------------------------------------------------------------------- +class CommonValidationTests(unittest.TestCase): + def test_missing_action_rejected(self): + with self.assertRaises(ValidationError): + parse_request({}) + + def test_unknown_action_rejected(self): + with self.assertRaises(ValidationError): + parse_request({"action": "smash_keyboard"}) + + def test_screenshot_no_args(self): + req = parse_request({"action": "screenshot"}) + self.assertEqual(req.action, "screenshot") + self.assertIsNone(req.region) + + def test_screenshot_region_validated(self): + req = parse_request({"action": "screenshot", "region": [0, 0, 100, 100]}) + self.assertEqual(req.region, [0, 0, 100, 100]) + with self.assertRaises(ValidationError): + parse_request({"action": "screenshot", "region": [0, 0, 100]}) + + def test_screenshot_redact_validated(self): + req = parse_request({ + "action": "screenshot", + "redact_regions": [[0, 0, 50, 50], [100, 100, 200, 200]], + }) + self.assertEqual(len(req.redact_regions), 2) + with self.assertRaises(ValidationError): + parse_request({ + "action": "screenshot", + "redact_regions": [[0, 0, 50]], + }) + + def test_left_click_requires_xy(self): + req = parse_request({"action": "left_click", "x": 10, "y": 20}) + self.assertEqual((req.x, req.y), (10, 20)) + with self.assertRaises(ValidationError): + parse_request({"action": "left_click", "x": 10}) + + def test_drag_requires_four_coords(self): + req = parse_request({ + "action": "mouse_drag", + "x": 1, "y": 2, "x2": 3, "y2": 4, + }) + self.assertEqual((req.x, req.y, req.x2, req.y2), (1, 2, 3, 4)) + with self.assertRaises(ValidationError): + parse_request({"action": "mouse_drag", "x": 1, "y": 2, "x2": 3}) + + def test_type_text_required_and_capped(self): + req = parse_request({"action": "type", "text": "hello"}) + self.assertEqual(req.text, "hello") + with self.assertRaises(ValidationError): + parse_request({"action": "type"}) + with self.assertRaises(ValidationError): + parse_request({"action": "type", "text": "x" * (MAX_TYPE_CHARS + 1)}) + + def test_key_keys_required_and_capped(self): + req = parse_request({"action": "key", "keys": "Cmd+Tab"}) + self.assertEqual(req.keys, "Cmd+Tab") + with self.assertRaises(ValidationError): + parse_request({"action": "key", "keys": ""}) + with self.assertRaises(ValidationError): + parse_request({"action": "key", "keys": "x" * (MAX_KEY_CHARS + 1)}) + + def test_scroll_direction_and_amount(self): + req = parse_request({"action": "scroll", "x": 0, "y": 0, "direction": "down"}) + self.assertEqual(req.direction, "down") + self.assertEqual(req.amount, 3) + with self.assertRaises(ValidationError): + parse_request({"action": "scroll", "x": 0, "y": 0, "direction": "diagonal"}) + with self.assertRaises(ValidationError): + parse_request({ + "action": "scroll", "x": 0, "y": 0, + "direction": "down", "amount": MAX_SCROLL_AMOUNT + 10, + }) + + def test_wait_bounds(self): + req = parse_request({"action": "wait", "ms": 100}) + self.assertEqual(req.ms, 100) + with self.assertRaises(ValidationError): + parse_request({"action": "wait", "ms": -1}) + with self.assertRaises(ValidationError): + parse_request({"action": "wait", "ms": MAX_WAIT_MS + 1}) + + def test_validate_coords_within(self): + req = ActionRequest(action="left_click", x=100, y=100) + validate_coords_within(req, 1920, 1080) + with self.assertRaises(ValidationError): + validate_coords_within(ActionRequest(action="left_click", x=-1, y=0), 1920, 1080) + with self.assertRaises(ValidationError): + validate_coords_within(ActionRequest(action="left_click", x=2000, y=0), 1920, 1080) + + def test_action_result_to_dict_omits_empty(self): + r = ActionResult(success=True, action="left_click") + d = r.to_dict() + self.assertEqual(d, {"success": True, "action": "left_click"}) + + def test_action_result_to_dict_includes_screenshot(self): + r = ActionResult(success=True, action="screenshot", screenshot_b64="abc") + d = r.to_dict() + self.assertIn("screenshot_b64", d) + self.assertEqual(d["screenshot_format"], "png") + + def test_build_schema_lists_all_actions(self): + schema = build_schema("computer_use_test", "Test OS") + enums = schema["parameters"]["properties"]["action"]["enum"] + self.assertEqual(set(enums), set(ACTIONS)) + + +# --------------------------------------------------------------------------- +# grammar +# --------------------------------------------------------------------------- + +class GrammarTests(unittest.TestCase): + def test_simple_letter(self): + p = parse_combo("a") + self.assertEqual(p.modifiers, set()) + self.assertEqual(p.key, "a") + + def test_modifier_aliases(self): + for combo in ("Cmd+T", "command+t", "meta+T", "Win+t"): + p = parse_combo(combo) + self.assertEqual(p.modifiers, {"cmd"}) + self.assertEqual(p.key, "t") + + def test_separator_dash(self): + p = parse_combo("ctrl-shift-A") + self.assertEqual(p.modifiers, {"ctrl", "shift"}) + self.assertEqual(p.key, "a") + + def test_function_keys(self): + p = parse_combo("F12") + self.assertEqual(p.key, "f12") + + def test_key_aliases(self): + for raw, expected in [("Esc", "escape"), ("Enter", "return"), ("Space", "space")]: + self.assertEqual(parse_combo(raw).key, expected) + + def test_unknown_modifier_rejected(self): + with self.assertRaises(KeyParseError): + parse_combo("hyper+t") + + def test_unknown_multichar_key_rejected(self): + with self.assertRaises(KeyParseError): + parse_combo("ctrl+notakey") + + def test_empty_rejected(self): + with self.assertRaises(KeyParseError): + parse_combo("") + with self.assertRaises(KeyParseError): + parse_combo(" ") + + def test_to_macos_cmd_t(self): + flags, code = to_macos(parse_combo("Cmd+T")) + self.assertEqual(flags, 0x00100000) + self.assertEqual(code, 0x11) + + def test_to_xdotool_cmd_to_super(self): + # macOS Cmd → Linux Super + self.assertEqual(to_xdotool(parse_combo("Cmd+Tab")), "super+Tab") + + def test_to_ydotool_codes(self): + codes = to_ydotool(parse_combo("ctrl+alt+t")) + self.assertIn(29, codes) # KEY_LEFTCTRL + self.assertIn(56, codes) # KEY_LEFTALT + self.assertIn(20, codes) # KEY_T + + def test_to_ydotool_f12(self): + codes = to_ydotool(parse_combo("F12")) + self.assertEqual(codes, [88]) + + def test_to_windows_letters(self): + codes = to_windows(parse_combo("ctrl+a")) + self.assertIn(0x11, codes) # VK_CONTROL + self.assertIn(0x41, codes) # VK_A + + def test_to_windows_f1_f24(self): + for i in range(1, 25): + codes = to_windows(parse_combo(f"F{i}")) + self.assertEqual(codes, [0x6F + i]) + + +# --------------------------------------------------------------------------- +# safety +# --------------------------------------------------------------------------- + +class SafetyTests(unittest.TestCase): + def setUp(self): + clear_kill_switch() + + def tearDown(self): + clear_kill_switch() + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + + def test_env_gate_default_off(self): + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + self.assertFalse(is_enabled()) + with self.assertRaises(SafetyRefusal): + gate("screenshot") + + def test_env_gate_truthy_values(self): + for v in ("true", "1", "yes", "on", "TRUE", "Yes"): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = v + self.assertTrue(is_enabled(), f"value {v!r} should enable") + + def test_env_gate_falsy_values(self): + for v in ("false", "0", "no", "off", "", "junk"): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = v + self.assertFalse(is_enabled(), f"value {v!r} should disable") + + def test_kill_switch(self): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = "true" + self.assertFalse(is_killed()) + gate("left_click") # works + set_kill_switch() + self.assertTrue(is_killed()) + with self.assertRaises(SafetyRefusal): + gate("left_click") + + def test_redact_image_blanks_region(self): + try: + from PIL import Image + import io + except ImportError: + self.skipTest("PIL not installed") + img = Image.new("RGB", (100, 100), (255, 0, 0)) + buf = io.BytesIO() + img.save(buf, format="PNG") + out = redact_image(buf.getvalue(), [[10, 10, 50, 50]]) + out_img = Image.open(io.BytesIO(out)).convert("RGB") + # Pixel inside the redacted region should be black. + self.assertEqual(out_img.getpixel((30, 30)), (0, 0, 0)) + # Pixel outside should still be red. + self.assertEqual(out_img.getpixel((80, 80)), (255, 0, 0)) + + def test_redact_image_no_regions_passthrough(self): + try: + from PIL import Image + import io + except ImportError: + self.skipTest("PIL not installed") + img = Image.new("RGB", (10, 10), (0, 255, 0)) + buf = io.BytesIO() + img.save(buf, format="PNG") + original = buf.getvalue() + self.assertEqual(redact_image(original, []), original) + + def test_log_action_writes_jsonl(self): + import tempfile + with tempfile.TemporaryDirectory() as tmpdir: + os.environ["HERMES_HOME"] = tmpdir + log_action("left_click", {"x": 10, "y": 20}, True) + log_action("type", {"text": "hi"}, False, error="oops") + log_path = Path(tmpdir) / "logs" / "computer_use.jsonl" + self.assertTrue(log_path.exists()) + lines = log_path.read_text().splitlines() + self.assertEqual(len(lines), 2) + r1 = json.loads(lines[0]) + self.assertEqual(r1["action"], "left_click") + self.assertTrue(r1["success"]) + r2 = json.loads(lines[1]) + self.assertEqual(r2["error"], "oops") + + +# --------------------------------------------------------------------------- +# macOS backend (mocked Quartz) +# --------------------------------------------------------------------------- + +class MacOSBackendTests(unittest.TestCase): + def setUp(self): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = "true" + clear_kill_switch() + + def tearDown(self): + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + + def _stub_quartz(self): + """Build a MagicMock that quacks like enough of pyobjc Quartz.""" + Q = MagicMock() + # Mouse event flag constants + for name in ( + "kCGEventLeftMouseDown", "kCGEventLeftMouseUp", "kCGEventLeftMouseDragged", + "kCGEventRightMouseDown", "kCGEventRightMouseUp", + "kCGEventOtherMouseDown", "kCGEventOtherMouseUp", + "kCGEventMouseMoved", "kCGHIDEventTap", "kCGScrollEventUnitLine", + "kCGMouseButtonLeft", "kCGMouseButtonRight", "kCGMouseButtonCenter", + "kCGWindowListOptionOnScreenOnly", "kCGWindowListExcludeDesktopElements", + "kCGNullWindowID", + ): + setattr(Q, name, 0) + Q.CGMainDisplayID.return_value = 1 + Q.CGDisplayPixelsWide.return_value = 1920 + Q.CGDisplayPixelsHigh.return_value = 1080 + # Cursor location + loc = MagicMock(); loc.x = 100; loc.y = 200 + Q.CGEventGetLocation.return_value = loc + Q.CGWindowListCopyWindowInfo.return_value = [ + {"kCGWindowOwnerName": "Safari", "kCGWindowName": "Apple"}, + ] + return Q + + @patch("tools.computer_use_macos._screenshot_full") + @patch("tools.computer_use_macos._load_quartz") + def test_screen_size(self, mock_load, mock_shot): + mock_load.return_value = self._stub_quartz() + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "screen_size"}) + self.assertTrue(out["success"]) + self.assertEqual(out["screen"], {"width": 1920, "height": 1080}) + + @patch("tools.computer_use_macos._load_quartz") + def test_cursor_position(self, mock_load): + mock_load.return_value = self._stub_quartz() + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "cursor_position"}) + self.assertEqual(out["cursor"], {"x": 100, "y": 200}) + + @patch("tools.computer_use_macos._load_quartz") + def test_active_window(self, mock_load): + mock_load.return_value = self._stub_quartz() + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "get_active_window"}) + self.assertEqual(out["active_window"]["app"], "Safari") + + @patch("tools.computer_use_macos._screenshot_full") + @patch("tools.computer_use_macos._load_quartz") + def test_screenshot_returns_b64(self, mock_load, mock_shot): + mock_load.return_value = self._stub_quartz() + mock_shot.return_value = b"\x89PNG\r\n\x1a\n" + b"x" * 50 + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "screenshot"}) + self.assertTrue(out["success"]) + self.assertEqual(base64.b64decode(out["screenshot_b64"])[:8], b"\x89PNG\r\n\x1a\n") + + @patch("tools.computer_use_macos._load_quartz") + def test_left_click_posts_two_events(self, mock_load): + Q = self._stub_quartz() + mock_load.return_value = Q + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "left_click", "x": 100, "y": 100}) + self.assertTrue(out["success"], out) + # CGEventPost called for both mouse-down and mouse-up + self.assertGreaterEqual(Q.CGEventPost.call_count, 2) + + @patch("tools.computer_use_macos._load_quartz") + def test_key_combo_uses_cgflags(self, mock_load): + Q = self._stub_quartz() + mock_load.return_value = Q + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "key", "keys": "Cmd+T"}) + self.assertTrue(out["success"], out) + # CGEventCreateKeyboardEvent called twice (down + up) + self.assertEqual(Q.CGEventCreateKeyboardEvent.call_count, 2) + Q.CGEventSetFlags.assert_called() + + @patch("tools.computer_use_macos._load_quartz") + def test_disabled_env_refuses(self, mock_load): + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + mock_load.return_value = self._stub_quartz() + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "left_click", "x": 100, "y": 100}) + self.assertFalse(out["success"]) + self.assertIn("refused", out["error"]) + + @patch("tools.computer_use_macos._load_quartz") + def test_off_screen_click_rejected(self, mock_load): + mock_load.return_value = self._stub_quartz() + from tools.computer_use_macos import computer_use_macos_tool + out = computer_use_macos_tool({"action": "left_click", "x": 5000, "y": 5000}) + self.assertFalse(out["success"]) + self.assertIn("validation", out["error"]) + + +# --------------------------------------------------------------------------- +# Linux backend (mocked subprocess) +# --------------------------------------------------------------------------- + +class LinuxBackendTests(unittest.TestCase): + def setUp(self): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = "true" + clear_kill_switch() + + def tearDown(self): + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + + def _x11_env(self): + return {"WAYLAND_DISPLAY": "", "XDG_SESSION_TYPE": "x11"} + + def _wayland_env(self): + return {"WAYLAND_DISPLAY": "wayland-0", "XDG_SESSION_TYPE": "wayland"} + + @patch.dict(os.environ, {"WAYLAND_DISPLAY": "", "XDG_SESSION_TYPE": "x11"}) + @patch("tools.computer_use_linux.shutil.which") + @patch("tools.computer_use_linux._run") + def test_x11_screen_size_from_xdpyinfo(self, mock_run, mock_which): + mock_which.side_effect = lambda c: f"/usr/bin/{c}" if c in ("xdotool", "scrot", "xdpyinfo") else None + mock_run.return_value = MagicMock(stdout="dimensions: 1920x1080 pixels\n") + from tools.computer_use_linux import computer_use_linux_tool + out = computer_use_linux_tool({"action": "screen_size"}) + self.assertTrue(out["success"], out) + self.assertEqual(out["screen"], {"width": 1920, "height": 1080}) + + @patch.dict(os.environ, {"WAYLAND_DISPLAY": "", "XDG_SESSION_TYPE": "x11"}) + @patch("tools.computer_use_linux.shutil.which") + @patch("tools.computer_use_linux._run") + def test_x11_left_click_invokes_xdotool(self, mock_run, mock_which): + mock_which.side_effect = lambda c: f"/usr/bin/{c}" if c in ("xdotool", "scrot", "xdpyinfo") else None + mock_run.return_value = MagicMock(stdout="dimensions: 1920x1080 pixels\n") + from tools.computer_use_linux import computer_use_linux_tool + out = computer_use_linux_tool({"action": "left_click", "x": 100, "y": 200}) + self.assertTrue(out["success"], out) + # _run called for screen_size probe + click invocation + click_calls = [c for c in mock_run.call_args_list if c.args and "xdotool" in c.args[0][0]] + self.assertTrue(any("mousemove" in str(c) and "click" in str(c) for c in click_calls)) + + @patch.dict(os.environ, {"WAYLAND_DISPLAY": "", "XDG_SESSION_TYPE": "x11"}) + @patch("tools.computer_use_linux.shutil.which") + @patch("tools.computer_use_linux._run") + def test_x11_screenshot_reads_scrot_output(self, mock_run, mock_which): + mock_which.side_effect = lambda c: f"/usr/bin/{c}" if c in ("xdotool", "scrot", "xdpyinfo") else None + mock_run.return_value = MagicMock(stdout="") + with patch("tools.computer_use_linux.Path") as mock_path: + mock_path.return_value.read_bytes.return_value = b"\x89PNG\r\n\x1a\nfake" + mock_path.return_value.unlink = lambda: None + from tools.computer_use_linux import computer_use_linux_tool + out = computer_use_linux_tool({"action": "screenshot"}) + self.assertTrue(out["success"], out) + self.assertEqual(base64.b64decode(out["screenshot_b64"])[:8], b"\x89PNG\r\n\x1a\n") + + @patch.dict(os.environ, {"WAYLAND_DISPLAY": "", "XDG_SESSION_TYPE": "x11"}) + @patch("tools.computer_use_linux.shutil.which") + @patch("tools.computer_use_linux._run") + def test_x11_key_combo_lowercases_modifier_string(self, mock_run, mock_which): + mock_which.side_effect = lambda c: f"/usr/bin/{c}" if c in ("xdotool", "scrot", "xdpyinfo") else None + mock_run.return_value = MagicMock(stdout="dimensions: 1920x1080 pixels\n") + from tools.computer_use_linux import computer_use_linux_tool + out = computer_use_linux_tool({"action": "key", "keys": "Ctrl+Alt+T"}) + self.assertTrue(out["success"], out) + # the xdotool key argument should be 'ctrl+alt+t' + key_calls = [c for c in mock_run.call_args_list if c.args and "key" in c.args[0]] + self.assertTrue(any("ctrl+alt+t" in str(c) for c in key_calls), key_calls) + + @patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0", "XDG_SESSION_TYPE": "wayland"}) + @patch("tools.computer_use_linux.shutil.which") + @patch("tools.computer_use_linux._run") + def test_wayland_uses_grim_and_ydotool(self, mock_run, mock_which): + mock_which.side_effect = lambda c: f"/usr/bin/{c}" if c in ("ydotool", "grim", "wlr-randr") else None + mock_run.return_value = MagicMock(stdout="HDMI-A-1 \"Mock\"\n current 1920x1080@60Hz\n") + from tools.computer_use_linux import computer_use_linux_tool + out = computer_use_linux_tool({"action": "screen_size"}) + self.assertTrue(out["success"], out) + # Wayland branch was selected; grim should also be runnable for screenshot + with patch("tools.computer_use_linux.Path") as mock_path: + mock_path.return_value.read_bytes.return_value = b"\x89PNG\r\n\x1a\nfake" + mock_path.return_value.unlink = lambda: None + out2 = computer_use_linux_tool({"action": "screenshot"}) + self.assertTrue(out2["success"]) + grim_calls = [c for c in mock_run.call_args_list if c.args and "grim" in c.args[0]] + self.assertTrue(grim_calls, "grim should have been invoked") + + +# --------------------------------------------------------------------------- +# Windows backend (mocked user32) +# --------------------------------------------------------------------------- + +class WindowsBackendTests(unittest.TestCase): + def setUp(self): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = "true" + clear_kill_switch() + + def tearDown(self): + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + + def _stub_user32(self): + u = MagicMock() + u.GetSystemMetrics.side_effect = lambda code: 1920 if code == 0 else 1080 + u.SendInput.side_effect = lambda n, arr, sz: n + u.GetForegroundWindow.return_value = 0xABCD + u.GetWindowTextLengthW.return_value = 5 + return u + + @patch("tools.computer_use_windows._load_user32") + def test_screen_size(self, mock_load): + mock_load.return_value = self._stub_user32() + from tools.computer_use_windows import computer_use_windows_tool + out = computer_use_windows_tool({"action": "screen_size"}) + self.assertTrue(out["success"]) + self.assertEqual(out["screen"], {"width": 1920, "height": 1080}) + + @patch("tools.computer_use_windows._load_user32") + def test_left_click_calls_sendinput(self, mock_load): + u = self._stub_user32() + mock_load.return_value = u + from tools.computer_use_windows import computer_use_windows_tool + out = computer_use_windows_tool({"action": "left_click", "x": 100, "y": 200}) + self.assertTrue(out["success"], out) + # _click() makes 2 SendInput calls: one for move, one with [down, up]. + self.assertGreaterEqual(u.SendInput.call_count, 2) + + @patch("tools.computer_use_windows._load_user32") + def test_key_combo_calls_sendinput(self, mock_load): + u = self._stub_user32() + mock_load.return_value = u + from tools.computer_use_windows import computer_use_windows_tool + out = computer_use_windows_tool({"action": "key", "keys": "Ctrl+T"}) + self.assertTrue(out["success"], out) + u.SendInput.assert_called() + + @patch("tools.computer_use_windows._load_user32") + def test_off_screen_click_rejected(self, mock_load): + mock_load.return_value = self._stub_user32() + from tools.computer_use_windows import computer_use_windows_tool + out = computer_use_windows_tool({"action": "left_click", "x": 5000, "y": 5000}) + self.assertFalse(out["success"]) + self.assertIn("validation", out["error"]) + + +# --------------------------------------------------------------------------- +# Registry integration — registration + check_fn gating +# --------------------------------------------------------------------------- + +class RegistryIntegrationTests(unittest.TestCase): + def test_all_three_register(self): + # Module imports already happened at file top + from tools.registry import registry + names = registry.get_all_tool_names() + for n in ("computer_use_macos", "computer_use_linux", "computer_use_windows"): + self.assertIn(n, names, f"{n} not in registry") + + def test_check_fn_off_when_env_unset(self): + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + from tools.registry import registry, invalidate_check_fn_cache + invalidate_check_fn_cache() + for n in ("computer_use_macos", "computer_use_linux", "computer_use_windows"): + self.assertFalse(registry.get_entry(n).check_fn(), f"{n} check_fn should be False") + + def test_only_host_os_passes_with_env(self): + os.environ["HERMES_COMPUTER_USE_ENABLED"] = "true" + from tools.registry import registry, invalidate_check_fn_cache + invalidate_check_fn_cache() + host = sys.platform + try: + for tool, expected_platform in [ + ("computer_use_macos", "darwin"), + ("computer_use_linux", "linux"), + ("computer_use_windows", "win32"), + ]: + check = registry.get_entry(tool).check_fn() + if host != expected_platform: + self.assertFalse(check, f"{tool} should be False on host {host}") + finally: + os.environ.pop("HERMES_COMPUTER_USE_ENABLED", None) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/computer_use_common.py b/tools/computer_use_common.py new file mode 100644 index 0000000000000..f1441285919d3 --- /dev/null +++ b/tools/computer_use_common.py @@ -0,0 +1,289 @@ +"""Shared types, schema, and validation primitives for the computer_use_* tools. + +The per-OS backends (computer_use_macos.py, computer_use_linux.py, +computer_use_windows.py) each register a tool against the same JSON schema +defined here, so a model trained on one platform's tool surface generalises +to the others. Per-OS skills handle the platform-specific shortcut idioms +(Cmd vs Ctrl, X11 vs Wayland, etc.). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + + +# --------------------------------------------------------------------------- +# Action set — mirrors Anthropic's computer_20251124 schema with a few +# practical additions (get_active_window, screen_size, find_text). +# --------------------------------------------------------------------------- + +ACTIONS: Tuple[str, ...] = ( + "screenshot", + "left_click", + "double_click", + "right_click", + "middle_click", + "left_button_press", + "left_button_release", + "mouse_move", + "mouse_drag", + "type", + "key", + "scroll", + "wait", + "cursor_position", + "screen_size", + "get_active_window", +) + +# Hard caps to prevent the model from clobbering the host with one bad call. +MAX_TYPE_CHARS = 10_000 +MAX_KEY_CHARS = 256 +MAX_WAIT_MS = 30_000 +MAX_SCROLL_AMOUNT = 50 + + +@dataclass +class ActionRequest: + """Validated request for a single computer-use action.""" + + action: str + x: Optional[int] = None + y: Optional[int] = None + x2: Optional[int] = None + y2: Optional[int] = None + text: Optional[str] = None + keys: Optional[str] = None + direction: Optional[str] = None + amount: Optional[int] = None + ms: Optional[int] = None + region: Optional[List[int]] = None + redact_regions: Optional[List[List[int]]] = None + + +@dataclass +class ActionResult: + """Result envelope returned to the model.""" + + success: bool + action: str + message: str = "" + screenshot_b64: Optional[str] = None + screenshot_format: str = "png" + cursor_x: Optional[int] = None + cursor_y: Optional[int] = None + screen_width: Optional[int] = None + screen_height: Optional[int] = None + active_window: Optional[Dict[str, Any]] = None + error: Optional[str] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + out: Dict[str, Any] = { + "success": self.success, + "action": self.action, + } + if self.message: + out["message"] = self.message + if self.screenshot_b64 is not None: + out["screenshot_b64"] = self.screenshot_b64 + out["screenshot_format"] = self.screenshot_format + if self.cursor_x is not None and self.cursor_y is not None: + out["cursor"] = {"x": self.cursor_x, "y": self.cursor_y} + if self.screen_width is not None and self.screen_height is not None: + out["screen"] = {"width": self.screen_width, "height": self.screen_height} + if self.active_window is not None: + out["active_window"] = self.active_window + if self.error: + out["error"] = self.error + if self.metadata: + out["metadata"] = self.metadata + return out + + +class ValidationError(Exception): + """Raised when an ActionRequest fails parameter validation.""" + + +def parse_request(args: Dict[str, Any]) -> ActionRequest: + """Parse and validate a tool-call arg dict into an ActionRequest. + + Raises ValidationError with a model-readable explanation when the input + is malformed. The handler should catch and turn it into an ActionResult + with success=False so the model can self-correct. + """ + if not isinstance(args, dict): + raise ValidationError("arguments must be a JSON object") + + action = args.get("action") + if action not in ACTIONS: + raise ValidationError( + f"action must be one of {ACTIONS}, got {action!r}" + ) + + req = ActionRequest(action=action) + + # Coordinate-bearing actions + coord_actions = { + "left_click", "double_click", "right_click", "middle_click", + "left_button_press", "left_button_release", + "mouse_move", "scroll", + } + if action in coord_actions: + req.x = _coerce_int(args, "x", required=True) + req.y = _coerce_int(args, "y", required=True) + + if action == "mouse_drag": + req.x = _coerce_int(args, "x", required=True) + req.y = _coerce_int(args, "y", required=True) + req.x2 = _coerce_int(args, "x2", required=True) + req.y2 = _coerce_int(args, "y2", required=True) + + if action == "type": + text = args.get("text") + if not isinstance(text, str) or not text: + raise ValidationError("'type' requires non-empty 'text' string") + if len(text) > MAX_TYPE_CHARS: + raise ValidationError(f"'text' exceeds {MAX_TYPE_CHARS}-char cap") + req.text = text + + if action == "key": + keys = args.get("keys") + if not isinstance(keys, str) or not keys: + raise ValidationError("'key' requires non-empty 'keys' string") + if len(keys) > MAX_KEY_CHARS: + raise ValidationError(f"'keys' exceeds {MAX_KEY_CHARS}-char cap") + req.keys = keys + + if action == "scroll": + direction = args.get("direction") + if direction not in {"up", "down", "left", "right"}: + raise ValidationError("'scroll' requires direction in {up,down,left,right}") + req.direction = direction + req.amount = _coerce_int(args, "amount", required=False, default=3) + if req.amount is not None and (req.amount < 1 or req.amount > MAX_SCROLL_AMOUNT): + raise ValidationError(f"'amount' must be in [1, {MAX_SCROLL_AMOUNT}]") + + if action == "wait": + req.ms = _coerce_int(args, "ms", required=True) + if req.ms is None or req.ms < 0 or req.ms > MAX_WAIT_MS: + raise ValidationError(f"'ms' must be in [0, {MAX_WAIT_MS}]") + + if action == "screenshot": + region = args.get("region") + if region is not None: + if ( + not isinstance(region, list) + or len(region) != 4 + or not all(isinstance(v, (int, float)) for v in region) + ): + raise ValidationError("'region' must be [x1,y1,x2,y2] integers") + req.region = [int(v) for v in region] + redact = args.get("redact_regions") + if redact is not None: + if not isinstance(redact, list): + raise ValidationError("'redact_regions' must be a list of [x1,y1,x2,y2]") + for r in redact: + if not isinstance(r, list) or len(r) != 4: + raise ValidationError("each redact region must be [x1,y1,x2,y2]") + req.redact_regions = [[int(v) for v in r] for r in redact] + + return req + + +def _coerce_int(args: Dict[str, Any], key: str, *, required: bool, default: Optional[int] = None) -> Optional[int]: + val = args.get(key) + if val is None: + if required: + raise ValidationError(f"missing required integer field {key!r}") + return default + if isinstance(val, bool) or not isinstance(val, (int, float)): + raise ValidationError(f"{key!r} must be a number") + return int(val) + + +def validate_coords_within(req: ActionRequest, screen_w: int, screen_h: int) -> None: + """Reject obviously-invalid coordinates before they reach the OS layer. + + Off-screen clicks waste an action and can cause unexpected focus changes + on multi-monitor setups. Better to fail early with a clear message. + """ + pairs: List[Tuple[Optional[int], Optional[int], str]] = [ + (req.x, req.y, "(x,y)"), + (req.x2, req.y2, "(x2,y2)"), + ] + for x, y, label in pairs: + if x is None or y is None: + continue + if x < 0 or y < 0 or x > screen_w or y > screen_h: + raise ValidationError( + f"{label}=({x},{y}) outside screen bounds {screen_w}x{screen_h}" + ) + + +# --------------------------------------------------------------------------- +# Shared JSON schema. Each per-OS tool registers under a distinct ``name`` +# (computer_use_macos / _linux / _windows) so the model can route by check_fn +# but presents an identical parameter surface. +# --------------------------------------------------------------------------- + +def build_schema(tool_name: str, platform_label: str) -> Dict[str, Any]: + """Build the JSON schema for an OS-specific computer_use tool variant.""" + return { + "name": tool_name, + "description": ( + f"Native desktop control on {platform_label}. Take screenshots, click, " + f"type, send key combinations, scroll, drag. Always start a task with " + f"a screenshot to ground subsequent actions in pixel coordinates. " + f"Coordinates are absolute screen pixels (origin top-left). " + f"For key combos use the per-OS skill grammar (Cmd+Tab on macOS, " + f"ctrl+alt+t on Linux, win+s on Windows)." + ), + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": list(ACTIONS), + "description": "Which desktop action to perform.", + }, + "x": {"type": "integer", "description": "Absolute x pixel."}, + "y": {"type": "integer", "description": "Absolute y pixel."}, + "x2": {"type": "integer", "description": "Drag end x (mouse_drag only)."}, + "y2": {"type": "integer", "description": "Drag end y (mouse_drag only)."}, + "text": { + "type": "string", + "description": f"Text to type (max {MAX_TYPE_CHARS} chars).", + }, + "keys": { + "type": "string", + "description": "Key combo string e.g. 'Cmd+Tab', 'ctrl+shift+T'.", + }, + "direction": { + "type": "string", + "enum": ["up", "down", "left", "right"], + "description": "Scroll direction.", + }, + "amount": { + "type": "integer", + "description": f"Scroll wheel ticks (1-{MAX_SCROLL_AMOUNT}, default 3).", + }, + "ms": { + "type": "integer", + "description": f"Wait duration in ms (0-{MAX_WAIT_MS}).", + }, + "region": { + "type": "array", + "items": {"type": "integer"}, + "description": "Screenshot crop [x1,y1,x2,y2]. Omit for full screen.", + }, + "redact_regions": { + "type": "array", + "items": {"type": "array", "items": {"type": "integer"}}, + "description": "Rectangles to blank in returned screenshot — for hiding password fields, MFA codes, etc. before the image reaches the model.", + }, + }, + "required": ["action"], + }, + } diff --git a/tools/computer_use_grammar.py b/tools/computer_use_grammar.py new file mode 100644 index 0000000000000..af4ae15220969 --- /dev/null +++ b/tools/computer_use_grammar.py @@ -0,0 +1,322 @@ +"""Per-OS key-combo string parser. + +Models reach for a familiar grammar when typing keys: ``Cmd+Tab`` on macOS, +``ctrl+alt+t`` on Linux, ``win+s`` on Windows. We parse the same +``mod+mod+key`` string into per-OS native primitives so each backend can +dispatch with the right call: + +* macOS → list of ``(modifier_flag, keycode)`` for ``CGEventCreateKeyboardEvent`` +* Linux X11 → xdotool key string (``"ctrl+shift+t"``, lowercase canonical) +* Linux Wayland → ydotool key code sequence (Linux input event codes) +* Windows → list of Win32 virtual-key codes for ``SendInput`` + +The grammar is forgiving: case-insensitive modifier names, both ``+`` and +``-`` accepted as separators, and the special token ``Return``/``Enter`` +maps to the same key on each platform. Unknown keys raise ``KeyParseError`` +with a helpful suggestion list. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Dict, List, Set, Tuple + +KeyParseError = ValueError + + +# --------------------------------------------------------------------------- +# Canonical modifier names — all lowercase. Aliases collapse to the same +# canonical token so "Cmd"/"command"/"meta" all mean the same modifier. +# --------------------------------------------------------------------------- + +_MODIFIER_ALIASES: Dict[str, str] = { + "ctrl": "ctrl", "control": "ctrl", + "shift": "shift", + "alt": "alt", "option": "alt", "opt": "alt", + "cmd": "cmd", "command": "cmd", "meta": "cmd", + "win": "cmd", "windows": "cmd", "super": "cmd", + "fn": "fn", +} + +CANONICAL_MODIFIERS: Set[str] = {"ctrl", "shift", "alt", "cmd", "fn"} + + +# --------------------------------------------------------------------------- +# Canonical key names — lowercase. Cover the keys models actually use: +# letters, digits, function keys, navigation, editing, and a few common +# whitespace/symbol names. +# --------------------------------------------------------------------------- + +_KEY_ALIASES: Dict[str, str] = { + "return": "return", "enter": "return", + "esc": "escape", "escape": "escape", + "tab": "tab", + "space": "space", "spacebar": "space", + "backspace": "backspace", "bksp": "backspace", + "delete": "delete", "del": "delete", + "up": "up", "down": "down", "left": "left", "right": "right", + "home": "home", "end": "end", + "pageup": "pageup", "pgup": "pageup", + "pagedown": "pagedown", "pgdn": "pagedown", + "insert": "insert", "ins": "insert", + "capslock": "capslock", "caps": "capslock", + "printscreen": "printscreen", "prtsc": "printscreen", + "minus": "minus", "-": "minus", + "equals": "equals", "=": "equals", + "comma": "comma", ",": "comma", + "period": "period", ".": "period", + "slash": "slash", "/": "slash", + "backslash": "backslash", "\\": "backslash", + "semicolon": "semicolon", ";": "semicolon", + "quote": "quote", "'": "quote", + "leftbracket": "leftbracket", "[": "leftbracket", + "rightbracket": "rightbracket", "]": "rightbracket", + "backtick": "backtick", "`": "backtick", +} + +# Function keys F1-F24 +for _i in range(1, 25): + _KEY_ALIASES[f"f{_i}"] = f"f{_i}" + +# Single letters and digits — the canonical form is the lowercase character. +for _c in "abcdefghijklmnopqrstuvwxyz0123456789": + _KEY_ALIASES[_c] = _c + + +@dataclass +class ParsedKey: + """Canonical form of a parsed key combo. + + Attributes + ---------- + modifiers + Subset of {"ctrl", "shift", "alt", "cmd", "fn"} — order is irrelevant. + key + Canonical lowercase key name (e.g. "t", "tab", "f5", "return"). + raw + Original input string, kept for error messages and logging. + """ + + modifiers: Set[str] = field(default_factory=set) + key: str = "" + raw: str = "" + + +def parse_combo(combo: str) -> ParsedKey: + """Parse ``"Cmd+Shift+T"`` (or ``"cmd-shift-t"``) into a ParsedKey. + + Raises KeyParseError on unknown tokens with the offending part included. + """ + if not isinstance(combo, str) or not combo.strip(): + raise KeyParseError("empty key combo") + + parts = re.split(r"[+\-]", combo.strip()) + parts = [p.strip() for p in parts if p.strip()] + if not parts: + raise KeyParseError(f"could not parse key combo {combo!r}") + + parsed = ParsedKey(raw=combo) + # Last token is the key; everything before is modifiers. + *mods, last = parts + for m in mods: + canon = _MODIFIER_ALIASES.get(m.lower()) + if canon is None: + raise KeyParseError( + f"unknown modifier {m!r} in {combo!r}; " + f"valid: {sorted(CANONICAL_MODIFIERS)}" + ) + parsed.modifiers.add(canon) + + canon_key = _KEY_ALIASES.get(last.lower()) + if canon_key is None: + if len(last) == 1: + # Allow any single printable character; backends type it literally. + canon_key = last.lower() + else: + raise KeyParseError( + f"unknown key {last!r} in {combo!r}; " + f"recognised tokens include letters, digits, function keys, " + f"and navigation/editing keys (return, tab, escape, etc.)" + ) + parsed.key = canon_key + return parsed + + +# --------------------------------------------------------------------------- +# macOS — Quartz keycodes + modifier flags. +# +# Keycodes come from /System/Library/Frameworks/Carbon.framework +# (kVK_ANSI_*). We hardcode the mapping rather than depend on Carbon to +# keep this module importable on non-macOS hosts. +# --------------------------------------------------------------------------- + +# kCGEventFlagMask* values from Quartz/CGEventTypes.h +MAC_FLAG = { + "shift": 0x00020000, + "ctrl": 0x00040000, + "alt": 0x00080000, + "cmd": 0x00100000, + "fn": 0x00800000, +} + +MAC_KEYCODE: Dict[str, int] = { + "a": 0x00, "s": 0x01, "d": 0x02, "f": 0x03, "h": 0x04, "g": 0x05, + "z": 0x06, "x": 0x07, "c": 0x08, "v": 0x09, "b": 0x0B, "q": 0x0C, + "w": 0x0D, "e": 0x0E, "r": 0x0F, "y": 0x10, "t": 0x11, + "1": 0x12, "2": 0x13, "3": 0x14, "4": 0x15, "6": 0x16, "5": 0x17, + "equals": 0x18, "9": 0x19, "7": 0x1A, "minus": 0x1B, "8": 0x1C, "0": 0x1D, + "rightbracket": 0x1E, "o": 0x1F, "u": 0x20, "leftbracket": 0x21, + "i": 0x22, "p": 0x23, "l": 0x25, "j": 0x26, "quote": 0x27, "k": 0x28, + "semicolon": 0x29, "backslash": 0x2A, "comma": 0x2B, "slash": 0x2C, + "n": 0x2D, "m": 0x2E, "period": 0x2F, "backtick": 0x32, + "return": 0x24, "tab": 0x30, "space": 0x31, "backspace": 0x33, + "escape": 0x35, "capslock": 0x39, + "left": 0x7B, "right": 0x7C, "down": 0x7D, "up": 0x7E, + "home": 0x73, "end": 0x77, "pageup": 0x74, "pagedown": 0x79, + "delete": 0x75, + "f1": 0x7A, "f2": 0x78, "f3": 0x63, "f4": 0x76, "f5": 0x60, "f6": 0x61, + "f7": 0x62, "f8": 0x64, "f9": 0x65, "f10": 0x6D, "f11": 0x67, "f12": 0x6F, + "f13": 0x69, "f14": 0x6B, "f15": 0x71, "f16": 0x6A, "f17": 0x40, "f18": 0x4F, + "f19": 0x50, "f20": 0x5A, +} + + +def to_macos(parsed: ParsedKey) -> Tuple[int, int]: + """Return ``(flags, keycode)`` for ``CGEventCreateKeyboardEvent``.""" + flags = 0 + for m in parsed.modifiers: + flags |= MAC_FLAG.get(m, 0) + keycode = MAC_KEYCODE.get(parsed.key) + if keycode is None: + raise KeyParseError(f"no macOS keycode for {parsed.key!r}") + return flags, keycode + + +# --------------------------------------------------------------------------- +# Linux X11 — xdotool string. Modifiers and key names map directly. +# --------------------------------------------------------------------------- + +XDOTOOL_MOD: Dict[str, str] = { + "ctrl": "ctrl", + "shift": "shift", + "alt": "alt", + "cmd": "super", # macOS Cmd → Linux Super (Win) key + "fn": "", # Linux has no separate Fn modifier; drop silently +} + +XDOTOOL_KEY: Dict[str, str] = { + "return": "Return", "tab": "Tab", "escape": "Escape", "space": "space", + "backspace": "BackSpace", "delete": "Delete", + "up": "Up", "down": "Down", "left": "Left", "right": "Right", + "home": "Home", "end": "End", "pageup": "Prior", "pagedown": "Next", + "insert": "Insert", "capslock": "Caps_Lock", "printscreen": "Print", + "minus": "minus", "equals": "equal", "comma": "comma", "period": "period", + "slash": "slash", "backslash": "backslash", "semicolon": "semicolon", + "quote": "apostrophe", "leftbracket": "bracketleft", + "rightbracket": "bracketright", "backtick": "grave", +} +for _i in range(1, 25): + XDOTOOL_KEY[f"f{_i}"] = f"F{_i}" + + +def to_xdotool(parsed: ParsedKey) -> str: + """Return an xdotool ``key`` argument like ``ctrl+shift+t``.""" + parts: List[str] = [] + for m in ("ctrl", "alt", "shift", "cmd"): # canonical order + if m in parsed.modifiers: + x = XDOTOOL_MOD.get(m, "") + if x: + parts.append(x) + key = XDOTOOL_KEY.get(parsed.key, parsed.key) + parts.append(key) + return "+".join(parts) + + +# --------------------------------------------------------------------------- +# Linux Wayland (ydotool) — Linux input event codes from +# /usr/include/linux/input-event-codes.h. ydotool's ``key`` syntax accepts +# decimal codes or symbolic names; we use the codes for portability. +# --------------------------------------------------------------------------- + +YDOTOOL_MOD: Dict[str, int] = { + "ctrl": 29, # KEY_LEFTCTRL + "shift": 42, # KEY_LEFTSHIFT + "alt": 56, # KEY_LEFTALT + "cmd": 125, # KEY_LEFTMETA +} + +YDOTOOL_KEY: Dict[str, int] = { + "a": 30, "b": 48, "c": 46, "d": 32, "e": 18, "f": 33, "g": 34, "h": 35, + "i": 23, "j": 36, "k": 37, "l": 38, "m": 50, "n": 49, "o": 24, "p": 25, + "q": 16, "r": 19, "s": 31, "t": 20, "u": 22, "v": 47, "w": 17, "x": 45, + "y": 21, "z": 44, + "1": 2, "2": 3, "3": 4, "4": 5, "5": 6, + "6": 7, "7": 8, "8": 9, "9": 10, "0": 11, + "return": 28, "tab": 15, "space": 57, "escape": 1, "backspace": 14, + "delete": 111, "insert": 110, "capslock": 58, + "up": 103, "down": 108, "left": 105, "right": 106, + "home": 102, "end": 107, "pageup": 104, "pagedown": 109, + "minus": 12, "equals": 13, "leftbracket": 26, "rightbracket": 27, + "backslash": 43, "semicolon": 39, "quote": 40, "backtick": 41, + "comma": 51, "period": 52, "slash": 53, +} +# Linux F1-F10 are KEY_F1=59 through KEY_F10=68; F11/F12 jump to 87/88 +# (per /usr/include/linux/input-event-codes.h). F13+ also non-contiguous; +# we cap at F12 since that covers every standard keyboard. +for _i in range(1, 11): + YDOTOOL_KEY[f"f{_i}"] = 58 + _i +YDOTOOL_KEY["f11"] = 87 +YDOTOOL_KEY["f12"] = 88 + + +def to_ydotool(parsed: ParsedKey) -> List[int]: + """Return [keycode, ...] press order for ydotool. Caller emits :1 for press, :0 for release.""" + codes = [YDOTOOL_MOD[m] for m in ("ctrl", "alt", "shift", "cmd") if m in parsed.modifiers] + key_code = YDOTOOL_KEY.get(parsed.key) + if key_code is None: + raise KeyParseError(f"no Linux input code for {parsed.key!r}") + codes.append(key_code) + return codes + + +# --------------------------------------------------------------------------- +# Windows — Win32 virtual-key codes. +# --------------------------------------------------------------------------- + +WIN_MOD_VK: Dict[str, int] = { + "ctrl": 0x11, # VK_CONTROL + "shift": 0x10, # VK_SHIFT + "alt": 0x12, # VK_MENU + "cmd": 0x5B, # VK_LWIN +} + +WIN_VK: Dict[str, int] = { + "return": 0x0D, "tab": 0x09, "space": 0x20, "escape": 0x1B, + "backspace": 0x08, "delete": 0x2E, "insert": 0x2D, "capslock": 0x14, + "up": 0x26, "down": 0x28, "left": 0x25, "right": 0x27, + "home": 0x24, "end": 0x23, "pageup": 0x21, "pagedown": 0x22, + "printscreen": 0x2C, + "minus": 0xBD, "equals": 0xBB, "comma": 0xBC, "period": 0xBE, + "slash": 0xBF, "backslash": 0xDC, "semicolon": 0xBA, "quote": 0xDE, + "leftbracket": 0xDB, "rightbracket": 0xDD, "backtick": 0xC0, +} +# Letters: A-Z VK codes are 0x41-0x5A +for _c in "abcdefghijklmnopqrstuvwxyz": + WIN_VK[_c] = 0x41 + (ord(_c) - ord("a")) +# Digits: 0x30-0x39 +for _c in "0123456789": + WIN_VK[_c] = 0x30 + (ord(_c) - ord("0")) +# Function keys F1-F24 +for _i in range(1, 25): + WIN_VK[f"f{_i}"] = 0x6F + _i # VK_F1=0x70 + + +def to_windows(parsed: ParsedKey) -> List[int]: + """Return [vk_code, ...] press order for SendInput.""" + codes = [WIN_MOD_VK[m] for m in ("ctrl", "alt", "shift", "cmd") if m in parsed.modifiers] + vk = WIN_VK.get(parsed.key) + if vk is None: + raise KeyParseError(f"no Windows VK code for {parsed.key!r}") + codes.append(vk) + return codes diff --git a/tools/computer_use_linux.py b/tools/computer_use_linux.py new file mode 100644 index 0000000000000..7f270f4b720f9 --- /dev/null +++ b/tools/computer_use_linux.py @@ -0,0 +1,552 @@ +"""Native Linux backend for computer_use. + +Linux is the awkward one — it's two display servers in a trench coat. The +backend detects ``$WAYLAND_DISPLAY`` / ``$XDG_SESSION_TYPE`` once at +import time and routes every action to the matching toolchain: + +* **X11 path** uses the venerable ``xdotool`` (``xdotool click``, + ``xdotool key``, ``xdotool mousemove``) plus ``scrot`` for screenshots. + Both packages are in every distro and have stable CLI surfaces. +* **Wayland path** uses ``ydotool`` (needs the ydotoold daemon running + and ``/dev/uinput`` accessible — that's a one-time host setup) plus + one of ``grim`` (wlroots compositors: Sway, Hyprland, labwc), + ``gnome-screenshot --file=`` (GNOME), or ``spectacle -bno`` (KDE). + Active-window / cursor-position queries are best-effort: Wayland's + application isolation makes some of these structurally impossible + on certain compositors. Tool returns empty dicts in that case + rather than failing. + +This backend is unit-test-only at the moment — author has no Linux +host immediately available for end-to-end validation. Mocked subprocess +tests cover every code path; real-world testing is deferred. +""" + +from __future__ import annotations + +import base64 +import logging +import os +import shutil +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from tools.computer_use_common import ( + ActionRequest, + ActionResult, + ValidationError, + build_schema, + parse_request, + validate_coords_within, +) +from tools.computer_use_grammar import parse_combo, to_xdotool, to_ydotool +from tools.computer_use_safety import ( + SafetyRefusal, + gate, + is_enabled, + log_action, + redact_image, +) +from tools.registry import registry + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Display server detection. We re-check on every action rather than caching +# at import — operators who run X11 sessions on a Wayland-default distro +# may have either set, and the env can flip across nested sessions. +# --------------------------------------------------------------------------- + +def _is_wayland() -> bool: + if os.environ.get("WAYLAND_DISPLAY"): + return True + if os.environ.get("XDG_SESSION_TYPE", "").lower() == "wayland": + return True + return False + + +def _x11_available() -> bool: + return bool(shutil.which("xdotool")) and bool(shutil.which("scrot") or shutil.which("import")) + + +def _wayland_available() -> bool: + has_input = bool(shutil.which("ydotool")) + has_capture = any(shutil.which(t) for t in ("grim", "gnome-screenshot", "spectacle")) + return has_input and has_capture + + +def _check_linux() -> bool: + if sys.platform != "linux": + return False + if not is_enabled(): + return False + return _x11_available() or _wayland_available() + + +# --------------------------------------------------------------------------- +# X11 implementations. +# --------------------------------------------------------------------------- + +def _run(cmd: List[str], *, timeout: float = 10.0, check: bool = True) -> subprocess.CompletedProcess: + return subprocess.run(cmd, check=check, capture_output=True, text=True, timeout=timeout) + + +def _x11_screenshot() -> bytes: + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: + path = tmp.name + try: + if shutil.which("scrot"): + _run(["scrot", "--silent", "--overwrite", path]) + else: + _run(["import", "-window", "root", path]) + return Path(path).read_bytes() + finally: + try: + Path(path).unlink() + except OSError: + pass + + +def _x11_screen_size() -> Tuple[int, int]: + if shutil.which("xdpyinfo"): + try: + out = _run(["xdpyinfo"]).stdout + for line in out.splitlines(): + line = line.strip() + if line.startswith("dimensions:"): + # "dimensions: 1920x1080 pixels (508x285 millimeters)" + parts = line.split() + if len(parts) >= 2 and "x" in parts[1]: + w, h = parts[1].split("x") + return int(w), int(h) + except (subprocess.SubprocessError, ValueError): + pass + if shutil.which("xrandr"): + try: + out = _run(["xrandr"]).stdout + for line in out.splitlines(): + if " connected primary " in line or " connected " in line: + # "HDMI-1 connected primary 1920x1080+0+0 ..." + for tok in line.split(): + if "x" in tok and "+" in tok: + res = tok.split("+", 1)[0] + if "x" in res: + w, h = res.split("x") + return int(w), int(h) + except (subprocess.SubprocessError, ValueError): + pass + return (0, 0) + + +def _x11_click(x: int, y: int, button: str = "left") -> None: + btn_map = {"left": "1", "middle": "2", "right": "3"} + _run(["xdotool", "mousemove", str(x), str(y), "click", btn_map[button]]) + + +def _x11_double_click(x: int, y: int) -> None: + _run(["xdotool", "mousemove", str(x), str(y), "click", "--repeat", "2", "--delay", "50", "1"]) + + +def _x11_button_event(x: int, y: int, kind: str) -> None: + cmd = "mousedown" if kind == "down" else "mouseup" + _run(["xdotool", "mousemove", str(x), str(y), cmd, "1"]) + + +def _x11_move(x: int, y: int) -> None: + _run(["xdotool", "mousemove", str(x), str(y)]) + + +def _x11_drag(x1: int, y1: int, x2: int, y2: int) -> None: + _run([ + "xdotool", "mousemove", str(x1), str(y1), + "mousedown", "1", + "mousemove", str(x2), str(y2), + "mouseup", "1", + ]) + + +def _x11_scroll(direction: str, amount: int) -> None: + btn = {"up": "4", "down": "5", "left": "6", "right": "7"}[direction] + _run(["xdotool", "click", "--repeat", str(amount), btn]) + + +def _x11_type(text: str) -> None: + _run(["xdotool", "type", "--delay", "5", "--", text]) + + +def _x11_key(combo: str) -> None: + parsed = parse_combo(combo) + arg = to_xdotool(parsed) + _run(["xdotool", "key", "--", arg]) + + +def _x11_cursor() -> Tuple[int, int]: + out = _run(["xdotool", "getmouselocation"]).stdout + # "x:123 y:456 screen:0 window:abc" + parts = dict(p.split(":") for p in out.split() if ":" in p) + return int(parts.get("x", 0)), int(parts.get("y", 0)) + + +def _x11_active_window() -> Optional[Dict[str, Any]]: + try: + wid = _run(["xdotool", "getactivewindow"]).stdout.strip() + if not wid: + return None + name = _run(["xdotool", "getwindowname", wid]).stdout.strip() + cls = "" + if shutil.which("xprop"): + xp = _run(["xprop", "-id", wid, "WM_CLASS"], check=False).stdout.strip() + if "=" in xp: + cls = xp.split("=", 1)[1].strip().strip('"').split('", "')[-1].strip('"') + return {"id": wid, "title": name, "app": cls} + except (subprocess.SubprocessError, ValueError): + return None + + +# --------------------------------------------------------------------------- +# Wayland implementations. +# --------------------------------------------------------------------------- + +def _wayland_screenshot() -> bytes: + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: + path = tmp.name + try: + if shutil.which("grim"): + _run(["grim", path]) + elif shutil.which("gnome-screenshot"): + _run(["gnome-screenshot", "-f", path]) + elif shutil.which("spectacle"): + _run(["spectacle", "-b", "-n", "-o", path]) + else: + raise RuntimeError("no Wayland screenshot tool available") + return Path(path).read_bytes() + finally: + try: + Path(path).unlink() + except OSError: + pass + + +def _wayland_screen_size() -> Tuple[int, int]: + # wlr-randr is the closest analog to xrandr on wlroots compositors. + if shutil.which("wlr-randr"): + try: + out = _run(["wlr-randr"]).stdout + for line in out.splitlines(): + line = line.strip() + if "current" in line.lower() and "x" in line: + parts = line.split() + for tok in parts: + if "x" in tok and tok[0].isdigit(): + try: + w, h = tok.split("x") + return int(w), int(h.split(",")[0].rstrip("p")) + except ValueError: + continue + except subprocess.SubprocessError: + pass + # swaymsg / hyprctl fallbacks + if shutil.which("swaymsg"): + try: + import json as _json + out = _run(["swaymsg", "-t", "get_outputs"]).stdout + for o in _json.loads(out): + if o.get("active"): + mode = o.get("current_mode") or {} + return int(mode.get("width", 0)), int(mode.get("height", 0)) + except (subprocess.SubprocessError, ValueError, KeyError): + pass + if shutil.which("hyprctl"): + try: + import json as _json + out = _run(["hyprctl", "-j", "monitors"]).stdout + for m in _json.loads(out): + if m.get("focused"): + return int(m.get("width", 0)), int(m.get("height", 0)) + except (subprocess.SubprocessError, ValueError, KeyError): + pass + return (0, 0) + + +def _ydotool_send_keys(codes: List[int]) -> None: + """Press codes in order, release in reverse — emulates a chord.""" + args = ["ydotool", "key"] + for code in codes: + args.append(f"{code}:1") + for code in reversed(codes): + args.append(f"{code}:0") + _run(args) + + +def _wayland_click(x: int, y: int, button: str = "left") -> None: + # ydotool button codes: BTN_LEFT=0xC0, BTN_RIGHT=0xC1, BTN_MIDDLE=0xC2 + code = {"left": 0xC0, "right": 0xC1, "middle": 0xC2}[button] + _run(["ydotool", "mousemove", "--absolute", "-x", str(x), "-y", str(y)]) + _run(["ydotool", "click", f"0x{code:X}"]) + + +def _wayland_double_click(x: int, y: int) -> None: + _wayland_click(x, y, "left") + time.sleep(0.05) + _wayland_click(x, y, "left") + + +def _wayland_button_event(x: int, y: int, kind: str) -> None: + code = 0xC0 + suffix = "1" if kind == "down" else "0" + _run(["ydotool", "mousemove", "--absolute", "-x", str(x), "-y", str(y)]) + _run(["ydotool", "click", f"{code:#x}:{suffix}"]) + + +def _wayland_move(x: int, y: int) -> None: + _run(["ydotool", "mousemove", "--absolute", "-x", str(x), "-y", str(y)]) + + +def _wayland_drag(x1: int, y1: int, x2: int, y2: int) -> None: + _wayland_button_event(x1, y1, "down") + _run(["ydotool", "mousemove", "--absolute", "-x", str(x2), "-y", str(y2)]) + _wayland_button_event(x2, y2, "up") + + +def _wayland_scroll(direction: str, amount: int) -> None: + # ydotool wheel positive = up, negative = down on most compositors. + delta = amount if direction == "up" else -amount if direction == "down" else 0 + if delta: + _run(["ydotool", "mousemove", "--wheel", "-y", str(delta)]) + + +def _wayland_type(text: str) -> None: + _run(["ydotool", "type", "--", text]) + + +def _wayland_key(combo: str) -> None: + parsed = parse_combo(combo) + codes = to_ydotool(parsed) + _ydotool_send_keys(codes) + + +def _wayland_cursor() -> Tuple[int, int]: + # No portable Wayland cursor query. Best-effort via Sway IPC if present. + if shutil.which("swaymsg"): + try: + import json as _json + out = _run(["swaymsg", "-t", "get_seats"]).stdout + for s in _json.loads(out): + pos = s.get("pointer", {}) + if "x" in pos and "y" in pos: + return int(pos["x"]), int(pos["y"]) + except (subprocess.SubprocessError, ValueError, KeyError): + pass + return (0, 0) + + +def _wayland_active_window() -> Optional[Dict[str, Any]]: + if shutil.which("swaymsg"): + try: + import json as _json + out = _run(["swaymsg", "-t", "get_tree"]).stdout + tree = _json.loads(out) + + def walk(node): + if node.get("focused"): + return node + for child in (node.get("nodes") or []) + (node.get("floating_nodes") or []): + found = walk(child) + if found: + return found + return None + + focused = walk(tree) + if focused: + return { + "id": str(focused.get("id", "")), + "title": focused.get("name") or "", + "app": (focused.get("app_id") or focused.get("window_properties", {}).get("class") or ""), + } + except (subprocess.SubprocessError, ValueError, KeyError): + pass + if shutil.which("hyprctl"): + try: + import json as _json + out = _run(["hyprctl", "-j", "activewindow"]).stdout + w = _json.loads(out) + return {"id": str(w.get("address", "")), "title": w.get("title", ""), "app": w.get("class", "")} + except (subprocess.SubprocessError, ValueError, KeyError): + pass + return None + + +# --------------------------------------------------------------------------- +# Routing layer. +# --------------------------------------------------------------------------- + +class Backend: + screenshot = staticmethod(lambda: b"") + screen_size = staticmethod(lambda: (0, 0)) + click = staticmethod(lambda x, y, b="left": None) + double_click = staticmethod(lambda x, y: None) + button_event = staticmethod(lambda x, y, kind: None) + move = staticmethod(lambda x, y: None) + drag = staticmethod(lambda x1, y1, x2, y2: None) + scroll = staticmethod(lambda d, a: None) + type_text = staticmethod(lambda t: None) + key = staticmethod(lambda c: None) + cursor = staticmethod(lambda: (0, 0)) + active_window = staticmethod(lambda: None) + + +def _select_backend() -> Backend: + """Choose X11 or Wayland routing once per call.""" + use_wayland = _is_wayland() and _wayland_available() + if use_wayland: + b = Backend() + b.screenshot = _wayland_screenshot + b.screen_size = _wayland_screen_size + b.click = _wayland_click + b.double_click = _wayland_double_click + b.button_event = _wayland_button_event + b.move = _wayland_move + b.drag = _wayland_drag + b.scroll = _wayland_scroll + b.type_text = _wayland_type + b.key = _wayland_key + b.cursor = _wayland_cursor + b.active_window = _wayland_active_window + return b + + if _x11_available(): + b = Backend() + b.screenshot = _x11_screenshot + b.screen_size = _x11_screen_size + b.click = _x11_click + b.double_click = _x11_double_click + b.button_event = _x11_button_event + b.move = _x11_move + b.drag = _x11_drag + b.scroll = _x11_scroll + b.type_text = _x11_type + b.key = _x11_key + b.cursor = _x11_cursor + b.active_window = _x11_active_window + return b + + raise RuntimeError( + "no Linux desktop automation backend available; install xdotool+scrot " + "for X11 or ydotool+grim for Wayland" + ) + + +def _crop_region(png_bytes: bytes, region) -> bytes: + try: + from PIL import Image # type: ignore + except ImportError: + return png_bytes + import io as _io + x1, y1, x2, y2 = (int(v) for v in region) + img = Image.open(_io.BytesIO(png_bytes)).convert("RGB") + cropped = img.crop((x1, y1, x2, y2)) + buf = _io.BytesIO() + cropped.save(buf, format="PNG", optimize=True) + return buf.getvalue() + + +def _handle(req: ActionRequest, backend: Backend) -> ActionResult: + a = req.action + + if a == "screen_size": + w, h = backend.screen_size() + return ActionResult(success=True, action=a, screen_width=w, screen_height=h) + + if a == "cursor_position": + x, y = backend.cursor() + return ActionResult(success=True, action=a, cursor_x=x, cursor_y=y) + + if a == "get_active_window": + return ActionResult(success=True, action=a, active_window=backend.active_window() or {}) + + if a == "screenshot": + png = backend.screenshot() + if req.region: + png = _crop_region(png, req.region) + if req.redact_regions: + png = redact_image(png, req.redact_regions) + return ActionResult( + success=True, + action=a, + screenshot_b64=base64.b64encode(png).decode("ascii"), + ) + + if a == "wait": + time.sleep((req.ms or 0) / 1000.0) + return ActionResult(success=True, action=a, message=f"waited {req.ms}ms") + + sw, sh = backend.screen_size() + if sw and sh: + validate_coords_within(req, sw, sh) + + if a == "left_click": + backend.click(req.x, req.y, "left") + elif a == "double_click": + backend.double_click(req.x, req.y) + elif a == "right_click": + backend.click(req.x, req.y, "right") + elif a == "middle_click": + backend.click(req.x, req.y, "middle") + elif a == "left_button_press": + backend.button_event(req.x, req.y, "down") + elif a == "left_button_release": + backend.button_event(req.x, req.y, "up") + elif a == "mouse_move": + backend.move(req.x, req.y) + elif a == "mouse_drag": + backend.drag(req.x, req.y, req.x2, req.y2) + elif a == "scroll": + backend.move(req.x, req.y) + backend.scroll(req.direction, req.amount or 3) + elif a == "type": + backend.type_text(req.text) + elif a == "key": + backend.key(req.keys) + else: + return ActionResult(success=False, action=a, error=f"unhandled action {a!r}") + + return ActionResult(success=True, action=a, screen_width=sw, screen_height=sh) + + +def computer_use_linux_tool(args: Dict[str, Any]) -> Dict[str, Any]: + try: + gate(args.get("action", "")) + req = parse_request(args) + backend = _select_backend() + result = _handle(req, backend) + except SafetyRefusal as e: + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"refused: {e}") + except ValidationError as e: + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"validation: {e}") + except subprocess.CalledProcessError as e: + logger.error("backend subprocess failed: %s\nstderr: %s", e, e.stderr) + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"backend: {e.stderr or e}") + except Exception as e: + logger.exception("computer_use_linux handler failed") + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"runtime: {e}") + + log_action(result.action, args, result.success, result.error) + return result.to_dict() + + +registry.register( + name="computer_use_linux", + toolset="computer_use", + schema=build_schema("computer_use_linux", "Linux"), + handler=lambda args, **kw: computer_use_linux_tool(args), + check_fn=_check_linux, + requires_env=["HERMES_COMPUTER_USE_ENABLED"], + is_async=False, + description="Native Linux desktop control (X11 + Wayland; xdotool/scrot or ydotool/grim).", + emoji="🐧", + max_result_size_chars=8_000_000, +) diff --git a/tools/computer_use_macos.py b/tools/computer_use_macos.py new file mode 100644 index 0000000000000..654e08617c557 --- /dev/null +++ b/tools/computer_use_macos.py @@ -0,0 +1,375 @@ +"""Native macOS backend for computer_use. + +Mouse and keyboard go through Quartz ``CGEvent`` calls (via +``pyobjc-framework-Quartz``); screenshots use Apple's ``screencapture`` +CLI which is always present on macOS and produces clean PNGs without +any extra dependencies. + +Two macOS-specific gotchas the user has to handle once, outside of code: + +* **Accessibility permission** — required for synthetic mouse/keyboard + events to reach other applications. macOS prompts the first time a + CGEvent is posted; the operator grants under + *System Settings → Privacy & Security → Accessibility*. +* **Screen Recording permission** — required for screenshots that + include other application windows. ``screencapture`` shows an + unrecognised permission dialog the first time it runs. + +These cannot be granted programmatically; the per-OS skill prompts the +operator through the setup once. + +The tool registers unconditionally so the AST tool scanner picks it up, +but ``check_fn`` returns False on non-Darwin hosts, when pyobjc isn't +importable, or when ``HERMES_COMPUTER_USE_ENABLED`` isn't truthy. So +agents on Linux/Windows never see this tool offered. +""" + +from __future__ import annotations + +import base64 +import logging +import shutil +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from typing import Any, Dict, Optional, Tuple + +from tools.computer_use_common import ( + ActionRequest, + ActionResult, + ValidationError, + build_schema, + parse_request, + validate_coords_within, +) +from tools.computer_use_grammar import parse_combo, to_macos +from tools.computer_use_safety import ( + SafetyRefusal, + gate, + is_enabled, + log_action, + redact_image, +) +from tools.registry import registry + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Lazy pyobjc loader. Called from check_fn and from each handler that needs +# Quartz. Cached on the module to avoid re-importing on every action. +# --------------------------------------------------------------------------- + +_quartz: Optional[Any] = None + + +def _load_quartz() -> Optional[Any]: + """Import Quartz lazily; return module or None if unavailable.""" + global _quartz + if _quartz is not None: + return _quartz + try: + import Quartz # type: ignore + except ImportError: + return None + _quartz = Quartz + return _quartz + + +def _check_macos() -> bool: + """check_fn for the registry — gates visibility of the tool.""" + if sys.platform != "darwin": + return False + if not is_enabled(): + return False + if _load_quartz() is None: + return False + if not shutil.which("screencapture"): + return False + return True + + +# --------------------------------------------------------------------------- +# Screenshots — call /usr/sbin/screencapture into a tempfile, read bytes. +# screencapture takes ~80-150ms; cheap enough for an action loop. +# --------------------------------------------------------------------------- + +def _screenshot_full() -> bytes: + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: + path = tmp.name + try: + subprocess.run( + ["screencapture", "-x", "-t", "png", path], + check=True, + capture_output=True, + timeout=10, + ) + return Path(path).read_bytes() + finally: + try: + Path(path).unlink() + except OSError: + pass + + +def _screen_size() -> Tuple[int, int]: + """Return (width, height) of the main display in pixels.""" + Q = _load_quartz() + if Q is None: + return (0, 0) + main = Q.CGMainDisplayID() + return int(Q.CGDisplayPixelsWide(main)), int(Q.CGDisplayPixelsHigh(main)) + + +# --------------------------------------------------------------------------- +# Mouse + keyboard — Quartz CGEvent. All synthetic events use HID source. +# --------------------------------------------------------------------------- + +def _post_mouse(event_type: int, x: int, y: int, button: int = 0) -> None: + Q = _load_quartz() + assert Q is not None, "pyobjc Quartz must be loaded before _post_mouse" + event = Q.CGEventCreateMouseEvent( + None, event_type, (x, y), button, + ) + Q.CGEventPost(Q.kCGHIDEventTap, event) + + +def _click(x: int, y: int, button: str = "left") -> None: + Q = _load_quartz() + assert Q is not None + btn_map = { + "left": (Q.kCGEventLeftMouseDown, Q.kCGEventLeftMouseUp, Q.kCGMouseButtonLeft), + "right": (Q.kCGEventRightMouseDown, Q.kCGEventRightMouseUp, Q.kCGMouseButtonRight), + "middle": (Q.kCGEventOtherMouseDown, Q.kCGEventOtherMouseUp, Q.kCGMouseButtonCenter), + } + down, up, btn = btn_map[button] + _post_mouse(down, x, y, btn) + time.sleep(0.02) + _post_mouse(up, x, y, btn) + + +def _double_click(x: int, y: int) -> None: + Q = _load_quartz() + assert Q is not None + # Quartz needs the same event with click count = 2 to register a double-click + # in apps like Finder. Easiest: post two clicks within the system threshold. + _click(x, y, "left") + time.sleep(0.05) + _click(x, y, "left") + + +def _move(x: int, y: int) -> None: + Q = _load_quartz() + assert Q is not None + _post_mouse(Q.kCGEventMouseMoved, x, y, 0) + + +def _drag(x1: int, y1: int, x2: int, y2: int) -> None: + Q = _load_quartz() + assert Q is not None + _post_mouse(Q.kCGEventLeftMouseDown, x1, y1, Q.kCGMouseButtonLeft) + time.sleep(0.05) + # Smooth interpolation so apps that need motion events (e.g. drag-drop + # validators) actually see the drag rather than a teleport. + steps = max(10, int(((x2 - x1) ** 2 + (y2 - y1) ** 2) ** 0.5 / 20)) + for i in range(1, steps + 1): + ix = x1 + (x2 - x1) * i // steps + iy = y1 + (y2 - y1) * i // steps + _post_mouse(Q.kCGEventLeftMouseDragged, ix, iy, Q.kCGMouseButtonLeft) + time.sleep(0.005) + _post_mouse(Q.kCGEventLeftMouseUp, x2, y2, Q.kCGMouseButtonLeft) + + +def _scroll(direction: str, amount: int) -> None: + Q = _load_quartz() + assert Q is not None + dy = -amount if direction == "down" else amount if direction == "up" else 0 + dx = -amount if direction == "right" else amount if direction == "left" else 0 + event = Q.CGEventCreateScrollWheelEvent( + None, Q.kCGScrollEventUnitLine, 2, dy, dx, + ) + Q.CGEventPost(Q.kCGHIDEventTap, event) + + +def _type_text(text: str) -> None: + Q = _load_quartz() + assert Q is not None + # Use CGEventKeyboardSetUnicodeString for full Unicode coverage — + # avoids per-character keycode lookups and works for non-ASCII. + for ch in text: + for kind in (True, False): + event = Q.CGEventCreateKeyboardEvent(None, 0, kind) + Q.CGEventKeyboardSetUnicodeString(event, 1, ch) + Q.CGEventPost(Q.kCGHIDEventTap, event) + time.sleep(0.005) + + +def _key_combo(combo: str) -> None: + Q = _load_quartz() + assert Q is not None + parsed = parse_combo(combo) + flags, keycode = to_macos(parsed) + + down = Q.CGEventCreateKeyboardEvent(None, keycode, True) + if flags: + Q.CGEventSetFlags(down, flags) + Q.CGEventPost(Q.kCGHIDEventTap, down) + + up = Q.CGEventCreateKeyboardEvent(None, keycode, False) + if flags: + Q.CGEventSetFlags(up, flags) + Q.CGEventPost(Q.kCGHIDEventTap, up) + + +def _cursor_position() -> Tuple[int, int]: + Q = _load_quartz() + assert Q is not None + event = Q.CGEventCreate(None) + pt = Q.CGEventGetLocation(event) + return int(pt.x), int(pt.y) + + +def _active_window() -> Optional[Dict[str, Any]]: + """Return basic info about the frontmost window.""" + Q = _load_quartz() + if Q is None: + return None + info_list = Q.CGWindowListCopyWindowInfo( + Q.kCGWindowListOptionOnScreenOnly | Q.kCGWindowListExcludeDesktopElements, + Q.kCGNullWindowID, + ) + if not info_list: + return None + # Frontmost is highest layer; pick the first on-screen window with a name. + for w in info_list: + name = w.get("kCGWindowName") or "" + owner = w.get("kCGWindowOwnerName") or "" + if owner: + return {"app": str(owner), "title": str(name)} + return None + + +# --------------------------------------------------------------------------- +# Handler entry point. +# --------------------------------------------------------------------------- + +def _handle(req: ActionRequest) -> ActionResult: + a = req.action + + if a == "screen_size": + w, h = _screen_size() + return ActionResult(success=True, action=a, screen_width=w, screen_height=h) + + if a == "cursor_position": + x, y = _cursor_position() + return ActionResult(success=True, action=a, cursor_x=x, cursor_y=y) + + if a == "get_active_window": + win = _active_window() + return ActionResult(success=True, action=a, active_window=win or {}) + + if a == "screenshot": + png = _screenshot_full() + if req.region: + png = _crop_region(png, req.region) + if req.redact_regions: + png = redact_image(png, req.redact_regions) + return ActionResult( + success=True, + action=a, + screenshot_b64=base64.b64encode(png).decode("ascii"), + ) + + if a == "wait": + time.sleep((req.ms or 0) / 1000.0) + return ActionResult(success=True, action=a, message=f"waited {req.ms}ms") + + # Coordinate-bearing actions: validate against screen bounds. + sw, sh = _screen_size() + validate_coords_within(req, sw, sh) + + if a == "left_click": + _click(req.x, req.y, "left") + elif a == "double_click": + _double_click(req.x, req.y) + elif a == "right_click": + _click(req.x, req.y, "right") + elif a == "middle_click": + _click(req.x, req.y, "middle") + elif a == "left_button_press": + Q = _load_quartz() + _post_mouse(Q.kCGEventLeftMouseDown, req.x, req.y, Q.kCGMouseButtonLeft) + elif a == "left_button_release": + Q = _load_quartz() + _post_mouse(Q.kCGEventLeftMouseUp, req.x, req.y, Q.kCGMouseButtonLeft) + elif a == "mouse_move": + _move(req.x, req.y) + elif a == "mouse_drag": + _drag(req.x, req.y, req.x2, req.y2) + elif a == "scroll": + _move(req.x, req.y) + _scroll(req.direction, req.amount or 3) + elif a == "type": + _type_text(req.text) + elif a == "key": + _key_combo(req.keys) + else: + return ActionResult(success=False, action=a, error=f"unhandled action {a!r}") + + return ActionResult(success=True, action=a, screen_width=sw, screen_height=sh) + + +def _crop_region(png_bytes: bytes, region) -> bytes: + try: + from PIL import Image # type: ignore + except ImportError: + return png_bytes + import io as _io + x1, y1, x2, y2 = (int(v) for v in region) + img = Image.open(_io.BytesIO(png_bytes)).convert("RGB") + cropped = img.crop((x1, y1, x2, y2)) + buf = _io.BytesIO() + cropped.save(buf, format="PNG", optimize=True) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# Tool entry point — called by the registry dispatcher. +# --------------------------------------------------------------------------- + +def computer_use_macos_tool(args: Dict[str, Any]) -> Dict[str, Any]: + try: + gate(args.get("action", "")) + req = parse_request(args) + result = _handle(req) + except SafetyRefusal as e: + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"refused: {e}") + except ValidationError as e: + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"validation: {e}") + except Exception as e: + logger.exception("computer_use_macos handler failed") + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"runtime: {e}") + + log_action(result.action, args, result.success, result.error) + return result.to_dict() + + +# --------------------------------------------------------------------------- +# Registration. AST scanner expects a top-level call; check_fn gates +# availability per platform/permission. +# --------------------------------------------------------------------------- + +registry.register( + name="computer_use_macos", + toolset="computer_use", + schema=build_schema("computer_use_macos", "macOS"), + handler=lambda args, **kw: computer_use_macos_tool(args), + check_fn=_check_macos, + requires_env=["HERMES_COMPUTER_USE_ENABLED"], + is_async=False, + description="Native macOS desktop control (screenshot, click, type, key combos).", + emoji="🖱", + max_result_size_chars=8_000_000, # base64 PNG can run large +) diff --git a/tools/computer_use_safety.py b/tools/computer_use_safety.py new file mode 100644 index 0000000000000..b8ee8f5c1c54c --- /dev/null +++ b/tools/computer_use_safety.py @@ -0,0 +1,179 @@ +"""Safety surface for the computer_use_* tools. + +Three concerns this module owns: + +1. **Default-off env gate** — the tool refuses to register unless + ``HERMES_COMPUTER_USE_ENABLED=true``. This mirrors the gating pattern + used by issue #15876's containerised proposal and keeps the feature + inert for users who haven't opted in. +2. **Append-only action log** — every action attempt is JSON-line logged + to ``$HERMES_HOME/logs/computer_use.jsonl`` with timestamp, action name, + parameters (without screenshot bytes), and the success bit. Useful for + post-incident review when an agent does something unexpected. +3. **Screenshot redaction** — caller-supplied rectangles are blanked in + the returned PNG before it reaches the model. The default ships with + redaction OFF; the model has to actively request it via the + ``redact_regions`` arg or the operator can wire a SOUL.md rule that + always redacts a known sensitive zone (e.g. password manager popup). + +The kill-switch hotkey hook is left as a flag-poll helper here; OS +backends that can register a global hotkey should set the flag from +their hotkey handler. A polling check is good enough — the worst case +is one in-flight action completes before the next one is blocked. +""" + +from __future__ import annotations + +import io +import json +import logging +import os +import threading +import time +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +logger = logging.getLogger(__name__) + + +COMPUTER_USE_ENV = "HERMES_COMPUTER_USE_ENABLED" + + +def is_enabled() -> bool: + """True iff the operator has explicitly opted in via env var. + + Accepts ``true``/``1``/``yes`` (case-insensitive). Anything else + (unset, ``false``, ``0``, junk) returns False. + """ + val = os.environ.get(COMPUTER_USE_ENV, "").strip().lower() + return val in {"true", "1", "yes", "on"} + + +# --------------------------------------------------------------------------- +# Kill-switch flag (process-global). OS backends call set_kill_switch() from +# their hotkey handler; handler() polls is_killed() before every action. +# --------------------------------------------------------------------------- + +_kill_switch = threading.Event() + + +def is_killed() -> bool: + return _kill_switch.is_set() + + +def set_kill_switch() -> None: + _kill_switch.set() + logger.warning("computer_use kill switch ENGAGED — subsequent actions will refuse") + + +def clear_kill_switch() -> None: + _kill_switch.clear() + + +# --------------------------------------------------------------------------- +# Append-only action log. +# --------------------------------------------------------------------------- + +def _log_path() -> Path: + home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")) + log_dir = Path(home) / "logs" + try: + log_dir.mkdir(parents=True, exist_ok=True) + except OSError: + pass + return log_dir / "computer_use.jsonl" + + +_log_lock = threading.Lock() + + +def log_action(action: str, params: Dict[str, Any], success: bool, error: Optional[str] = None) -> None: + """Append one JSON line describing an action attempt. + + Strips screenshot bytes / large blobs from logged params so the log + file stays browsable. + """ + safe_params = { + k: ("" if isinstance(v, (bytes, bytearray)) else v) + for k, v in params.items() + if k not in {"screenshot_b64"} + } + record = { + "ts": time.time(), + "action": action, + "params": safe_params, + "success": success, + } + if error: + record["error"] = error[:500] + line = json.dumps(record, ensure_ascii=False, default=str) + try: + with _log_lock: + with _log_path().open("a", encoding="utf-8") as f: + f.write(line + "\n") + except OSError as e: + logger.debug("could not write computer_use log: %s", e) + + +# --------------------------------------------------------------------------- +# Screenshot redaction. Pure-Python via PIL — already a Hermes dependency. +# --------------------------------------------------------------------------- + +def redact_image(png_bytes: bytes, regions: Iterable[List[int]]) -> bytes: + """Return a copy of ``png_bytes`` with each region rectangle filled black. + + Regions are ``[x1, y1, x2, y2]`` in image pixel space. Coordinates that + fall outside the image are clamped silently. If PIL isn't importable we + fall back to returning the original bytes (logged as a warning) so a + missing dependency degrades to "no redaction" rather than "no screenshot". + """ + try: + from PIL import Image, ImageDraw # type: ignore + except ImportError: + logger.warning("PIL unavailable — redaction skipped, returning unredacted screenshot") + return png_bytes + + if not regions: + return png_bytes + + img = Image.open(io.BytesIO(png_bytes)).convert("RGB") + draw = ImageDraw.Draw(img) + w, h = img.size + for r in regions: + if len(r) != 4: + continue + x1, y1, x2, y2 = (int(v) for v in r) + x1 = max(0, min(x1, w)) + y1 = max(0, min(y1, h)) + x2 = max(0, min(x2, w)) + y2 = max(0, min(y2, h)) + if x2 <= x1 or y2 <= y1: + continue + draw.rectangle([x1, y1, x2, y2], fill=(0, 0, 0)) + + buf = io.BytesIO() + img.save(buf, format="PNG", optimize=True) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# Common pre-action gate. Each OS backend calls this at the top of every +# action handler. +# --------------------------------------------------------------------------- + +class SafetyRefusal(Exception): + """Raised by gate() when an action must be refused for safety reasons.""" + + +def gate(action: str) -> None: + """Refuse the action if safety conditions aren't met.""" + if not is_enabled(): + raise SafetyRefusal( + f"{COMPUTER_USE_ENV} is not set — computer_use is disabled. " + f"Set the env var to true to opt in." + ) + if is_killed(): + raise SafetyRefusal( + "computer_use kill switch is engaged. Call clear_kill_switch() " + "or restart the gateway to resume." + ) diff --git a/tools/computer_use_windows.py b/tools/computer_use_windows.py new file mode 100644 index 0000000000000..05af89952c749 --- /dev/null +++ b/tools/computer_use_windows.py @@ -0,0 +1,492 @@ +"""Native Windows backend for computer_use. + +Mouse and keyboard go through ``user32.SendInput`` via ``ctypes`` — +``SendInput`` is the modern Win32 input API and the only one that works +reliably with DirectInput games and UAC-aware applications. The legacy +``keybd_event`` / ``mouse_event`` calls are avoided. + +Screenshots use ``mss`` when available (BitBlt-based, MIT-licensed, +fast and cross-Windows-version). When ``mss`` isn't installed we fall +back to a minimal ctypes BitBlt path so the tool still functions on a +default Python install. + +Two Windows-specific gotchas: + +* **UAC / UIPI** — synthetic input from a non-elevated process can't + reach an elevated window (User Interface Privilege Isolation). If + the agent's host process is non-elevated and the target window is, + clicks land in dead air. Run the gateway elevated for full coverage. +* **DPI awareness** — the tool runs the process as per-monitor DPI + aware so click coordinates match what the user sees on HiDPI + displays. Without this, a click at (1000, 1000) on a 200 % screen + lands at (500, 500). + +This backend is unit-test-only — author has no Windows host for +integration validation. Mocks cover every code path. +""" + +from __future__ import annotations + +import base64 +import ctypes +import ctypes.wintypes as wt +import logging +import os +import sys +import time +from typing import Any, Dict, List, Optional, Tuple + +from tools.computer_use_common import ( + ActionRequest, + ActionResult, + ValidationError, + build_schema, + parse_request, + validate_coords_within, +) +from tools.computer_use_grammar import parse_combo, to_windows +from tools.computer_use_safety import ( + SafetyRefusal, + gate, + is_enabled, + log_action, + redact_image, +) +from tools.registry import registry + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Lazy load user32 + supporting structs. Importing this module on non-Windows +# must not raise, hence the guarded loader. +# --------------------------------------------------------------------------- + +_user32 = None +_gdi32 = None + + +def _load_user32(): + global _user32, _gdi32 + if _user32 is not None: + return _user32 + if sys.platform != "win32": + return None + try: + _user32 = ctypes.WinDLL("user32", use_last_error=True) + _gdi32 = ctypes.WinDLL("gdi32", use_last_error=True) + try: + # Per-monitor v2 DPI awareness so coordinates aren't scaled. + _user32.SetProcessDpiAwarenessContext(ctypes.c_void_p(-4)) + except (AttributeError, OSError): + pass + return _user32 + except OSError: + return None + + +def _check_windows() -> bool: + if sys.platform != "win32": + return False + if not is_enabled(): + return False + return _load_user32() is not None + + +# --------------------------------------------------------------------------- +# SendInput structures. +# --------------------------------------------------------------------------- + +ULONG_PTR = ctypes.c_size_t + + +class _MOUSEINPUT(ctypes.Structure): + _fields_ = [ + ("dx", wt.LONG), + ("dy", wt.LONG), + ("mouseData", wt.DWORD), + ("dwFlags", wt.DWORD), + ("time", wt.DWORD), + ("dwExtraInfo", ULONG_PTR), + ] + + +class _KEYBDINPUT(ctypes.Structure): + _fields_ = [ + ("wVk", wt.WORD), + ("wScan", wt.WORD), + ("dwFlags", wt.DWORD), + ("time", wt.DWORD), + ("dwExtraInfo", ULONG_PTR), + ] + + +class _HARDWAREINPUT(ctypes.Structure): + _fields_ = [("uMsg", wt.DWORD), ("wParamL", wt.WORD), ("wParamH", wt.WORD)] + + +class _INPUTunion(ctypes.Union): + _fields_ = [("mi", _MOUSEINPUT), ("ki", _KEYBDINPUT), ("hi", _HARDWAREINPUT)] + + +class _INPUT(ctypes.Structure): + _anonymous_ = ("u",) + _fields_ = [("type", wt.DWORD), ("u", _INPUTunion)] + + +INPUT_MOUSE = 0 +INPUT_KEYBOARD = 1 + +# Mouse flags +MOUSEEVENTF_MOVE = 0x0001 +MOUSEEVENTF_LEFTDOWN = 0x0002 +MOUSEEVENTF_LEFTUP = 0x0004 +MOUSEEVENTF_RIGHTDOWN = 0x0008 +MOUSEEVENTF_RIGHTUP = 0x0010 +MOUSEEVENTF_MIDDLEDOWN = 0x0020 +MOUSEEVENTF_MIDDLEUP = 0x0040 +MOUSEEVENTF_WHEEL = 0x0800 +MOUSEEVENTF_HWHEEL = 0x01000 +MOUSEEVENTF_ABSOLUTE = 0x8000 + +# Keyboard flags +KEYEVENTF_KEYUP = 0x0002 +KEYEVENTF_UNICODE = 0x0004 + + +def _send_inputs(inputs: List[_INPUT]) -> None: + user32 = _load_user32() + assert user32 is not None + n = len(inputs) + arr_t = _INPUT * n + arr = arr_t(*inputs) + sent = user32.SendInput(n, ctypes.byref(arr), ctypes.sizeof(_INPUT)) + if sent != n: + err = ctypes.get_last_error() + raise RuntimeError(f"SendInput sent {sent}/{n} (last error {err})") + + +def _mouse_input(flags: int, dx: int = 0, dy: int = 0, data: int = 0) -> _INPUT: + inp = _INPUT() + inp.type = INPUT_MOUSE + inp.mi = _MOUSEINPUT(dx=dx, dy=dy, mouseData=data, dwFlags=flags, time=0, dwExtraInfo=0) + return inp + + +def _key_input(vk: int, key_up: bool = False, scan: int = 0, unicode_char: bool = False) -> _INPUT: + inp = _INPUT() + inp.type = INPUT_KEYBOARD + flags = 0 + if key_up: + flags |= KEYEVENTF_KEYUP + if unicode_char: + flags |= KEYEVENTF_UNICODE + inp.ki = _KEYBDINPUT(wVk=vk, wScan=scan, dwFlags=flags, time=0, dwExtraInfo=0) + return inp + + +# --------------------------------------------------------------------------- +# Mouse helpers — convert absolute pixel to virtual desktop normalized coords. +# --------------------------------------------------------------------------- + +SM_CXSCREEN = 0 +SM_CYSCREEN = 1 + + +def _screen_size() -> Tuple[int, int]: + user32 = _load_user32() + if user32 is None: + return (0, 0) + return int(user32.GetSystemMetrics(SM_CXSCREEN)), int(user32.GetSystemMetrics(SM_CYSCREEN)) + + +def _to_normalized(x: int, y: int) -> Tuple[int, int]: + sw, sh = _screen_size() + sw = max(sw, 1) + sh = max(sh, 1) + return int(x * 65535 / sw), int(y * 65535 / sh) + + +def _move(x: int, y: int) -> None: + nx, ny = _to_normalized(x, y) + _send_inputs([_mouse_input(MOUSEEVENTF_MOVE | MOUSEEVENTF_ABSOLUTE, nx, ny)]) + + +def _click(x: int, y: int, button: str = "left") -> None: + _move(x, y) + if button == "left": + down, up = MOUSEEVENTF_LEFTDOWN, MOUSEEVENTF_LEFTUP + elif button == "right": + down, up = MOUSEEVENTF_RIGHTDOWN, MOUSEEVENTF_RIGHTUP + else: + down, up = MOUSEEVENTF_MIDDLEDOWN, MOUSEEVENTF_MIDDLEUP + _send_inputs([_mouse_input(down), _mouse_input(up)]) + + +def _double_click(x: int, y: int) -> None: + _click(x, y, "left") + time.sleep(0.05) + _click(x, y, "left") + + +def _button_event(x: int, y: int, kind: str) -> None: + _move(x, y) + flag = MOUSEEVENTF_LEFTDOWN if kind == "down" else MOUSEEVENTF_LEFTUP + _send_inputs([_mouse_input(flag)]) + + +def _drag(x1: int, y1: int, x2: int, y2: int) -> None: + _button_event(x1, y1, "down") + steps = max(10, int(((x2 - x1) ** 2 + (y2 - y1) ** 2) ** 0.5 / 20)) + for i in range(1, steps + 1): + ix = x1 + (x2 - x1) * i // steps + iy = y1 + (y2 - y1) * i // steps + _move(ix, iy) + time.sleep(0.005) + _button_event(x2, y2, "up") + + +def _scroll(direction: str, amount: int) -> None: + # WHEEL_DELTA = 120 per notch + delta = 120 * amount + if direction == "down": + delta = -delta + if direction in ("up", "down"): + _send_inputs([_mouse_input(MOUSEEVENTF_WHEEL, data=ctypes.c_int32(delta).value)]) + else: + if direction == "left": + delta = -delta + _send_inputs([_mouse_input(MOUSEEVENTF_HWHEEL, data=ctypes.c_int32(delta).value)]) + + +def _type_text(text: str) -> None: + inputs: List[_INPUT] = [] + for ch in text: + code = ord(ch) + # KEYEVENTF_UNICODE drives wScan with the codepoint, vk=0. + inputs.append(_key_input(0, key_up=False, scan=code, unicode_char=True)) + inputs.append(_key_input(0, key_up=True, scan=code, unicode_char=True)) + if len(inputs) >= 100: + _send_inputs(inputs) + inputs = [] + if inputs: + _send_inputs(inputs) + + +def _key_combo(combo: str) -> None: + parsed = parse_combo(combo) + vks = to_windows(parsed) + inputs: List[_INPUT] = [_key_input(vk, key_up=False) for vk in vks] + inputs += [_key_input(vk, key_up=True) for vk in reversed(vks)] + _send_inputs(inputs) + + +def _cursor_position() -> Tuple[int, int]: + user32 = _load_user32() + if user32 is None: + return (0, 0) + pt = wt.POINT() + user32.GetCursorPos(ctypes.byref(pt)) + return int(pt.x), int(pt.y) + + +def _active_window() -> Optional[Dict[str, Any]]: + user32 = _load_user32() + if user32 is None: + return None + hwnd = user32.GetForegroundWindow() + if not hwnd: + return None + length = user32.GetWindowTextLengthW(hwnd) + 1 + buf = ctypes.create_unicode_buffer(length) + user32.GetWindowTextW(hwnd, buf, length) + return {"id": str(int(hwnd)), "title": buf.value} + + +# --------------------------------------------------------------------------- +# Screenshot — mss preferred, ctypes BitBlt fallback. +# --------------------------------------------------------------------------- + +def _screenshot_mss() -> Optional[bytes]: + try: + import mss # type: ignore + import mss.tools # type: ignore + except ImportError: + return None + with mss.mss() as sct: + monitor = sct.monitors[1] # primary + sct_img = sct.grab(monitor) + return mss.tools.to_png(sct_img.rgb, sct_img.size) + + +def _screenshot_bitblt() -> bytes: + """Minimal ctypes BitBlt screenshot. Slower but no extra deps.""" + user32 = _load_user32() + gdi32 = _gdi32 + assert user32 is not None and gdi32 is not None + sw, sh = _screen_size() + # GetDC, CreateCompatibleDC, CreateCompatibleBitmap, BitBlt → DIB + hdc_screen = user32.GetDC(0) + hdc_mem = gdi32.CreateCompatibleDC(hdc_screen) + hbm = gdi32.CreateCompatibleBitmap(hdc_screen, sw, sh) + gdi32.SelectObject(hdc_mem, hbm) + SRCCOPY = 0x00CC0020 + gdi32.BitBlt(hdc_mem, 0, 0, sw, sh, hdc_screen, 0, 0, SRCCOPY) + + # Read pixels as 32bpp RGBA + class BITMAPINFOHEADER(ctypes.Structure): + _fields_ = [ + ("biSize", wt.DWORD), ("biWidth", wt.LONG), ("biHeight", wt.LONG), + ("biPlanes", wt.WORD), ("biBitCount", wt.WORD), + ("biCompression", wt.DWORD), ("biSizeImage", wt.DWORD), + ("biXPelsPerMeter", wt.LONG), ("biYPelsPerMeter", wt.LONG), + ("biClrUsed", wt.DWORD), ("biClrImportant", wt.DWORD), + ] + + class BITMAPINFO(ctypes.Structure): + _fields_ = [("bmiHeader", BITMAPINFOHEADER), ("bmiColors", wt.DWORD * 3)] + + bmi = BITMAPINFO() + bmi.bmiHeader.biSize = ctypes.sizeof(BITMAPINFOHEADER) + bmi.bmiHeader.biWidth = sw + bmi.bmiHeader.biHeight = -sh # top-down + bmi.bmiHeader.biPlanes = 1 + bmi.bmiHeader.biBitCount = 32 + bmi.bmiHeader.biCompression = 0 # BI_RGB + + buf = (ctypes.c_ubyte * (sw * sh * 4))() + gdi32.GetDIBits(hdc_mem, hbm, 0, sh, buf, ctypes.byref(bmi), 0) + + gdi32.DeleteObject(hbm) + gdi32.DeleteDC(hdc_mem) + user32.ReleaseDC(0, hdc_screen) + + try: + from PIL import Image # type: ignore + # BGRA → RGB + img = Image.frombuffer("RGBA", (sw, sh), bytes(buf), "raw", "BGRA", 0, 1).convert("RGB") + import io as _io + out = _io.BytesIO() + img.save(out, format="PNG", optimize=True) + return out.getvalue() + except ImportError: + # Without PIL we can't encode PNG; return raw BGRA. Honest failure. + raise RuntimeError("PIL not installed and mss not available; cannot encode screenshot") + + +def _screenshot() -> bytes: + png = _screenshot_mss() + if png is not None: + return png + return _screenshot_bitblt() + + +def _crop_region(png_bytes: bytes, region) -> bytes: + try: + from PIL import Image # type: ignore + except ImportError: + return png_bytes + import io as _io + x1, y1, x2, y2 = (int(v) for v in region) + img = Image.open(_io.BytesIO(png_bytes)).convert("RGB") + cropped = img.crop((x1, y1, x2, y2)) + buf = _io.BytesIO() + cropped.save(buf, format="PNG", optimize=True) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# Handler. +# --------------------------------------------------------------------------- + +def _handle(req: ActionRequest) -> ActionResult: + a = req.action + + if a == "screen_size": + w, h = _screen_size() + return ActionResult(success=True, action=a, screen_width=w, screen_height=h) + + if a == "cursor_position": + x, y = _cursor_position() + return ActionResult(success=True, action=a, cursor_x=x, cursor_y=y) + + if a == "get_active_window": + return ActionResult(success=True, action=a, active_window=_active_window() or {}) + + if a == "screenshot": + png = _screenshot() + if req.region: + png = _crop_region(png, req.region) + if req.redact_regions: + png = redact_image(png, req.redact_regions) + return ActionResult( + success=True, + action=a, + screenshot_b64=base64.b64encode(png).decode("ascii"), + ) + + if a == "wait": + time.sleep((req.ms or 0) / 1000.0) + return ActionResult(success=True, action=a, message=f"waited {req.ms}ms") + + sw, sh = _screen_size() + validate_coords_within(req, sw, sh) + + if a == "left_click": + _click(req.x, req.y, "left") + elif a == "double_click": + _double_click(req.x, req.y) + elif a == "right_click": + _click(req.x, req.y, "right") + elif a == "middle_click": + _click(req.x, req.y, "middle") + elif a == "left_button_press": + _button_event(req.x, req.y, "down") + elif a == "left_button_release": + _button_event(req.x, req.y, "up") + elif a == "mouse_move": + _move(req.x, req.y) + elif a == "mouse_drag": + _drag(req.x, req.y, req.x2, req.y2) + elif a == "scroll": + _move(req.x, req.y) + _scroll(req.direction, req.amount or 3) + elif a == "type": + _type_text(req.text) + elif a == "key": + _key_combo(req.keys) + else: + return ActionResult(success=False, action=a, error=f"unhandled action {a!r}") + + return ActionResult(success=True, action=a, screen_width=sw, screen_height=sh) + + +def computer_use_windows_tool(args: Dict[str, Any]) -> Dict[str, Any]: + try: + gate(args.get("action", "")) + req = parse_request(args) + result = _handle(req) + except SafetyRefusal as e: + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"refused: {e}") + except ValidationError as e: + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"validation: {e}") + except Exception as e: + logger.exception("computer_use_windows handler failed") + result = ActionResult(success=False, action=str(args.get("action", "")), error=f"runtime: {e}") + + log_action(result.action, args, result.success, result.error) + return result.to_dict() + + +registry.register( + name="computer_use_windows", + toolset="computer_use", + schema=build_schema("computer_use_windows", "Windows"), + handler=lambda args, **kw: computer_use_windows_tool(args), + check_fn=_check_windows, + requires_env=["HERMES_COMPUTER_USE_ENABLED"], + is_async=False, + description="Native Windows desktop control (SendInput + mss/BitBlt).", + emoji="🪟", + max_result_size_chars=8_000_000, +)