diff --git a/.gitignore b/.gitignore index 2143926b6..f99525468 100644 --- a/.gitignore +++ b/.gitignore @@ -101,3 +101,6 @@ exp/ .playwright-mcp/ bridge/node_modules/ webui/.verify-* + +# Computer-use E2E run artifacts +tests/e2e/runs/ diff --git a/nanobot/agent/tools/browser_tool.py b/nanobot/agent/tools/browser_tool.py new file mode 100644 index 000000000..f65e935a1 --- /dev/null +++ b/nanobot/agent/tools/browser_tool.py @@ -0,0 +1,312 @@ +"""browser tool: DOM/accessibility-based web automation (model-agnostic actions). + +Unlike the pixel-based ``computer_use`` tool, this drives the page by **element +ref** instead of by screen coordinates. Each action returns a fresh snapshot of +the page's interactive elements (a numbered list), and the model acts by picking +a ``[ref]`` — no pixel grounding required. This makes web *actions* reliable +across ANY tool-calling model (including non-vision models), which pixel-based +computer use cannot do (only computer-use-trained models ground pixels well). + +It reuses the Playwright ``BrowserBackend`` from ``computer_use_backends``. The +heavy ``playwright`` dependency is imported lazily, so importing this module at +tool auto-discovery time stays cheap. +""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import urlparse + +from loguru import logger +from pydantic import Field + +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import ( + BooleanSchema, + IntegerSchema, + StringSchema, + tool_parameters_schema, +) +from nanobot.config.schema import Base +from nanobot.utils.helpers import build_image_content_blocks + +_ACTIONS = [ + "navigate", + "snapshot", + "click", + "type", + "select", + "scroll", + "key", + "back", + "read_text", +] +_ENABLE_WARNED = False + + +class BrowserToolConfig(Base): + """browser (DOM) tool configuration.""" + + enable: bool = False # off by default — opt-in + start_url: str = "about:blank" + headless: bool = True + width: int = 1280 + height: int = 800 + allowed_domains: list[str] = Field(default_factory=list) # empty = all + include_screenshot: bool = False # also attach a screenshot (for vision models) + max_elements: int = 200 # cap interactive elements per snapshot + + +def _format_elements(elements: list[dict]) -> str: + if not elements: + return "Interactive elements: (none found — try scrolling or read_text)" + lines = [] + for e in elements: + tag = e.get("tag", "") + typ = e.get("type") or "" + label = tag + (f"[{typ}]" if typ else "") + line = f"[{e.get('ref')}] {label}" + name = (e.get("name") or "").strip() + if name: + line += f' "{name}"' + href = e.get("href") or "" + if href and tag == "a": + line += f" -> {href[:60]}" + lines.append(line) + return "Interactive elements (act with the [ref] number):\n" + "\n".join(lines) + + +@tool_parameters( + tool_parameters_schema( + action=StringSchema("The action to perform.", enum=_ACTIONS), + ref=IntegerSchema( + description="Element ref number from the latest snapshot (click/type/select).", + nullable=True, + ), + text=StringSchema( + "Text to type (action=type) or key/combo like 'Enter'/'ctrl+a' (action=key).", + nullable=True, + ), + url=StringSchema("URL to open (action=navigate).", nullable=True), + value=StringSchema("Option value/label to choose (action=select).", nullable=True), + submit=BooleanSchema(description="Press Enter after typing (action=type).", nullable=True), + scroll_direction=StringSchema( + "Scroll direction (action=scroll).", enum=["up", "down", "left", "right"], nullable=True + ), + scroll_amount=IntegerSchema(description="Scroll clicks (action=scroll).", nullable=True), + required=["action"], + ) +) +class BrowserTool(Tool): + """Browse and act on web pages by element ref (DOM-based, works with any model).""" + + _scopes = {"core"} + + name = "browser" + description = ( + "Control a web browser by acting on page elements by their [ref] number. " + "Each call returns the current page URL plus a fresh numbered list of the page's " + "interactive elements; pick a [ref] to click/type/select — no pixel coordinates " + "needed. A page may already be open: call 'snapshot' FIRST to see it. Only use " + "'navigate' for a specific URL you were explicitly given — never guess a URL. " + "Move between pages by clicking links/buttons via their [ref]. Use 'read_text' to " + "read page text. Re-read the element list after each action; refs are reassigned." + ) + + config_key = "browser" + + @classmethod + def config_cls(cls) -> type[BrowserToolConfig]: + return BrowserToolConfig + + @classmethod + def enabled(cls, ctx: Any) -> bool: + return bool(ctx.config.browser.enable) + + @classmethod + def create(cls, ctx: Any) -> Tool: + cfg = ctx.config.browser + global _ENABLE_WARNED + if not _ENABLE_WARNED: + logger.warning( + "browser tool is ENABLED — it can navigate and act on web pages. " + "Restrict with allowed_domains, run in a sandbox, and beware prompt " + "injection from page content." + ) + _ENABLE_WARNED = True + return cls( + start_url=cfg.start_url, + headless=cfg.headless, + width=cfg.width, + height=cfg.height, + allowed_domains=list(cfg.allowed_domains), + include_screenshot=cfg.include_screenshot, + max_elements=cfg.max_elements, + ) + + def __init__( + self, + *, + start_url: str = "about:blank", + headless: bool = True, + width: int = 1280, + height: int = 800, + allowed_domains: list[str] | None = None, + include_screenshot: bool = False, + max_elements: int = 200, + backend_impl: Any = None, + ) -> None: + self.start_url = start_url + self.headless = headless + self.width = width + self.height = height + self.allowed_domains = allowed_domains or [] + self.include_screenshot = include_screenshot + self.max_elements = max_elements + self._backend = backend_impl + + @property + def read_only(self) -> bool: + return False + + @property + def exclusive(self) -> bool: + return True + + async def _get_backend(self) -> Any: + if self._backend is not None: + return self._backend + from nanobot.agent.tools.computer_use_backends.browser_playwright import BrowserBackend + self._backend = BrowserBackend( + width=self.width, + height=self.height, + headless=self.headless, + start_url=self.start_url, + ) + return self._backend + + def _domain_allowed(self, url: str) -> bool: + if not self.allowed_domains: + return True + host = (urlparse(url).hostname or "").lower() + if not host: + return False + for dom in self.allowed_domains: + d = dom.lower().lstrip(".") + if d and (host == d or host.endswith("." + d)): + return True + return False + + @staticmethod + def _req_ref(params: dict[str, Any], action: str) -> Any: + ref = params.get("ref") + if ref is None: + raise ValueError(f"action '{action}' requires an element 'ref' from the snapshot") + return ref + + async def _dispatch(self, backend: Any, action: str, p: dict[str, Any]) -> tuple[str, str | None]: + """Return (status, direct_text). If direct_text is set, it is returned as-is + (no snapshot appended).""" + if action == "navigate": + url = p.get("url") + if not url: + raise ValueError("action 'navigate' requires 'url'") + if not self._domain_allowed(str(url)): + raise ValueError(f"navigation to '{url}' is blocked by the allowed_domains policy") + await backend.navigate(str(url)) + return f"Navigated to {url}", None + + if action == "snapshot": + return "Snapshot of the current page", None + + if action == "click": + ref = self._req_ref(p, action) + await backend.click_ref(ref) + return f"Clicked element [{ref}]", None + + if action == "type": + ref = self._req_ref(p, action) + text = p.get("text") + if text is None: + raise ValueError("action 'type' requires 'text'") + submit = bool(p.get("submit")) + await backend.fill_ref(ref, str(text), submit=submit) + return f"Typed into [{ref}]" + (" and pressed Enter" if submit else ""), None + + if action == "select": + ref = self._req_ref(p, action) + value = p.get("value") + if value is None: + raise ValueError("action 'select' requires 'value'") + await backend.select_ref(ref, str(value)) + return f"Selected '{value}' in [{ref}]", None + + if action == "scroll": + direction = str(p.get("scroll_direction") or "down").lower() + if direction not in ("up", "down", "left", "right"): + raise ValueError("'scroll_direction' must be up/down/left/right") + await backend.scroll_page(direction, int(p.get("scroll_amount") or 3)) + return f"Scrolled {direction}", None + + if action == "key": + combo = p.get("text") + if not combo: + raise ValueError("action 'key' requires 'text' (e.g. 'Enter')") + await backend.key(str(combo)) + return f"Pressed {combo}", None + + if action == "back": + await backend.go_back() + return "Navigated back", None + + if action == "read_text": + txt = await backend.read_text() + return "", f"Page text:\n{txt}" + + raise ValueError(f"unknown action '{action}'") + + async def execute(self, action: str | None = None, **kwargs: Any) -> Any: + action = (action or "").strip() + if action not in _ACTIONS: + return f"Error: unknown action '{action}'. Valid actions: {', '.join(_ACTIONS)}" + + try: + backend = await self._get_backend() + except ImportError as exc: + return f"Error: {exc}" + except Exception as exc: + return f"Error: could not initialize browser backend: {type(exc).__name__}: {exc}" + + try: + status, direct = await self._dispatch(backend, action, kwargs) + except ValueError as exc: + return f"Error: {exc}" + except Exception as exc: + return f"Error executing browser '{action}': {type(exc).__name__}: {exc}" + + if direct is not None: + return direct + + try: + elements = await backend.dom_snapshot(self.max_elements) + snapshot = _format_elements(elements) + except Exception as exc: + snapshot = f"(could not read page elements: {type(exc).__name__}: {exc})" + try: + current = await backend.current_url() + except Exception: + current = "" + header = f"{status}\nCurrent page: {current}" if current else status + text_out = f"{header}\n\n{snapshot}" + + if self.include_screenshot: + try: + png = await backend.screenshot() + return build_image_content_blocks(png, "image/png", "", text_out) + except Exception: + return text_out + return text_out + + async def close(self) -> None: + if self._backend is not None: + await self._backend.close() diff --git a/nanobot/agent/tools/computer_use.py b/nanobot/agent/tools/computer_use.py new file mode 100644 index 000000000..088e42a1c --- /dev/null +++ b/nanobot/agent/tools/computer_use.py @@ -0,0 +1,345 @@ +"""computer_use tool: control a desktop or browser via screenshots + mouse/keyboard. + +Model-agnostic by design. The tool returns each screenshot as ``image_url`` +content blocks; the runner delivers those to the model as a follow-up user +message, so any vision + tool-calling model works (Claude, GPT, Gemini, ... via +any gateway such as OpenRouter) without provider-specific plumbing. + +Backends (selected via config) are pluggable: +- ``desktop`` — PyAutoGUI, controls the local GUI (Codex-style). +- ``browser`` — Playwright, controls a headless web page (also supports navigate). + +Both heavy deps are optional and imported lazily, so importing this module at +tool auto-discovery time is cheap and never requires pyautogui/playwright. +""" + +from __future__ import annotations + +import asyncio +import io +from typing import Any +from urllib.parse import urlparse + +from loguru import logger +from pydantic import Field + +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import ( + IntegerSchema, + NumberSchema, + StringSchema, + tool_parameters_schema, +) +from nanobot.config.schema import Base +from nanobot.utils.helpers import build_image_content_blocks +from nanobot.utils.screen_scale import ScreenScaler, fit_target_size + +_ACTIONS = [ + "screenshot", + "left_click", + "right_click", + "middle_click", + "double_click", + "triple_click", + "mouse_move", + "left_click_drag", + "scroll", + "type", + "key", + "wait", + "navigate", +] + +_CLICK_BUTTONS = { + "left_click": "left", + "double_click": "left", + "triple_click": "left", + "right_click": "right", + "middle_click": "middle", +} +_CLICK_COUNTS = {"double_click": 2, "triple_click": 3} + +_MAX_WAIT_S = 10.0 +_ENABLE_WARNED = False # log the security warning at most once per process + + +class ComputerUseToolConfig(Base): + """computer_use tool configuration.""" + + enable: bool = False # off by default — security-sensitive, opt-in + backend: str = "desktop" # "desktop" | "browser" + target_width: int = 1280 # screenshot is downscaled to fit this box (model space) + target_height: int = 800 + require_approval: bool = True # gate destructive actions (enforced by the agent layer) + allowed_domains: list[str] = Field(default_factory=list) # browser allowlist; empty = all + start_url: str = "about:blank" # browser initial page + headless: bool = True # browser headless mode + + +@tool_parameters( + tool_parameters_schema( + action=StringSchema("The action to perform.", enum=_ACTIONS), + x=IntegerSchema( + description="X coordinate in the pixel space of the screenshot you were last shown.", + nullable=True, + ), + y=IntegerSchema( + description="Y coordinate in the pixel space of the screenshot you were last shown.", + nullable=True, + ), + text=StringSchema( + "Text to type (action=type), or a key/combo like 'ctrl+s' or 'Enter' (action=key).", + nullable=True, + ), + scroll_direction=StringSchema( + "Scroll direction (action=scroll).", enum=["up", "down", "left", "right"], nullable=True + ), + scroll_amount=IntegerSchema( + description="Number of scroll clicks (action=scroll).", nullable=True + ), + duration=NumberSchema(description="Seconds to wait (action=wait).", nullable=True), + url=StringSchema("URL to open (action=navigate, browser backend only).", nullable=True), + required=["action"], + ) +) +class ComputerUseTool(Tool): + """Control a computer (desktop or browser) by looking at screenshots and acting.""" + + _scopes = {"core"} # never exposed to subagents — security-sensitive + + name = "computer_use" + description = ( + "Control a computer via screenshots and mouse/keyboard. Each call performs ONE " + "action and returns a fresh screenshot of the resulting screen. Coordinates (x, y) " + "are in the pixel space of the screenshot you were last shown (top-left is 0,0). " + "The 'browser' backend additionally supports the 'navigate' action. Always start " + "with a 'screenshot' to see the screen, then act based on what you observe; after " + "each action re-check the new screenshot before the next step." + ) + + config_key = "computer_use" + + @classmethod + def config_cls(cls) -> type[ComputerUseToolConfig]: + return ComputerUseToolConfig + + @classmethod + def enabled(cls, ctx: Any) -> bool: + return bool(ctx.config.computer_use.enable) + + @classmethod + def create(cls, ctx: Any) -> Tool: + cfg = ctx.config.computer_use + global _ENABLE_WARNED + if not _ENABLE_WARNED: + logger.warning( + "computer_use tool is ENABLED (backend={}). It can control the real " + "{} and affect state outside the workspace. Run nanobot in a sandbox/VM, " + "restrict to trusted models/inputs, and beware prompt injection from " + "on-screen/web content.", + cfg.backend, + "browser" if cfg.backend == "browser" else "desktop", + ) + _ENABLE_WARNED = True + return cls( + backend=cfg.backend, + target_width=cfg.target_width, + target_height=cfg.target_height, + require_approval=cfg.require_approval, + allowed_domains=list(cfg.allowed_domains), + start_url=cfg.start_url, + headless=cfg.headless, + ) + + def __init__( + self, + *, + backend: str = "desktop", + target_width: int = 1280, + target_height: int = 800, + require_approval: bool = True, + allowed_domains: list[str] | None = None, + start_url: str = "about:blank", + headless: bool = True, + backend_impl: Any = None, + ) -> None: + self.backend_name = backend + self.target_width = target_width + self.target_height = target_height + self.require_approval = require_approval + self.allowed_domains = allowed_domains or [] + self.start_url = start_url + self.headless = headless + self._backend = backend_impl # injectable for tests + + @property + def read_only(self) -> bool: + return False + + @property + def exclusive(self) -> bool: + # Stateful single environment; must not run alongside other tools. + return True + + async def _get_backend(self) -> Any: + if self._backend is not None: + return self._backend + if self.backend_name == "browser": + from nanobot.agent.tools.computer_use_backends.browser_playwright import BrowserBackend + self._backend = BrowserBackend( + width=self.target_width, + height=self.target_height, + headless=self.headless, + start_url=self.start_url, + ) + else: + from nanobot.agent.tools.computer_use_backends.desktop_pyautogui import DesktopBackend + self._backend = DesktopBackend() + return self._backend + + @staticmethod + def _downscale_png(png: bytes, target: tuple[int, int]) -> bytes: + try: + from PIL import Image # noqa: PLC0415 + except Exception as exc: + raise ImportError( + "Pillow is required for computer_use. Install: pip install 'nanobot-ai[computer-use]'" + ) from exc + tw, th = target + with Image.open(io.BytesIO(png)) as img: + if (img.width, img.height) == (tw, th): + return png + resized = img.convert("RGB").resize((tw, th)) + out = io.BytesIO() + resized.save(out, format="PNG") + return out.getvalue() + + def _domain_allowed(self, url: str) -> bool: + """Check a navigation URL against the browser allowed_domains policy. + + Empty allowlist = allow all. A domain entry matches the exact host or any + subdomain of it (``example.com`` allows ``app.example.com``). + """ + if not self.allowed_domains: + return True + host = (urlparse(url).hostname or "").lower() + if not host: + return False + for dom in self.allowed_domains: + d = dom.lower().lstrip(".") + if d and (host == d or host.endswith("." + d)): + return True + return False + + async def _dispatch(self, backend: Any, scaler: ScreenScaler, action: str, params: dict[str, Any]) -> str: + def _xy() -> tuple[int, int]: + x, y = params.get("x"), params.get("y") + # Some models emit a combined [x, y] array in the x field (this is + # Anthropic's native ``coordinate`` convention). Accept that too so + # the tool is not tied to one provider's calling style. + if isinstance(x, (list, tuple)) and len(x) == 2 and y is None: + x, y = x[0], x[1] + if x is None or y is None: + raise ValueError(f"action '{action}' requires integer 'x' and 'y'") + return scaler.to_real(int(x), int(y)) + + if action == "screenshot": + return "Took a screenshot" + + if action == "wait": + secs = float(params.get("duration") or 1.0) + secs = max(0.0, min(secs, _MAX_WAIT_S)) + await asyncio.sleep(secs) + return f"Waited {secs:g}s" + + if action in _CLICK_BUTTONS: + rx, ry = _xy() + await backend.click(rx, ry, _CLICK_BUTTONS[action], _CLICK_COUNTS.get(action, 1)) + return f"{action} at ({rx}, {ry})" + + if action == "mouse_move": + rx, ry = _xy() + await backend.move(rx, ry) + return f"Moved to ({rx}, {ry})" + + if action == "left_click_drag": + rx, ry = _xy() + await backend.drag(rx, ry) + return f"Dragged to ({rx}, {ry})" + + if action == "scroll": + rx, ry = _xy() + direction = str(params.get("scroll_direction") or "down").lower() + if direction not in ("up", "down", "left", "right"): + raise ValueError("'scroll_direction' must be up/down/left/right") + amount = int(params.get("scroll_amount") or 3) + await backend.scroll(rx, ry, direction, amount) + return f"Scrolled {direction} by {amount} at ({rx}, {ry})" + + if action == "type": + text = params.get("text") + if not text: + raise ValueError("action 'type' requires 'text'") + await backend.type_text(str(text)) + return f"Typed {len(str(text))} characters" + + if action == "key": + combo = params.get("text") + if not combo: + raise ValueError("action 'key' requires 'text' (e.g. 'ctrl+s')") + await backend.key(str(combo)) + return f"Pressed {combo}" + + if action == "navigate": + url = params.get("url") + if not url: + raise ValueError("action 'navigate' requires 'url'") + if not self._domain_allowed(str(url)): + raise ValueError( + f"navigation to '{url}' is blocked by the allowed_domains policy" + ) + await backend.navigate(str(url)) + return f"Navigated to {url}" + + raise ValueError(f"unknown action '{action}'") + + async def execute(self, action: str | None = None, **kwargs: Any) -> Any: + action = (action or "").strip() + if action not in _ACTIONS: + return f"Error: unknown action '{action}'. Valid actions: {', '.join(_ACTIONS)}" + + try: + backend = await self._get_backend() + real_w, real_h = await backend.dimensions() + except ImportError as exc: + return f"Error: {exc}" + except Exception as exc: + return f"Error: could not initialize computer_use backend: {type(exc).__name__}: {exc}" + + scaler = ScreenScaler.for_screen(real_w, real_h, self.target_width, self.target_height) + + try: + status = await self._dispatch(backend, scaler, action, kwargs) + except ValueError as exc: + return f"Error: {exc}" + except NotImplementedError as exc: + return f"Error: {exc}" + except Exception as exc: + return f"Error executing computer_use '{action}': {type(exc).__name__}: {exc}" + + # Return a fresh screenshot so the model sees the result of its action. + try: + png = await backend.screenshot() + target = fit_target_size(real_w, real_h, self.target_width, self.target_height) + png = self._downscale_png(png, target) + except ImportError as exc: + return f"Error: {exc}" + except Exception as exc: + return f"{status}\n(Could not capture screenshot: {type(exc).__name__}: {exc})" + + label = f"{status} | screen {target[0]}x{target[1]} ({backend.environment})" + return build_image_content_blocks(png, "image/png", "", label) + + async def close(self) -> None: + if self._backend is not None: + await self._backend.close() diff --git a/nanobot/agent/tools/computer_use_backends/__init__.py b/nanobot/agent/tools/computer_use_backends/__init__.py new file mode 100644 index 000000000..3b2fe8608 --- /dev/null +++ b/nanobot/agent/tools/computer_use_backends/__init__.py @@ -0,0 +1,11 @@ +"""Pluggable execution backends for the ``computer_use`` tool. + +Backends are plain classes (NOT ``Tool`` subclasses), so ``ToolLoader`` never +mistakes them for tools. Importing this package must stay cheap: the heavy, +optional dependencies (pyautogui / Pillow / playwright) are imported lazily +inside the concrete backend modules and never at import time here. +""" + +from nanobot.agent.tools.computer_use_backends.base import ComputerBackend + +__all__ = ["ComputerBackend"] diff --git a/nanobot/agent/tools/computer_use_backends/base.py b/nanobot/agent/tools/computer_use_backends/base.py new file mode 100644 index 000000000..d237562d5 --- /dev/null +++ b/nanobot/agent/tools/computer_use_backends/base.py @@ -0,0 +1,65 @@ +"""Backend interface for the ``computer_use`` tool. + +A backend is the *actuator* + *screenshot source* for one execution environment +(the local desktop, a headless browser, a VM, ...). The tool layer owns the +agent loop, coordinate scaling, screenshot downscaling and safety gating; a +backend only has to perform primitive actions and grab a screenshot. + +Coordinate contract: every ``x``/``y`` passed to a backend is already in **real +device pixels** (the same pixel space as :meth:`screenshot`). The tool scales the +model's target-space coordinates to real pixels before calling the backend, so +backends never deal with the downscaled space. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class ComputerBackend(ABC): + """Primitive GUI actions + screenshot for one execution environment.""" + + #: "desktop" or "browser" — surfaced to the model so it knows the context. + environment: str = "desktop" + + @abstractmethod + async def dimensions(self) -> tuple[int, int]: + """Return the real screenshot pixel size as ``(width, height)``.""" + + @abstractmethod + async def screenshot(self) -> bytes: + """Return a PNG screenshot of the current screen at real pixel size.""" + + @abstractmethod + async def click(self, x: int, y: int, button: str = "left", count: int = 1) -> None: + """Click at ``(x, y)``. ``button`` in {left,right,middle}; ``count`` for double/triple.""" + + @abstractmethod + async def move(self, x: int, y: int) -> None: + """Move the cursor to ``(x, y)`` without clicking.""" + + @abstractmethod + async def drag(self, x: int, y: int) -> None: + """Press at the current cursor position and drag to ``(x, y)``, then release.""" + + @abstractmethod + async def scroll(self, x: int, y: int, direction: str, amount: int) -> None: + """Scroll at ``(x, y)``. ``direction`` in {up,down,left,right}; ``amount`` in clicks.""" + + @abstractmethod + async def type_text(self, text: str) -> None: + """Type ``text`` at the current focus.""" + + @abstractmethod + async def key(self, combo: str) -> None: + """Press a key or combo, e.g. ``"ctrl+s"`` / ``"Enter"`` (backend-specific syntax).""" + + async def navigate(self, url: str) -> None: + """Navigate to ``url`` (browser backends only).""" + raise NotImplementedError( + f"'navigate' is not supported by the {self.environment} backend" + ) + + async def close(self) -> None: + """Release any resources (browser process, etc.). Safe to call repeatedly.""" + return None diff --git a/nanobot/agent/tools/computer_use_backends/browser_playwright.py b/nanobot/agent/tools/computer_use_backends/browser_playwright.py new file mode 100644 index 000000000..2470675df --- /dev/null +++ b/nanobot/agent/tools/computer_use_backends/browser_playwright.py @@ -0,0 +1,241 @@ +"""Browser backend using Playwright (DOM-aware, deterministic, sandbox-friendly). + +Drives a headless Chromium page via pixel coordinates + screenshots, mirroring +the desktop backend's surface so the same model/tool loop works against the web. +The viewport uses ``deviceScaleFactor=1`` so screenshot pixels == mouse +coordinates (no HiDPI scaling to undo). The browser/page persist across actions +within one tool instance so navigation and state carry between turns. +""" + +from __future__ import annotations + +from typing import Any + +from nanobot.agent.tools.computer_use_backends.base import ComputerBackend + +_MISSING = ( + "Browser computer-use backend needs 'playwright'. Install with: " + "pip install 'nanobot-ai[computer-use]' && playwright install chromium" +) + +_SCROLL_PIXELS = 100 # one "scroll click" ~= this many pixels + +# Tags visible interactive elements with data-nanobot-ref and returns a compact +# list. Refs are reassigned per call. Used by DOM/accessibility mode. +_SNAPSHOT_JS = r""" +(max) => { + const SEL = 'a,button,input,textarea,select,[role=button],[role=link],[role=checkbox],[role=radio],[role=tab],[role=menuitem],[role=switch],[onclick],[contenteditable=""],[contenteditable=true]'; + const out = []; + let ref = 0; + for (const el of document.querySelectorAll(SEL)) { + const r = el.getBoundingClientRect(); + const s = getComputedStyle(el); + if (r.width <= 0 || r.height <= 0) continue; + if (s.visibility === 'hidden' || s.display === 'none' || s.opacity === '0') continue; + ref++; + el.setAttribute('data-nanobot-ref', String(ref)); + let name = (el.getAttribute('aria-label') || el.innerText || el.value || + el.getAttribute('placeholder') || el.getAttribute('name') || + el.getAttribute('title') || ''); + name = name.replace(/\s+/g, ' ').trim().slice(0, 120); + out.push({ + ref: ref, + tag: el.tagName.toLowerCase(), + role: el.getAttribute('role') || '', + type: el.getAttribute('type') || '', + name: name, + href: el.getAttribute('href') || '' + }); + if (out.length >= max) break; + } + return out; +} +""" + +# CUA/xdotool-ish modifier names -> Playwright modifiers. +_MODIFIERS = { + "ctrl": "Control", "control": "Control", + "alt": "Alt", "option": "Alt", + "shift": "Shift", + "cmd": "Meta", "meta": "Meta", "super": "Meta", "win": "Meta", +} +# Common single-key names -> Playwright key names. +_KEYS = { + "return": "Enter", "enter": "Enter", "tab": "Tab", "esc": "Escape", + "escape": "Escape", "backspace": "Backspace", "delete": "Delete", + "space": "Space", "up": "ArrowUp", "down": "ArrowDown", + "left": "ArrowLeft", "right": "ArrowRight", + "page_down": "PageDown", "pagedown": "PageDown", + "page_up": "PageUp", "pageup": "PageUp", "home": "Home", "end": "End", +} + + +def _playwright_key(combo: str) -> str: + parts = [p.strip() for p in combo.split("+") if p.strip()] + out: list[str] = [] + for part in parts: + low = part.lower() + if low in _MODIFIERS: + out.append(_MODIFIERS[low]) + elif low in _KEYS: + out.append(_KEYS[low]) + elif len(part) == 1: + out.append(part) + else: + out.append(part.capitalize()) + return "+".join(out) + + +class BrowserBackend(ComputerBackend): + environment = "browser" + + def __init__( + self, + *, + width: int = 1280, + height: int = 800, + headless: bool = True, + start_url: str = "about:blank", + ) -> None: + self._width = width + self._height = height + self._headless = headless + self._start_url = start_url + self._pw: Any = None + self._browser: Any = None + self._page: Any = None + self._last_pos = (0, 0) + + async def _ensure(self) -> Any: + if self._page is not None: + return self._page + try: + from playwright.async_api import async_playwright # noqa: PLC0415 + except Exception as exc: + raise ImportError(_MISSING) from exc + self._pw = await async_playwright().start() + self._browser = await self._pw.chromium.launch(headless=self._headless) + context = await self._browser.new_context( + viewport={"width": self._width, "height": self._height}, + device_scale_factor=1, + ) + self._page = await context.new_page() + if self._start_url and self._start_url != "about:blank": + await self._page.goto(self._start_url) + return self._page + + async def dimensions(self) -> tuple[int, int]: + await self._ensure() + vp = self._page.viewport_size or {"width": self._width, "height": self._height} + return vp["width"], vp["height"] + + async def screenshot(self) -> bytes: + page = await self._ensure() + return await page.screenshot() + + async def click(self, x: int, y: int, button: str = "left", count: int = 1) -> None: + page = await self._ensure() + await page.mouse.click(x, y, button=button, click_count=count) + self._last_pos = (x, y) + + async def move(self, x: int, y: int) -> None: + page = await self._ensure() + await page.mouse.move(x, y) + self._last_pos = (x, y) + + async def drag(self, x: int, y: int) -> None: + page = await self._ensure() + sx, sy = self._last_pos + await page.mouse.move(sx, sy) + await page.mouse.down() + await page.mouse.move(x, y) + await page.mouse.up() + self._last_pos = (x, y) + + async def scroll(self, x: int, y: int, direction: str, amount: int) -> None: + page = await self._ensure() + await page.mouse.move(x, y) + pixels = max(1, amount) * _SCROLL_PIXELS + dx = pixels if direction == "right" else -pixels if direction == "left" else 0 + dy = pixels if direction == "down" else -pixels if direction == "up" else 0 + await page.mouse.wheel(dx, dy) + + async def type_text(self, text: str) -> None: + page = await self._ensure() + await page.keyboard.type(text) + + async def key(self, combo: str) -> None: + page = await self._ensure() + key = _playwright_key(combo) + if key: + await page.keyboard.press(key) + + async def navigate(self, url: str) -> None: + page = await self._ensure() + await page.goto(url) + self._last_pos = (0, 0) + + # --- DOM / accessibility mode (act by element ref, not pixels) --- + + async def dom_snapshot(self, max_elements: int = 200) -> list[dict]: + """Tag visible interactive elements with ``data-nanobot-ref`` and return them. + + Each entry: ``{ref, tag, role, type, name, href}``. Refs are reassigned on + every snapshot, so callers should act on the latest snapshot. + """ + page = await self._ensure() + return await page.evaluate(_SNAPSHOT_JS, max_elements) + + def _ref_selector(self, ref) -> str: + return f'[data-nanobot-ref="{int(ref)}"]' + + async def click_ref(self, ref) -> None: + page = await self._ensure() + await page.click(self._ref_selector(ref), timeout=5000) + + async def fill_ref(self, ref, text: str, submit: bool = False) -> None: + page = await self._ensure() + sel = self._ref_selector(ref) + await page.fill(sel, text, timeout=5000) + if submit: + await page.press(sel, "Enter") + + async def select_ref(self, ref, value: str) -> None: + page = await self._ensure() + sel = self._ref_selector(ref) + try: + await page.select_option(sel, value, timeout=3000) + except Exception: + # Models usually pass the visible label, not the option value. + await page.select_option(sel, label=value, timeout=3000) + + async def scroll_page(self, direction: str, amount: int) -> None: + page = await self._ensure() + pixels = max(1, amount) * _SCROLL_PIXELS + dx = pixels if direction == "right" else -pixels if direction == "left" else 0 + dy = pixels if direction == "down" else -pixels if direction == "up" else 0 + await page.evaluate("([x, y]) => window.scrollBy(x, y)", [dx, dy]) + + async def go_back(self) -> None: + page = await self._ensure() + await page.go_back() + + async def read_text(self, max_chars: int = 4000) -> str: + page = await self._ensure() + txt = await page.evaluate("() => document.body ? document.body.innerText : ''") + return (txt or "")[:max_chars] + + async def current_url(self) -> str: + page = await self._ensure() + return page.url + + async def close(self) -> None: + try: + if self._browser is not None: + await self._browser.close() + finally: + if self._pw is not None: + await self._pw.stop() + self._browser = None + self._page = None + self._pw = None diff --git a/nanobot/agent/tools/computer_use_backends/desktop_pyautogui.py b/nanobot/agent/tools/computer_use_backends/desktop_pyautogui.py new file mode 100644 index 000000000..dee752eb9 --- /dev/null +++ b/nanobot/agent/tools/computer_use_backends/desktop_pyautogui.py @@ -0,0 +1,139 @@ +"""Desktop GUI backend using PyAutoGUI (the Codex-style "control the real machine"). + +Drives the local desktop: screenshots via PyAutoGUI/Pillow, mouse + keyboard via +PyAutoGUI. Works on macOS (needs Screen Recording + Accessibility permissions), +Windows, and Linux/X11 (incl. a headless Xvfb display, which is how the e2e +sandbox runs it). + +Retina / HiDPI note: on macOS the screenshot is in *physical* pixels (e.g. 2x) +while PyAutoGUI's mouse API uses *logical* points. We compute the ratio from the +screenshot size vs ``pyautogui.size()`` and convert real (screenshot-pixel) +coordinates to logical points before actuating. +""" + +from __future__ import annotations + +import asyncio +import io +from typing import Any + +from nanobot.agent.tools.computer_use_backends.base import ComputerBackend + +_MISSING = ( + "Desktop computer-use backend needs 'pyautogui' and 'pillow'. " + "Install with: pip install 'nanobot-ai[computer-use]'" +) + +# xdotool/CUA-style key names -> PyAutoGUI key names. +_KEY_ALIASES = { + "return": "enter", + "ctrl": "ctrl", + "control": "ctrl", + "cmd": "command", + "super": "win", + "win": "win", + "page_down": "pagedown", + "page_up": "pageup", + "pagedown": "pagedown", + "pageup": "pageup", + "esc": "esc", + "escape": "esc", +} + +_SCROLL_CLICK_PIXELS = 100 # one "scroll click" ~= this many pixels + + +class DesktopBackend(ComputerBackend): + environment = "desktop" + + def __init__(self) -> None: + self._pg: Any = None + self._image_mod: Any = None + self._ratio_x = 1.0 + self._ratio_y = 1.0 + self._dims: tuple[int, int] | None = None + + def _ensure(self) -> Any: + if self._pg is not None: + return self._pg + try: + import pyautogui # noqa: PLC0415 + from PIL import Image # noqa: PLC0415 + except Exception as exc: # ImportError, or platform display errors + raise ImportError(_MISSING) from exc + pyautogui.FAILSAFE = False + self._pg = pyautogui + self._image_mod = Image + return pyautogui + + def _grab_png_and_size(self) -> tuple[bytes, int, int]: + pg = self._ensure() + img = pg.screenshot() + buf = io.BytesIO() + img.save(buf, format="PNG") + width, height = img.size + # Refresh logical<->physical ratio from the actual grab. + try: + logical_w, logical_h = pg.size() + self._ratio_x = (logical_w / width) if width else 1.0 + self._ratio_y = (logical_h / height) if height else 1.0 + except Exception: + self._ratio_x = self._ratio_y = 1.0 + self._dims = (width, height) + return buf.getvalue(), width, height + + def _to_logical(self, x: int, y: int) -> tuple[int, int]: + return round(x * self._ratio_x), round(y * self._ratio_y) + + async def dimensions(self) -> tuple[int, int]: + if self._dims is not None: + return self._dims + _, w, h = await asyncio.to_thread(self._grab_png_and_size) + return w, h + + async def screenshot(self) -> bytes: + png, _, _ = await asyncio.to_thread(self._grab_png_and_size) + return png + + async def click(self, x: int, y: int, button: str = "left", count: int = 1) -> None: + pg = self._ensure() + lx, ly = self._to_logical(x, y) + await asyncio.to_thread(pg.click, lx, ly, count, 0.0, button) + + async def move(self, x: int, y: int) -> None: + pg = self._ensure() + lx, ly = self._to_logical(x, y) + await asyncio.to_thread(pg.moveTo, lx, ly) + + async def drag(self, x: int, y: int) -> None: + pg = self._ensure() + lx, ly = self._to_logical(x, y) + await asyncio.to_thread(pg.dragTo, lx, ly, 0.3, pg.easeInOutQuad, False, "left") + + async def scroll(self, x: int, y: int, direction: str, amount: int) -> None: + pg = self._ensure() + lx, ly = self._to_logical(x, y) + clicks = max(1, amount) * _SCROLL_CLICK_PIXELS + await asyncio.to_thread(pg.moveTo, lx, ly) + if direction in ("up", "down"): + await asyncio.to_thread(pg.scroll, clicks if direction == "up" else -clicks) + else: + await asyncio.to_thread(pg.hscroll, clicks if direction == "right" else -clicks) + + async def type_text(self, text: str) -> None: + pg = self._ensure() + await asyncio.to_thread(pg.typewrite, text, 0.01) + + async def key(self, combo: str) -> None: + pg = self._ensure() + keys = [ + _KEY_ALIASES.get(part.strip().lower(), part.strip().lower()) + for part in combo.split("+") + if part.strip() + ] + if not keys: + return + if len(keys) == 1: + await asyncio.to_thread(pg.press, keys[0]) + else: + await asyncio.to_thread(pg.hotkey, *keys) diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py index e997fea30..e26653189 100644 --- a/nanobot/config/schema.py +++ b/nanobot/config/schema.py @@ -12,7 +12,9 @@ from nanobot.config_base import Base from nanobot.cron.types import CronSchedule if TYPE_CHECKING: + from nanobot.agent.tools.browser_tool import BrowserToolConfig from nanobot.agent.tools.cli_apps import CliAppsToolConfig + from nanobot.agent.tools.computer_use import ComputerUseToolConfig from nanobot.agent.tools.filesystem import FileToolsConfig from nanobot.agent.tools.image_generation import ImageGenerationToolConfig from nanobot.agent.tools.self import MyToolConfig @@ -399,6 +401,16 @@ class ToolsConfig(Base): """ web: WebToolsConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.web", "WebToolsConfig")) + browser: BrowserToolConfig = Field( + default_factory=lambda: _lazy_default( + "nanobot.agent.tools.browser_tool", "BrowserToolConfig" + ) + ) + computer_use: ComputerUseToolConfig = Field( + default_factory=lambda: _lazy_default( + "nanobot.agent.tools.computer_use", "ComputerUseToolConfig" + ) + ) exec: ExecToolConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.shell", "ExecToolConfig")) file: FileToolsConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.filesystem", "FileToolsConfig")) cli_apps: CliAppsToolConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.cli_apps", "CliAppsToolConfig")) @@ -670,7 +682,9 @@ def _resolve_tool_config_refs() -> None: """ import sys + from nanobot.agent.tools.browser_tool import BrowserToolConfig from nanobot.agent.tools.cli_apps import CliAppsToolConfig + from nanobot.agent.tools.computer_use import ComputerUseToolConfig from nanobot.agent.tools.filesystem import FileToolsConfig from nanobot.agent.tools.image_generation import ImageGenerationToolConfig from nanobot.agent.tools.self import MyToolConfig @@ -680,6 +694,8 @@ def _resolve_tool_config_refs() -> None: # Re-export into this module's namespace mod = sys.modules[__name__] mod.ExecToolConfig = ExecToolConfig # type: ignore[attr-defined] + mod.BrowserToolConfig = BrowserToolConfig # type: ignore[attr-defined] + mod.ComputerUseToolConfig = ComputerUseToolConfig # type: ignore[attr-defined] mod.FileToolsConfig = FileToolsConfig # type: ignore[attr-defined] mod.CliAppsToolConfig = CliAppsToolConfig # type: ignore[attr-defined] mod.WebToolsConfig = WebToolsConfig # type: ignore[attr-defined] diff --git a/nanobot/utils/screen_scale.py b/nanobot/utils/screen_scale.py new file mode 100644 index 000000000..02d7dff2e --- /dev/null +++ b/nanobot/utils/screen_scale.py @@ -0,0 +1,72 @@ +"""Coordinate scaling between the screenshot the model sees and the real device. + +The computer-use loop sends the model a screenshot that is downscaled to a +target size (smaller screenshots cost fewer tokens, are faster, and match the +resolutions vision models are tuned for). The model then replies with click / +move coordinates *in that downscaled space*. Those must be scaled back to the +real device's pixel space before we actuate mouse events — getting this wrong is +the single most common cause of "the clicks miss" bugs. + +This module is pure (no I/O, no optional deps) so it is cheap to import at tool +auto-discovery time and trivial to unit test. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +def fit_target_size( + real_width: int, + real_height: int, + max_width: int, + max_height: int, +) -> tuple[int, int]: + """Target size fitting within ``(max_width, max_height)``, preserving aspect. + + Never upscales: if the real screen is already smaller than the max, the real + size is returned unchanged. + """ + if real_width <= 0 or real_height <= 0: + return max(1, max_width), max(1, max_height) + if max_width <= 0 or max_height <= 0: + return real_width, real_height + scale = min(max_width / real_width, max_height / real_height, 1.0) + return max(1, round(real_width * scale)), max(1, round(real_height * scale)) + + +@dataclass(frozen=True) +class ScreenScaler: + """Maps model (target-space) coordinates onto real device pixels.""" + + real_width: int + real_height: int + target_width: int + target_height: int + + @property + def scale_x(self) -> float: + return self.real_width / self.target_width if self.target_width else 1.0 + + @property + def scale_y(self) -> float: + return self.real_height / self.target_height if self.target_height else 1.0 + + def to_real(self, x: float, y: float) -> tuple[int, int]: + """Scale a model-space ``(x, y)`` to a real device pixel, clamped in-bounds.""" + rx = round(x * self.scale_x) + ry = round(y * self.scale_y) + rx = max(0, min(rx, max(0, self.real_width - 1))) + ry = max(0, min(ry, max(0, self.real_height - 1))) + return rx, ry + + @classmethod + def for_screen( + cls, + real_width: int, + real_height: int, + max_width: int, + max_height: int, + ) -> ScreenScaler: + tw, th = fit_target_size(real_width, real_height, max_width, max_height) + return cls(real_width, real_height, tw, th) diff --git a/pyproject.toml b/pyproject.toml index eb7c10187..75b4f4243 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,6 +88,11 @@ pdf = [ olostep = [ "olostep>=0.1.0; python_version < '3.14'", ] +computer-use = [ + "pyautogui>=0.9.54", + "pillow>=10.0.0", + "playwright>=1.40.0", +] dev = [ "pytest>=9.0.0,<10.0.0", "pytest-asyncio>=1.3.0,<2.0.0", diff --git a/tests/e2e/Dockerfile b/tests/e2e/Dockerfile new file mode 100644 index 000000000..a118c7908 --- /dev/null +++ b/tests/e2e/Dockerfile @@ -0,0 +1,40 @@ +# Sandbox for computer_use end-to-end tests. +# +# Provides BOTH backends in an isolated Linux container: +# - browser : headless Chromium via Playwright (no display needed) +# - desktop : PyAutoGUI driving a virtual X11 display (Xvfb) — Codex-style, +# but contained so it never touches a real machine. +# +# Build: docker build -f tests/e2e/Dockerfile -t nanobot-cu-e2e . +# Run: docker run --rm -e OPENROUTER_API_KEY=sk-or-... -e COMPUTER_USE_E2E=1 \ +# nanobot-cu-e2e +# +# The desktop backend needs a display; the entrypoint starts Xvfb on :99 and +# exports DISPLAY so pyautogui/scrot work headlessly. +FROM python:3.11-slim + +ENV DEBIAN_FRONTEND=noninteractive \ + DISPLAY=:99 \ + PYTHONUNBUFFERED=1 + +# X11 virtual display + screenshot/input tooling for the desktop backend, +# plus the system libs Chromium needs. +RUN apt-get update && apt-get install -y --no-install-recommends \ + xvfb x11-utils scrot xauth \ + libxtst6 libxrandr2 libxss1 libnss3 libasound2 libgbm1 libgtk-3-0 \ + fonts-liberation ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app +COPY . /app + +RUN pip install --no-cache-dir -e ".[computer-use,dev]" \ + && python -m playwright install --with-deps chromium + +# Start Xvfb, wait for it, then run the e2e suite. +RUN printf '#!/bin/sh\nset -e\nXvfb :99 -screen 0 1280x800x24 >/tmp/xvfb.log 2>&1 &\nfor i in $(seq 1 30); do xdpyinfo -display :99 >/dev/null 2>&1 && break; sleep 0.3; done\nexec "$@"\n' > /usr/local/bin/with-xvfb \ + && chmod +x /usr/local/bin/with-xvfb + +ENV COMPUTER_USE_E2E=1 +ENTRYPOINT ["with-xvfb"] +CMD ["pytest", "tests/e2e/test_computer_use_e2e.py", "-v", "-s"] diff --git a/tests/e2e/README.md b/tests/e2e/README.md new file mode 100644 index 000000000..7ec673031 --- /dev/null +++ b/tests/e2e/README.md @@ -0,0 +1,124 @@ +# computer_use end-to-end tests + +Real vision models (via **OpenRouter**) drive the real backends through the full +agent loop. Gated by env vars, excluded from default CI. + +## What it proves + +- Model-agnostic screenshot delivery: every model goes through one + OpenAI-compatible endpoint (OpenRouter); screenshots come back as follow-up + **user** messages, so no provider-specific path is exercised. +- The `computer_use` tool + backends actually see the screen and act + (read a number, click a button, type into a field). + +## Requirements + +```bash +pip install 'nanobot-ai[computer-use]' +playwright install chromium # for the browser backend +``` + +Models must be **vision + tool-calling** capable on OpenRouter. + +## Run the browser e2e locally (no Docker, headless) + +```bash +COMPUTER_USE_E2E=1 OPENROUTER_API_KEY=sk-or-... \ + pytest tests/e2e/test_computer_use_e2e.py -v -s +``` + +Pick models explicitly (default: claude-sonnet-4.5 / gpt-4o / gemini-2.5-pro): + +```bash +COMPUTER_USE_E2E_MODELS="anthropic/claude-sonnet-4.5,openai/gpt-4o" \ +COMPUTER_USE_E2E=1 OPENROUTER_API_KEY=sk-or-... \ + pytest tests/e2e/test_computer_use_e2e.py -v -s +``` + +Vision models are non-deterministic; each scenario retries +`COMPUTER_USE_E2E_ATTEMPTS` (default 2) times and passes if any attempt succeeds. + +## Run in the sandbox (Docker; also covers the desktop/pyautogui backend) + +```bash +docker build -f tests/e2e/Dockerfile -t nanobot-cu-e2e . +docker run --rm -e OPENROUTER_API_KEY=sk-or-... -e COMPUTER_USE_E2E=1 nanobot-cu-e2e +``` + +The container starts Xvfb on `:99` so the desktop backend (PyAutoGUI + scrot) +works headlessly and contained — it never touches a real machine. + +## Scenarios (`model × scenario` matrix) + +| Test | What it checks | +|------|----------------| +| `test_read_screen` | reads the rendered number `42` from a screenshot (perception) | +| `test_click_button` | clicks **Submit**; asserts `#status == "SUBMITTED"` via the DOM (click accuracy) | +| `test_type_and_submit` | types a name + clicks **Greet**; asserts the greeting (multi-step loop) | + +## Findings (first real run, June 2026, via OpenRouter) + +Matrix of 5 model families × 3 scenarios: + +| Model | read_screen | click_button | type_and_submit | +|-------|:----------:|:-----------:|:--------------:| +| anthropic/claude-sonnet-4.5 | ✅ | ✅ | ✅ | +| openai/gpt-4o | ✅ | ❌ | ❌ | +| google/gemini-2.5-pro | ✅ | ❌ | ❌ | +| qwen/qwen3-vl-235b-a22b-instruct | ✅ | ❌ | ❌ | +| x-ai/grok-4.3 | ✅ | ❌ | ❌ | + +**Takeaway:** the model-agnostic *plumbing* works for everyone — all 5 models +received the screenshot (delivered as a user message) and read the on-screen +number. But **pixel-precise clicking only worked with the computer-use-trained +model (Claude)**. Diagnostics showed gpt-4o clicking a round-number guess +`(100,100)` for a button at `(37,156)`, and qwen3-vl emitting an `[x,y]` array +(now supported) but still mis-grounding the y coordinate. This matches the +field: general/GUI VLMs perceive well but ground pixel coordinates poorly. + +**Implication for broad model support:** for the *browser* backend, a +DOM/accessibility-based interaction mode (click an element by ref/description +instead of by pixel — how browser-use / Playwright-MCP work) would make action +reliable across *any* tool-capable model, including non-vision ones. That is the +recommended next step beyond this pixel-based v1. + +## Browser DOM mode (model-agnostic) — recommended for the web + +`tests/e2e/test_browser_dom_e2e.py` and `tests/e2e/test_browser_complex_e2e.py` +exercise the **`browser`** tool, which acts by element **ref** (DOM/accessibility +snapshot) instead of pixels. This removes the pixel-grounding limitation above: +the same models that miss pixel clicks succeed here. + +DOM matrix (simple scenarios) — the models that FAILED pixel clicking: + +| Model | DOM click | DOM type+submit | read_text (no vision) | +|-------|:--------:|:--------------:|:--------------------:| +| openai/gpt-4o | ✅ | ✅ | flaky | +| google/gemini-2.5-pro | ✅ | flaky | ✅ | +| qwen/qwen3-vl-235b | ✅ | ✅ | ✅ | + +Click went from **0/3 in pixel mode to 3/3 in DOM mode.** Because DOM mode needs +no pixel grounding (run with `include_screenshot=false`), it works with ANY +tool-calling model — including non-vision ones. + +**Complex flows** (`test_browser_complex_e2e.py`) drive a small local web app +(served over HTTP so localStorage + navigation work), pages in `tests/e2e/pages/`: + +- multi-field form: text inputs + `

+

+

+ +

+
+ Plan + + + +
+

+ + +
+ + diff --git a/tests/e2e/pages/list.html b/tests/e2e/pages/list.html new file mode 100644 index 000000000..bfa70d88f --- /dev/null +++ b/tests/e2e/pages/list.html @@ -0,0 +1,24 @@ + +Long list + +

Pick a row

+
none
+ + + diff --git a/tests/e2e/pages/login.html b/tests/e2e/pages/login.html new file mode 100644 index 000000000..530cb1ab9 --- /dev/null +++ b/tests/e2e/pages/login.html @@ -0,0 +1,24 @@ + +Login + +

Sign in

+
+

+

+ +
+
+ + diff --git a/tests/e2e/pages/shop.html b/tests/e2e/pages/shop.html new file mode 100644 index 000000000..2883a7027 --- /dev/null +++ b/tests/e2e/pages/shop.html @@ -0,0 +1,23 @@ + +Shop + +

Gadget Shop

+
Cart items: 0
+ +

View cart

+ + diff --git a/tests/e2e/test_browser_complex_e2e.py b/tests/e2e/test_browser_complex_e2e.py new file mode 100644 index 000000000..83364a730 --- /dev/null +++ b/tests/e2e/test_browser_complex_e2e.py @@ -0,0 +1,194 @@ +"""Complex, multi-step DOM-browser e2e against a small local web app. + +Pages live in ``tests/e2e/pages/`` and are served over a local HTTP server (so +localStorage and cross-page navigation work). Scenarios exercise realistic +flows: a multi-field form (text + select + radio + checkbox), an add-to-cart + +checkout flow across pages, picking one row out of 60, and a login → dashboard +redirect. All run in DOM mode (act by element ref), so they are model-agnostic. + +Run: + pip install 'nanobot-ai[computer-use]' && playwright install chromium + COMPUTER_USE_E2E=1 OPENROUTER_API_KEY=sk-or-... \ + pytest tests/e2e/test_browser_complex_e2e.py -v -s +""" + +from __future__ import annotations + +import functools +import http.server +import os +import socketserver +import threading +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from nanobot.agent.runner import AgentRunner, AgentRunSpec +from nanobot.agent.tools.browser_tool import BrowserTool +from nanobot.agent.tools.registry import ToolRegistry +from nanobot.config.schema import AgentDefaults + +pytestmark = pytest.mark.skipif( + not (os.getenv("COMPUTER_USE_E2E") and os.getenv("OPENROUTER_API_KEY")), + reason="set COMPUTER_USE_E2E=1 and OPENROUTER_API_KEY to run complex browser e2e tests", +) + +_DEFAULT_MODELS = [ + "openai/gpt-5.1", + "anthropic/claude-sonnet-4.5", +] +_MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars +_PAGES = Path(__file__).parent / "pages" + + +def _models() -> list[str]: + env = os.getenv("COMPUTER_USE_E2E_MODELS", "").strip() + return [m.strip() for m in env.split(",") if m.strip()] or _DEFAULT_MODELS + + +def _attempts() -> int: + try: + return max(1, int(os.getenv("COMPUTER_USE_E2E_ATTEMPTS", "2"))) + except ValueError: + return 2 + + +def _make_provider(model: str): + from nanobot.providers.openai_compat_provider import OpenAICompatProvider + + provider = OpenAICompatProvider( + api_key=os.environ["OPENROUTER_API_KEY"], + api_base="https://openrouter.ai/api/v1", + default_model=model, + ) + provider.generation = SimpleNamespace(max_tokens=4096, temperature=0.0, reasoning_effort=None) + return provider + + +class _QuietHandler(http.server.SimpleHTTPRequestHandler): + def log_message(self, *args): # silence per-request stderr logging + pass + + +@pytest.fixture(scope="module") +def base_url() -> str: + handler = functools.partial(_QuietHandler, directory=str(_PAGES)) + httpd = socketserver.TCPServer(("127.0.0.1", 0), handler) + port = httpd.server_address[1] + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + yield f"http://127.0.0.1:{port}" + httpd.shutdown() + httpd.server_close() + + +def _tool(start_url: str) -> BrowserTool: + return BrowserTool(start_url=start_url, headless=True, include_screenshot=False, max_elements=120) + + +async def _run(provider, tool: BrowserTool, objective: str, max_iterations: int = 24): + tools = ToolRegistry() + tools.register(tool) + return await AgentRunner(provider).run(AgentRunSpec( + initial_messages=[{"role": "user", "content": objective}], + tools=tools, + model=provider.get_default_model(), + max_iterations=max_iterations, + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + )) + + +async def _eval(tool: BrowserTool, expr: str): + backend = await tool._get_backend() + return await backend._page.evaluate(expr) + + +async def _attempt(model, start_page, objective, check, base_url, max_iterations=24): + """Run up to N attempts; check(tool)->bool decides success. Returns bool.""" + for _ in range(_attempts()): + tool = _tool(f"{base_url}/{start_page}") + try: + await _run(_make_provider(model), tool, objective, max_iterations) + if await check(tool): + return True + finally: + await tool.close() + return False + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_multifield_form(model: str, base_url: str): + """Fill a form with text inputs, a + +
+""" + + +def _models() -> list[str]: + env = os.getenv("COMPUTER_USE_E2E_MODELS", "").strip() + return [m.strip() for m in env.split(",") if m.strip()] or _DEFAULT_MODELS + + +def _make_provider(model: str): + from nanobot.providers.openai_compat_provider import OpenAICompatProvider + + provider = OpenAICompatProvider( + api_key=os.environ["OPENROUTER_API_KEY"], + api_base="https://openrouter.ai/api/v1", + default_model=model, + ) + provider.generation = SimpleNamespace(max_tokens=2048, temperature=0.0, reasoning_effort=None) + return provider + + +@pytest.fixture(scope="module") +def page_url() -> str: + fd, path = tempfile.mkstemp(suffix=".html", prefix="nanobot_dom_") + Path(path).write_text(_TEST_PAGE, encoding="utf-8") + os.close(fd) + yield f"file://{path}" + try: + os.unlink(path) + except OSError: + pass + + +def _tool(start_url: str) -> BrowserTool: + # No screenshot — pure DOM, to prove vision is not required. + return BrowserTool(start_url=start_url, headless=True, include_screenshot=False) + + +async def _run(provider, tool: BrowserTool, objective: str, max_iterations: int = 12): + tools = ToolRegistry() + tools.register(tool) + return await AgentRunner(provider).run(AgentRunSpec( + initial_messages=[{"role": "user", "content": objective}], + tools=tools, + model=provider.get_default_model(), + max_iterations=max_iterations, + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + )) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_dom_click_button(model: str, page_url: str): + attempts = max(1, int(os.getenv("COMPUTER_USE_E2E_ATTEMPTS", "2"))) + ok = False + for _ in range(attempts): + tool = _tool(page_url) + try: + await _run( + _make_provider(model), + tool, + "The page is already open. Use the browser tool (start with action=snapshot) " + "to find and click the 'Submit' button.", + ) + backend = await tool._get_backend() + status = await backend._page.evaluate("document.getElementById('status').textContent") + if status == "SUBMITTED": + ok = True + break + finally: + await tool.close() + assert ok, f"[{model}] DOM click should set #status to SUBMITTED" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_dom_type_and_submit(model: str, page_url: str): + attempts = max(1, int(os.getenv("COMPUTER_USE_E2E_ATTEMPTS", "2"))) + ok = False + for _ in range(attempts): + tool = _tool(page_url) + try: + await _run( + _make_provider(model), + tool, + "The page is already open. Use the browser tool: snapshot the page, type " + "'Ada' into the name input, then click the 'Greet' button.", + max_iterations=16, + ) + backend = await tool._get_backend() + greeting = await backend._page.evaluate( + "document.getElementById('greeting').textContent" + ) + if greeting and "ADA" in greeting.upper(): + ok = True + break + finally: + await tool.close() + assert ok, f"[{model}] DOM type+greet should produce a greeting with the typed name" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_dom_read_text_no_vision(model: str, page_url: str): + """Pure text path: the model reads the page via read_text (no screenshot at all).""" + attempts = max(1, int(os.getenv("COMPUTER_USE_E2E_ATTEMPTS", "2"))) + ok = False + for _ in range(attempts): + tool = _tool(page_url) + try: + result = await _run( + _make_provider(model), + tool, + "Use the browser tool's read_text action to read the page, then tell me the " + "number shown. Reply with just the number.", + ) + if result.final_content and "42" in result.final_content: + ok = True + break + finally: + await tool.close() + assert ok, f"[{model}] should read '42' via read_text (no vision)" diff --git a/tests/e2e/test_computer_use_e2e.py b/tests/e2e/test_computer_use_e2e.py new file mode 100644 index 000000000..616220151 --- /dev/null +++ b/tests/e2e/test_computer_use_e2e.py @@ -0,0 +1,205 @@ +"""End-to-end computer_use tests: REAL vision models (via OpenRouter) driving the +REAL Playwright browser backend through the full AgentRunner loop. + +These are gated and excluded from normal CI. To run: + + pip install 'nanobot-ai[computer-use]' + playwright install chromium + COMPUTER_USE_E2E=1 OPENROUTER_API_KEY=sk-or-... \ + pytest tests/e2e/test_computer_use_e2e.py -v -s + +Model selection (must be vision + tool-calling capable on OpenRouter): + COMPUTER_USE_E2E_MODELS="anthropic/claude-sonnet-4.5,openai/gpt-4o,google/gemini-2.5-pro" + +Vision models are not deterministic, so each scenario is allowed a few attempts +(COMPUTER_USE_E2E_ATTEMPTS, default 2) and passes if any attempt succeeds. + +NOTE: This harness is model-agnostic by design — every model goes through the +same OpenAI-compatible OpenRouter endpoint, and screenshots are delivered as +follow-up user messages (see runner._split_tool_result_media), so no +provider-specific code path is exercised. +""" + +from __future__ import annotations + +import os +import tempfile +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from nanobot.agent.runner import AgentRunner, AgentRunSpec +from nanobot.agent.tools.computer_use import ComputerUseTool +from nanobot.agent.tools.registry import ToolRegistry +from nanobot.config.schema import AgentDefaults + +pytestmark = pytest.mark.skipif( + not (os.getenv("COMPUTER_USE_E2E") and os.getenv("OPENROUTER_API_KEY")), + reason="set COMPUTER_USE_E2E=1 and OPENROUTER_API_KEY to run computer_use e2e tests", +) + +# Default to a computer-use-trained model that reliably grounds pixel +# coordinates. Pass COMPUTER_USE_E2E_MODELS to test others — but note that +# general/GUI VLMs reliably PERCEIVE the screen (test_read_screen) yet usually +# miss pixel-precise clicks (see tests/e2e/README.md "Findings"). +_DEFAULT_MODELS = [ + "anthropic/claude-sonnet-4.5", +] +_MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars + +_TEST_PAGE = """ +nanobot cu e2e + + +
42
+ +
idle
+
+ + +
+ +""" + + +def _models() -> list[str]: + env = os.getenv("COMPUTER_USE_E2E_MODELS", "").strip() + return [m.strip() for m in env.split(",") if m.strip()] or _DEFAULT_MODELS + + +def _attempts() -> int: + try: + return max(1, int(os.getenv("COMPUTER_USE_E2E_ATTEMPTS", "2"))) + except ValueError: + return 2 + + +def _make_provider(model: str): + from nanobot.providers.openai_compat_provider import OpenAICompatProvider + + provider = OpenAICompatProvider( + api_key=os.environ["OPENROUTER_API_KEY"], + api_base="https://openrouter.ai/api/v1", + default_model=model, + ) + # The runner reads provider.generation for sampling params (mirrors factory). + provider.generation = SimpleNamespace(max_tokens=2048, temperature=0.0, reasoning_effort=None) + return provider + + +@pytest.fixture(scope="module") +def page_url() -> str: + fd, path = tempfile.mkstemp(suffix=".html", prefix="nanobot_cu_") + Path(path).write_text(_TEST_PAGE, encoding="utf-8") + os.close(fd) + yield f"file://{path}" + try: + os.unlink(path) + except OSError: + pass + + +async def _run(provider, tool: ComputerUseTool, objective: str, max_iterations: int = 12): + tools = ToolRegistry() + tools.register(tool) + runner = AgentRunner(provider) + return await runner.run(AgentRunSpec( + initial_messages=[{"role": "user", "content": objective}], + tools=tools, + model=provider.get_default_model(), + max_iterations=max_iterations, + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + )) + + +def _browser_tool(start_url: str) -> ComputerUseTool: + return ComputerUseTool( + backend="browser", + target_width=1280, + target_height=800, + start_url=start_url, + headless=True, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_read_screen(model: str, page_url: str): + """The model must read the number rendered on the page from a screenshot.""" + attempts = _attempts() + ok = False + for _ in range(attempts): + tool = _browser_tool(page_url) + try: + result = await _run( + _make_provider(model), + tool, + "Use the computer_use tool: take a screenshot of the page and tell me " + "the big number shown. Reply with just the number.", + ) + if result.final_content and "42" in result.final_content: + ok = True + break + finally: + await tool.close() + assert ok, f"[{model}] expected the model to read '42' from the screenshot" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_click_button(model: str, page_url: str): + """The model must locate and click the Submit button (asserted via the DOM).""" + attempts = _attempts() + ok = False + for _ in range(attempts): + tool = _browser_tool(page_url) + try: + await _run( + _make_provider(model), + tool, + "Use the computer_use tool to click the 'Submit' button on the page. " + "Take a screenshot first to locate it.", + ) + backend = await tool._get_backend() + status = await backend._page.evaluate("document.getElementById('status').textContent") + if status == "SUBMITTED": + ok = True + break + finally: + await tool.close() + assert ok, f"[{model}] expected #status to become 'SUBMITTED' after clicking Submit" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model", _models()) +async def test_type_and_submit(model: str, page_url: str): + """The model must type into a field and click a button (multi-step).""" + attempts = _attempts() + ok = False + for _ in range(attempts): + tool = _browser_tool(page_url) + try: + await _run( + _make_provider(model), + tool, + "Use the computer_use tool: click the name input, type 'Ada', then click " + "the 'Greet' button. Take screenshots to guide yourself.", + max_iterations=16, + ) + backend = await tool._get_backend() + greeting = await backend._page.evaluate( + "document.getElementById('greeting').textContent" + ) + if greeting and "ADA" in greeting.upper(): + ok = True + break + finally: + await tool.close() + assert ok, f"[{model}] expected greeting to contain the typed name" diff --git a/tests/tools/test_browser_tool.py b/tests/tools/test_browser_tool.py new file mode 100644 index 000000000..fc5f15e18 --- /dev/null +++ b/tests/tools/test_browser_tool.py @@ -0,0 +1,193 @@ +"""Tests for the DOM-based browser tool (model-agnostic web actions). + +Backend is a duck-typed fake, so no playwright is needed (only Pillow for the +optional screenshot path).""" + +from __future__ import annotations + +import io +from unittest.mock import MagicMock + +import pytest + +from nanobot.agent.tools.browser_tool import BrowserTool, BrowserToolConfig + + +class _FakeDomBackend: + environment = "browser" + + def __init__(self): + self.calls: list[tuple] = [] + self.elements = [ + {"ref": 1, "tag": "button", "role": "", "type": "", "name": "Submit", "href": ""}, + {"ref": 2, "tag": "input", "role": "", "type": "text", "name": "your name", "href": ""}, + ] + + async def navigate(self, url): + self.calls.append(("navigate", url)) + + async def dom_snapshot(self, max_elements=200): + return self.elements + + async def click_ref(self, ref): + self.calls.append(("click", ref)) + + async def fill_ref(self, ref, text, submit=False): + self.calls.append(("fill", ref, text, submit)) + + async def select_ref(self, ref, value): + self.calls.append(("select", ref, value)) + + async def scroll_page(self, direction, amount): + self.calls.append(("scroll", direction, amount)) + + async def key(self, combo): + self.calls.append(("key", combo)) + + async def go_back(self): + self.calls.append(("back",)) + + async def read_text(self, max_chars=4000): + return "the number is 42" + + async def current_url(self): + return "http://test.local/page" + + async def screenshot(self): + from PIL import Image + buf = io.BytesIO() + Image.new("RGB", (1280, 800), (0, 0, 0)).save(buf, format="PNG") + return buf.getvalue() + + async def close(self): + self.calls.append(("close",)) + + +def _tool(**kw): + fb = _FakeDomBackend() + return BrowserTool(backend_impl=fb, **kw), fb + + +class TestConfigAndMetadata: + def test_defaults_off(self): + cfg = BrowserToolConfig() + assert cfg.enable is False + assert cfg.headless is True + assert cfg.include_screenshot is False + assert cfg.max_elements == 200 + + def test_enabled_reads_config(self): + ctx = MagicMock() + ctx.config.browser.enable = True + assert BrowserTool.enabled(ctx) is True + ctx.config.browser.enable = False + assert BrowserTool.enabled(ctx) is False + + def test_create_from_ctx(self): + ctx = MagicMock() + ctx.config.browser = BrowserToolConfig(enable=True, allowed_domains=["example.com"]) + tool = BrowserTool.create(ctx) + assert isinstance(tool, BrowserTool) + assert tool.allowed_domains == ["example.com"] + + def test_metadata(self): + tool, _ = _tool() + assert tool.name == "browser" + assert tool.exclusive is True + assert tool.read_only is False + assert "subagent" not in tool._scopes + + def test_schema_actions(self): + tool, _ = _tool() + enum = tool.parameters["properties"]["action"]["enum"] + for a in ("navigate", "snapshot", "click", "type", "read_text"): + assert a in enum + + +class TestDispatch: + @pytest.mark.asyncio + async def test_navigate_returns_snapshot(self): + tool, fb = _tool() + result = await tool.execute(action="navigate", url="https://example.com") + assert ("navigate", "https://example.com") in fb.calls + assert isinstance(result, str) + assert "Navigated to https://example.com" in result + # snapshot of interactive elements is appended + assert '[1] button "Submit"' in result + assert '[2] input[text] "your name"' in result + + @pytest.mark.asyncio + async def test_click_by_ref(self): + tool, fb = _tool() + result = await tool.execute(action="click", ref=1) + assert ("click", 1) in fb.calls + assert "Clicked element [1]" in result + # fresh snapshot returned so refs stay current + assert "Interactive elements" in result + + @pytest.mark.asyncio + async def test_type_with_submit(self): + tool, fb = _tool() + await tool.execute(action="type", ref=2, text="Ada", submit=True) + assert ("fill", 2, "Ada", True) in fb.calls + + @pytest.mark.asyncio + async def test_select(self): + tool, fb = _tool() + await tool.execute(action="select", ref=2, value="opt1") + assert ("select", 2, "opt1") in fb.calls + + @pytest.mark.asyncio + async def test_scroll_and_key_and_back(self): + tool, fb = _tool() + await tool.execute(action="scroll", scroll_direction="down", scroll_amount=4) + await tool.execute(action="key", text="Enter") + await tool.execute(action="back") + assert ("scroll", "down", 4) in fb.calls + assert ("key", "Enter") in fb.calls + assert ("back",) in fb.calls + + @pytest.mark.asyncio + async def test_read_text_returns_text_no_snapshot(self): + tool, _ = _tool() + result = await tool.execute(action="read_text") + assert isinstance(result, str) + assert "the number is 42" in result + assert "Interactive elements" not in result + + @pytest.mark.asyncio + async def test_include_screenshot_returns_blocks(self): + tool, _ = _tool(include_screenshot=True) + result = await tool.execute(action="click", ref=1) + assert isinstance(result, list) + imgs = [b for b in result if b.get("type") == "image_url"] + texts = [b for b in result if b.get("type") == "text"] + assert imgs and texts + assert "Clicked element [1]" in texts[-1]["text"] + + +class TestErrorsAndPolicy: + @pytest.mark.asyncio + async def test_unknown_action(self): + tool, _ = _tool() + result = await tool.execute(action="teleport") + assert isinstance(result, str) and "unknown action" in result + + @pytest.mark.asyncio + async def test_click_requires_ref(self): + tool, _ = _tool() + result = await tool.execute(action="click") + assert isinstance(result, str) and "requires an element 'ref'" in result + + @pytest.mark.asyncio + async def test_navigate_blocked_by_allowlist(self): + tool, fb = _tool(allowed_domains=["example.com"]) + result = await tool.execute(action="navigate", url="https://evil.test/") + assert "blocked by the allowed_domains" in result + assert not any(c[0] == "navigate" for c in fb.calls) + + @pytest.mark.asyncio + async def test_navigate_allowed_subdomain(self): + tool, fb = _tool(allowed_domains=["example.com"]) + await tool.execute(action="navigate", url="https://app.example.com/x") + assert ("navigate", "https://app.example.com/x") in fb.calls diff --git a/tests/tools/test_computer_use_tool.py b/tests/tools/test_computer_use_tool.py new file mode 100644 index 000000000..760781010 --- /dev/null +++ b/tests/tools/test_computer_use_tool.py @@ -0,0 +1,276 @@ +"""Tests for the computer_use tool, its config wiring, and coordinate scaling. + +Backend actuation is exercised through an injected fake backend, so these tests +need neither pyautogui nor playwright (only Pillow, for screenshot downscaling). +""" + +from __future__ import annotations + +import io +from unittest.mock import MagicMock + +import pytest + +from nanobot.agent.tools.computer_use import ComputerUseTool, ComputerUseToolConfig +from nanobot.agent.tools.computer_use_backends.base import ComputerBackend +from nanobot.utils.screen_scale import ScreenScaler, fit_target_size + + +class _FakeBackend(ComputerBackend): + """Records actuation calls and serves a solid-colour PNG of a fixed size.""" + + environment = "desktop" + + def __init__(self, width: int = 2560, height: int = 1600): + self.calls: list[tuple] = [] + self._w, self._h = width, height + + async def dimensions(self) -> tuple[int, int]: + return (self._w, self._h) + + async def screenshot(self) -> bytes: + from PIL import Image + img = Image.new("RGB", (self._w, self._h), (10, 20, 30)) + buf = io.BytesIO() + img.save(buf, format="PNG") + return buf.getvalue() + + async def click(self, x, y, button="left", count=1): + self.calls.append(("click", x, y, button, count)) + + async def move(self, x, y): + self.calls.append(("move", x, y)) + + async def drag(self, x, y): + self.calls.append(("drag", x, y)) + + async def scroll(self, x, y, direction, amount): + self.calls.append(("scroll", x, y, direction, amount)) + + async def type_text(self, text): + self.calls.append(("type", text)) + + async def key(self, combo): + self.calls.append(("key", combo)) + # navigate() inherited -> raises NotImplementedError (desktop has no navigate) + + +def _split(result): + assert isinstance(result, list), f"expected content blocks, got {result!r}" + images = [b for b in result if isinstance(b, dict) and b.get("type") == "image_url"] + texts = [b for b in result if isinstance(b, dict) and b.get("type") == "text"] + return images, texts + + +def _tool(**kw): + fb = _FakeBackend(width=kw.pop("w", 2560), height=kw.pop("h", 1600)) + tool = ComputerUseTool(backend_impl=fb, target_width=1280, target_height=800, **kw) + return tool, fb + + +# --------------------------- coordinate scaling (pure) --------------------------- + +class TestScreenScale: + def test_fit_never_upscales(self): + assert fit_target_size(800, 600, 1280, 800) == (800, 600) + + def test_fit_preserves_aspect(self): + # 2560x1600 (16:10) into 1280x800 box -> exactly halved. + assert fit_target_size(2560, 1600, 1280, 800) == (1280, 800) + + def test_fit_landscape_into_box(self): + # 3000x1000 into 1280x800 -> width-bound: scale 1280/3000. + w, h = fit_target_size(3000, 1000, 1280, 800) + assert w == 1280 and h == round(1000 * 1280 / 3000) + + def test_to_real_scales_up(self): + sc = ScreenScaler.for_screen(2560, 1600, 1280, 800) + assert sc.to_real(100, 50) == (200, 100) + + def test_to_real_clamps_in_bounds(self): + sc = ScreenScaler.for_screen(1000, 1000, 1000, 1000) + assert sc.to_real(5000, 5000) == (999, 999) + assert sc.to_real(-10, -10) == (0, 0) + + +# --------------------------- config + metadata --------------------------- + +class TestConfigAndMetadata: + def test_defaults_off(self): + cfg = ComputerUseToolConfig() + assert cfg.enable is False + assert cfg.backend == "desktop" + assert (cfg.target_width, cfg.target_height) == (1280, 800) + assert cfg.require_approval is True + + def test_enabled_reads_config(self): + ctx = MagicMock() + ctx.config.computer_use.enable = True + assert ComputerUseTool.enabled(ctx) is True + ctx.config.computer_use.enable = False + assert ComputerUseTool.enabled(ctx) is False + + def test_create_from_ctx(self): + ctx = MagicMock() + ctx.config.computer_use = ComputerUseToolConfig( + enable=True, backend="browser", target_width=1024, target_height=768 + ) + tool = ComputerUseTool.create(ctx) + assert isinstance(tool, ComputerUseTool) + assert tool.backend_name == "browser" + assert (tool.target_width, tool.target_height) == (1024, 768) + + def test_tool_metadata(self): + tool, _ = _tool() + assert tool.name == "computer_use" + assert tool.exclusive is True + assert tool.read_only is False + assert tool.concurrency_safe is False + # not exposed to subagents + assert "subagent" not in tool._scopes + + def test_schema_has_action_enum(self): + tool, _ = _tool() + action = tool.parameters["properties"]["action"] + assert "screenshot" in action["enum"] + assert "left_click" in action["enum"] + assert tool.parameters["required"] == ["action"] + + +# --------------------------- execute dispatch --------------------------- + +class TestExecute: + @pytest.mark.asyncio + async def test_screenshot_returns_image_blocks(self): + tool, fb = _tool() + result = await tool.execute(action="screenshot") + images, texts = _split(result) + assert len(images) == 1 + assert images[0]["image_url"]["url"].startswith("data:image/png;base64,") + assert "1280x800" in texts[-1]["text"] + assert fb.calls == [] # screenshot performs no actuation + + @pytest.mark.asyncio + async def test_left_click_scales_coordinates(self): + tool, fb = _tool() # real 2560x1600 -> target 1280x800 (2x) + result = await tool.execute(action="left_click", x=100, y=50) + assert fb.calls == [("click", 200, 100, "left", 1)] + _, texts = _split(result) + assert "left_click at (200, 100)" in texts[-1]["text"] + + @pytest.mark.asyncio + async def test_click_accepts_coordinate_array(self): + # Some models emit a combined [x, y] array (Anthropic's native convention). + tool, fb = _tool() + await tool.execute(action="left_click", x=[100, 50]) + assert fb.calls == [("click", 200, 100, "left", 1)] + + @pytest.mark.asyncio + async def test_double_and_triple_click_counts(self): + tool, fb = _tool() + await tool.execute(action="double_click", x=10, y=10) + await tool.execute(action="triple_click", x=10, y=10) + assert fb.calls[0] == ("click", 20, 20, "left", 2) + assert fb.calls[1] == ("click", 20, 20, "left", 3) + + @pytest.mark.asyncio + async def test_right_and_middle_click_buttons(self): + tool, fb = _tool() + await tool.execute(action="right_click", x=5, y=5) + await tool.execute(action="middle_click", x=5, y=5) + assert fb.calls[0][3] == "right" + assert fb.calls[1][3] == "middle" + + @pytest.mark.asyncio + async def test_scroll_defaults_and_args(self): + tool, fb = _tool() + await tool.execute(action="scroll", x=100, y=100, scroll_direction="down", scroll_amount=5) + assert fb.calls == [("scroll", 200, 200, "down", 5)] + + @pytest.mark.asyncio + async def test_type_and_key(self): + tool, fb = _tool() + await tool.execute(action="type", text="hello") + await tool.execute(action="key", text="ctrl+s") + assert ("type", "hello") in fb.calls + assert ("key", "ctrl+s") in fb.calls + + @pytest.mark.asyncio + async def test_drag_and_move(self): + tool, fb = _tool() + await tool.execute(action="mouse_move", x=10, y=10) + await tool.execute(action="left_click_drag", x=20, y=30) + assert ("move", 20, 20) in fb.calls + assert ("drag", 40, 60) in fb.calls + + @pytest.mark.asyncio + async def test_wait(self): + tool, fb = _tool() + result = await tool.execute(action="wait", duration=0.0) + _, texts = _split(result) + assert "Waited" in texts[-1]["text"] + + # ---- error paths return a plain string (model can self-correct) ---- + + @pytest.mark.asyncio + async def test_unknown_action_errors(self): + tool, _ = _tool() + result = await tool.execute(action="frobnicate") + assert isinstance(result, str) and "unknown action" in result + + @pytest.mark.asyncio + async def test_click_requires_coordinates(self): + tool, _ = _tool() + result = await tool.execute(action="left_click") + assert isinstance(result, str) and "requires" in result + + @pytest.mark.asyncio + async def test_navigate_unsupported_on_desktop(self): + tool, _ = _tool() + result = await tool.execute(action="navigate", url="https://example.com") + assert isinstance(result, str) and "Error" in result + + +class _BrowserFakeBackend(_FakeBackend): + environment = "browser" + + async def navigate(self, url): + self.calls.append(("navigate", url)) + + +class TestAllowedDomainsPolicy: + def _browser_tool(self, allowed): + fb = _BrowserFakeBackend(width=1280, height=800) + tool = ComputerUseTool( + backend_impl=fb, + backend="browser", + target_width=1280, + target_height=800, + allowed_domains=allowed, + ) + return tool, fb + + @pytest.mark.asyncio + async def test_empty_allowlist_allows_all(self): + tool, fb = self._browser_tool([]) + await tool.execute(action="navigate", url="https://anything.example/") + assert ("navigate", "https://anything.example/") in fb.calls + + @pytest.mark.asyncio + async def test_exact_domain_allowed(self): + tool, fb = self._browser_tool(["example.com"]) + await tool.execute(action="navigate", url="https://example.com/path") + assert any(c[0] == "navigate" for c in fb.calls) + + @pytest.mark.asyncio + async def test_subdomain_allowed(self): + tool, fb = self._browser_tool(["example.com"]) + await tool.execute(action="navigate", url="https://app.example.com/") + assert any(c[0] == "navigate" for c in fb.calls) + + @pytest.mark.asyncio + async def test_disallowed_domain_blocked(self): + tool, fb = self._browser_tool(["example.com"]) + result = await tool.execute(action="navigate", url="https://evil.test/") + assert isinstance(result, str) and "blocked by the allowed_domains" in result + assert not any(c[0] == "navigate" for c in fb.calls)