diff --git a/code_puppy_core_plugins/browser_harness/README.md b/code_puppy_core_plugins/browser_harness/README.md new file mode 100644 index 0000000..39cc975 --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/README.md @@ -0,0 +1,84 @@ +# Browser Harness + +This opt-in plugin lets Code Puppy drive **your real browser** — your logins, +cookies, extensions, and tabs — through +[browser-harness](https://github.com/browser-use/browser-harness), a small CDP +harness that keeps one tab attached across calls. It is a thin, honest wrapper: +Code Puppy shells out to the `browser-harness` CLI, which owns the browser +connection. + +## Two installs, on purpose + +1. The plugin itself ships with `code-puppy-core-plugins`; nothing to do. +2. The harness is a separate tool, because it owns its own daemon and upgrades: + +```bash +uv tool install --python 3.12 --upgrade --force browser-harness +``` + +`/browser status` tells you which of the two is missing. + +## Consent and configuration + +Browser control is off by default; its tools are not exposed to the model until +you opt in, exactly like macOS Computer Use. + +```text +/browser status # install state, consent, endpoints, browsers +/browser enable | disable # persisted one-time consent +/browser doctor # harness health, with the exact fix +/browser connect # pin http://127.0.0.1:9222 or a wss:// URL +/browser disconnect # back to auto-discovery +/browser install # commands to install a drivable browser +/browser recordings [on|off] # the harness's own local trace recording +``` + +Disabling removes the tools again and blocks further calls. `/browser enable` +merges the tools into the running session, so no restart is needed. + +## Tools + +- `browser_harness(script, browser_name=None, timeout=120)` — run Python with + the harness helpers (`new_tab`, `js`, `click_at_xy`, `cdp`, `wait_for_load`, + …) pre-imported. stdout is the only channel back. +- `browser_screenshot(full=False, max_dim=1568)` — capture the attached tab. It + renders inline in Ghostty, Kitty, WezTerm, and iTerm2, and the same PNG rides + along on the tool result for multimodal models. +- `browser_doctor()` — connection health with the fix that clears it. + +The bundled `SKILL.md` teaches the workflow: one tab per task, accessibility +tree before pixels, verify each action, stop for passwords and purchases. + +## Which browsers work + +browser-harness speaks the Chrome DevTools Protocol, so it needs a +Chromium-family browser: **Chrome, Chrome Canary, Chromium, Brave, Edge, Arc, +and Helium** are detected and drivable. + +**Firefox and Safari cannot be driven.** Firefox implements WebDriver BiDi and +Safari implements the Apple WebKit inspector protocol; neither exposes the CDP +endpoint the harness requires. `/browser status` lists them as *present but not +drivable* rather than pretending they are usable. Options: + +- install a Chromium-family browser (`/browser install`), or +- point at any CDP endpoint elsewhere: `/browser connect http://host:9222` for a + Chromium started with `--remote-debugging-port=9222`, or a hosted browser. + +macOS shows a per-connection "Allow remote debugging?" sheet, and Chrome's +`chrome://inspect/#remote-debugging` toggle must be ticked once. `browser-harness +mac-approve` clears the sheet; `browser_doctor()` tells you when either is the +blocker. + +## Privacy + +Nothing here phones home beyond browser-harness's own optional telemetry +(`browser-harness telemetry disable`). Code Puppy stores only two settings — the +consent flag and an optional endpoint — in its config directory. Page content, +screenshots, and any recordings stay on your machine under the harness's state +directory. + +## Test + +```bash +uv run pytest tests/test_browser_harness_*.py -q --no-cov +``` diff --git a/code_puppy_core_plugins/browser_harness/SKILL.md b/code_puppy_core_plugins/browser_harness/SKILL.md new file mode 100644 index 0000000..d810295 --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/SKILL.md @@ -0,0 +1,87 @@ +--- +name: browser-harness +description: Use for any web task that needs the user's real browser - logged-in + sessions, their cookies, clicks, uploads, downloads, forms, or local web apps. + Public read-only pages need no browser. +version: "1.0" +author: code-puppy +tags: + - web + - browser-automation + - cdp +--- + +# Browser Harness + +Drive Mike's actual browser through `browser-harness`, a CDP harness that keeps +one tab attached across calls. Adapted from +[browser-use/browser-harness](https://github.com/browser-use/browser-harness) +(MIT); the upstream `SKILL.md` (`browser-harness skill`) is the canonical, +version-matched reference and goes deeper than this file. + +## Pick the right browser tool + +- Plain HTTP gets it (public page, API, docs): use `curl`/`web_fetch`. Do not + open a tab. +- Sandboxed scraping, crawling, or parallel extraction: delegate to the + `web-retriever` agent, which drives a throwaway Playwright browser. +- Anything needing *this* browser - existing logins, cookies, local + `localhost` apps, extensions, uploads, downloads, or "watch me do it": + use the `browser_harness` tool. + +## Tools + +- `browser_harness(script)` - run Python with the harness helpers pre-imported. + stdout is the only channel back, so `print()` what you need. +- `browser_screenshot()` - capture the attached tab; it renders inline for the + user. Pixels never tell you what is clickable, so confirm with `page_info()` + or `js()`. +- `browser_doctor()` - install/daemon/browser health, with the exact fix. + +## Workflow + +1. First navigation of a task: `new_tab(url)`. The attached tab survives across + calls, so do **not** call `new_tab()` again in every script. Check + `current_tab()` / `list_tabs()` and `switch_tab()` before opening duplicates; + never close a tab you did not create. +2. After navigation call `wait_for_load()`; after a click that triggers a + request, `wait_for_network_idle()`. +3. Find elements in the accessibility tree, not pixels: + `cdp("Accessibility.getFullAXTree")["nodes"]` carries role, name, and + `backendDOMNodeId`. Filter it in Python - it is thousands of nodes. Then box + center -> `click_at_xy(x, y)` -> verify with a targeted `js()` or + `page_info()` check. +4. Fall back to `js(...)` for DOM/extract work, and screenshots only when + layout or imagery is the question. +5. An action that does nothing usually means the attached tab is hidden: call + `activate_tab(current_tab())`, retry the same action once, then re-check. + This visibly switches tabs, so skip it if the user asked you not to touch + their foreground. +6. Write the reusable part into `$BH_AGENT_WORKSPACE/agent_helpers.py` when a + site-specific trick took real discovery; keep task code in the tool call. + +## Which browser + +The harness speaks CDP, so Chrome, Chromium, Brave, Edge, Arc, or Helium work. +**Firefox and Safari cannot** - they expose no CDP endpoint. If asked for +Firefox, say so and offer a Chromium-family browser or an explicit endpoint. + +- `/browser status` lists what is installed, running, and reachable. +- `/browser connect http://127.0.0.1:9222` (or a `wss://` URL) pins a specific + endpoint; leave it unset to auto-discover the running Chromium browser. +- Cloud browsers: `browser_harness(script='start_remote_daemon("name")')`, then + pass `browser_name="name"` on every later call. Ask before leaving one + running, and stop it with `stop_remote_daemon("name")`. + +## Ask first + +Stop and ask before typing passwords, solving MFA, approving a payment or +purchase, deleting an account, sending a message, or anything that posts +content as the user. Being already signed in is not consent to act. + +## When it will not connect + +`browser_doctor()` (or `/browser doctor`) names the fix. The usual three: +Chrome's `chrome://inspect/#remote-debugging` toggle is off; macOS is waiting on +the "Allow remote debugging?" sheet (`browser-harness mac-approve`); or no +Chromium browser is running at all. diff --git a/code_puppy_core_plugins/browser_harness/__init__.py b/code_puppy_core_plugins/browser_harness/__init__.py new file mode 100644 index 0000000..ea4299e --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/__init__.py @@ -0,0 +1 @@ +"""Opt-in browser control through browser-harness.""" diff --git a/code_puppy_core_plugins/browser_harness/browser.py b/code_puppy_core_plugins/browser_harness/browser.py new file mode 100644 index 0000000..f1aacab --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/browser.py @@ -0,0 +1,248 @@ +"""Discover local browsers, running state, and reachable CDP endpoints. + +browser-harness drives browsers over the Chrome DevTools Protocol, so only the +Chromium family is drivable. Firefox and Safari do not expose a CDP endpoint; +they are reported as present-but-undrivable instead of being silently ignored. +""" + +from __future__ import annotations + +import json +import os +import platform +import shutil +import subprocess +import urllib.request +from dataclasses import dataclass +from pathlib import Path + +#: Ports browser-harness probes for a local DevTools endpoint. +DEFAULT_CDP_PORTS = (9222, 9223) +_PROBE_TIMEOUT_SECONDS = 0.4 +_MAC_APPLICATION_DIRS = (Path("/Applications"), Path.home() / "Applications") +_WINDOWS_ROOTS = tuple( + Path(root) + for root in ( + os.environ.get("PROGRAMFILES", r"C:\Program Files"), + os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)"), + ) + if root +) + +UNSUPPORTED_NOTE = ( + "browser-harness speaks the Chrome DevTools Protocol. Firefox and Safari do " + "not expose one, so they cannot be driven. Any Chromium-family browser " + "(Chrome, Chromium, Brave, Edge, Arc, Helium) can be." +) + + +@dataclass(frozen=True) +class _Spec: + """Where one browser lives on each platform, plus the names it runs as.""" + + name: str + drivable: bool + mac: str + linux: tuple[str, ...] = () + windows: str | None = None + brew: str | None = None + + def process_names(self) -> set[str]: + """Executable names this browser shows up as in a process list.""" + names = {Path(self.mac).name.casefold()} + names.update(Path(binary).name.casefold() for binary in self.linux) + if self.windows: + names.add(Path(self.windows.replace("\\", "/")).name.casefold()) + return names + + +_SPECS: tuple[_Spec, ...] = ( + _Spec( + "Chrome", + True, + "Google Chrome.app/Contents/MacOS/Google Chrome", + ("google-chrome", "google-chrome-stable"), + "Google\\Chrome\\Application\\chrome.exe", + "google-chrome", + ), + _Spec( + "Chrome Canary", + True, + "Google Chrome Canary.app/Contents/MacOS/Google Chrome Canary", + ), + _Spec( + "Chromium", + True, + "Chromium.app/Contents/MacOS/Chromium", + ("chromium", "chromium-browser"), + "Chromium\\Application\\chrome.exe", + "chromium", + ), + _Spec( + "Brave", + True, + "Brave Browser.app/Contents/MacOS/Brave Browser", + ("brave-browser",), + "BraveSoftware\\Brave-Browser\\Application\\brave.exe", + "brave-browser", + ), + _Spec( + "Edge", + True, + "Microsoft Edge.app/Contents/MacOS/Microsoft Edge", + ("microsoft-edge",), + "Microsoft\\Edge\\Application\\msedge.exe", + "microsoft-edge", + ), + _Spec("Arc", True, "Arc.app/Contents/MacOS/Arc"), + _Spec("Helium", True, "Helium.app/Contents/MacOS/Helium"), + _Spec( + "Firefox", + False, + "Firefox.app/Contents/MacOS/firefox", + ("firefox",), + "Mozilla Firefox\\firefox.exe", + ), + _Spec("Safari", False, "Safari.app/Contents/MacOS/Safari"), +) + + +@dataclass(frozen=True) +class Browser: + name: str + path: str + drivable: bool + running: bool + + +@dataclass(frozen=True) +class Endpoint: + url: str + reachable: bool | None # None means "not probeable" (raw WebSocket URL) + product: str + + +def detect_browsers() -> list[Browser]: + """Return every known browser installed on this machine.""" + system = platform.system() + commands = _running_commands() + found: list[Browser] = [] + for spec in _SPECS: + path = _spec_path(spec, system) + if path is None: + continue + found.append( + Browser( + name=spec.name, + path=str(path), + drivable=spec.drivable, + running=_spec_running(spec, commands), + ) + ) + return found + + +def drivable_browsers(browsers: list[Browser] | None = None) -> list[Browser]: + browsers = detect_browsers() if browsers is None else browsers + return [browser for browser in browsers if browser.drivable] + + +def undrivable_browsers(browsers: list[Browser] | None = None) -> list[Browser]: + browsers = detect_browsers() if browsers is None else browsers + return [browser for browser in browsers if not browser.drivable] + + +def install_suggestions() -> list[str]: + """Print-ready commands that install a drivable browser.""" + system = platform.system() + if system == "Darwin": + return [ + f"brew install --cask {spec.brew}" + for spec in _SPECS + if spec.drivable and spec.brew + ] + if system == "Windows": + return ["winget install -e --id Google.Chrome"] + return [ + "sudo apt install chromium # Debian/Ubuntu", + "sudo dnf install chromium # Fedora", + ] + + +def probe_endpoint(url: str) -> Endpoint: + """Ask a DevTools endpoint who it is. Only http(s) URLs are probeable.""" + if not url.startswith(("http://", "https://")): + return Endpoint(url=url, reachable=None, product="websocket endpoint") + try: + with urllib.request.urlopen( + url.rstrip("/") + "/json/version", timeout=_PROBE_TIMEOUT_SECONDS + ) as response: + payload = json.loads(response.read()) + except (OSError, ValueError): + return Endpoint(url=url, reachable=False, product="unreachable") + product = str( + payload.get("Browser") or payload.get("product") or "DevTools endpoint" + ) + return Endpoint(url=url, reachable=True, product=product) + + +def reachable_endpoints(extra: list[str] | tuple[str, ...] = ()) -> list[Endpoint]: + """Probe the harness's default ports plus any caller-supplied endpoint.""" + urls = [f"http://127.0.0.1:{port}" for port in DEFAULT_CDP_PORTS] + urls += [url for url in extra if url and url not in urls] + return [probe_endpoint(url) for url in urls] + + +def _spec_path(spec: _Spec, system: str) -> Path | None: + if system == "Darwin": + for base in _MAC_APPLICATION_DIRS: + candidate = base / spec.mac + if candidate.exists(): + return candidate + return None + if system == "Windows": + if not spec.windows: + return None + for root in _WINDOWS_ROOTS: + candidate = root / spec.windows + if candidate.exists(): + return candidate + return None + for binary in spec.linux: + found = shutil.which(binary) + if found: + return Path(found) + return None + + +def _spec_running(spec: _Spec, commands: set[str]) -> bool: + return bool(spec.process_names() & commands) + + +def _running_commands() -> set[str]: + """Executable names currently running. Empty when the OS has no cheap probe.""" + argv = ( + ["tasklist", "/FO", "CSV", "/NH"] + if platform.system() == "Windows" + else ["ps", "-Ao", "args="] + ) + try: + completed = subprocess.run( + argv, capture_output=True, text=True, timeout=3.0, check=False + ) + except (OSError, subprocess.TimeoutExpired): + return set() + if completed.returncode != 0: + return set() + names: set[str] = set() + for line in completed.stdout.splitlines(): + line = line.strip().strip('"') + if not line: + continue + first = ( + line.split(",")[0] + if platform.system() == "Windows" + else line.split(maxsplit=1)[0] + ) + names.add(Path(first.replace("\\", "/")).name.casefold()) + return names diff --git a/code_puppy_core_plugins/browser_harness/cli.py b/code_puppy_core_plugins/browser_harness/cli.py new file mode 100644 index 0000000..1662ab7 --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/cli.py @@ -0,0 +1,222 @@ +"""Locate and drive the ``browser-harness`` CLI as a subprocess. + +Everything that talks to the outside world lives here: executable discovery, +environment composition, timeouts, and translating the harness's own failure +strings into fixes that actually work. +""" + +from __future__ import annotations + +import os +import re +import shutil +import subprocess +from dataclasses import dataclass +from pathlib import Path + +from . import policy +from .policy import BrowserHarnessError + +EXECUTABLE_ENV_VAR = "CODE_PUPPY_BROWSER_HARNESS_BIN" +DEFAULT_TIMEOUT_SECONDS = 120.0 +MAX_CAPTURED_CHARS = 20_000 +INSTALL_COMMAND = "uv tool install --python 3.12 --upgrade --force browser-harness" +_DAEMON_NAME_RE = re.compile(r"\A[A-Za-z0-9_-]{1,64}\Z") +_UV_TOOL_BIN = Path.home() / ".local" / "bin" / "browser-harness" + +# Substrings the harness emits when a connection cannot be made, mapped to the +# fix that clears it. Source of truth: src/browser_harness/daemon.py and +# admin.py in browser-use/browser-harness. +_FIXUPS: tuple[tuple[str, str], ...] = ( + ( + "permission-blocked", + "Chrome showed its 'Allow remote debugging?' sheet. Run " + "`browser-harness mac-approve`, then retry.", + ), + ( + "remote debugging is turned off", + "Open chrome://inspect/#remote-debugging and tick 'Allow remote debugging " + "for this browser instance', then retry.", + ), + ( + "devtoolsactiveport not found", + "No Chrome profile with remote debugging was found. Open " + "chrome://inspect/#remote-debugging and tick the toggle, or point Code " + "Puppy at an explicit endpoint with `/browser connect `.", + ), + ( + "chrome-not-running", + "No Chromium-family browser is running. Start one; Code Puppy's `/browser " + "status` lists the ones installed here.", + ), +) + + +@dataclass(frozen=True) +class HarnessResult: + """One completed ``browser-harness`` invocation. Both streams are capped.""" + + ok: bool + exit_code: int + stdout: str + stderr: str + timed_out: bool = False + + def failure(self) -> str: + """A failure description with the concrete next step appended.""" + detail = (self.stderr or self.stdout).strip() or "no output" + if self.timed_out: + detail += " (timed out)" + fixup = fixup_for(f"{self.stderr}\n{self.stdout}") + return detail if fixup is None else f"{detail}\n\nFix: {fixup}" + + +def fixup_for(text: str) -> str | None: + """Return the documented fix for a known harness error, if any matches.""" + lowered = text.casefold() + for needle, fix in _FIXUPS: + if needle in lowered: + return fix + return None + + +def executable() -> str | None: + """Resolve the ``browser-harness`` entry point, honouring an override.""" + override = os.environ.get(EXECUTABLE_ENV_VAR, "").strip() + if override: + candidate = Path(override).expanduser() + if candidate.is_file(): + return str(candidate) + found = shutil.which(override) + if found: + return found + raise BrowserHarnessError( + f"{EXECUTABLE_ENV_VAR} points at {override!r}, which is not an " + "executable. Unset it or fix the path." + ) + return shutil.which("browser-harness") or ( + str(_UV_TOOL_BIN) if _UV_TOOL_BIN.is_file() else None + ) + + +def installed() -> bool: + return executable() is not None + + +def require_executable() -> str: + path = executable() + if path is None: + raise BrowserHarnessError( + "browser-harness is not installed. Install it with:\n " + f"{INSTALL_COMMAND}\nThen run `/browser status` to check the " + "connection." + ) + return path + + +def version() -> str | None: + """Return the installed harness version, or None when unavailable.""" + path = executable() + if path is None: + return None + return ( + _invoke([path, "--version"], timeout=15.0, env=environment()).stdout.strip() + or None + ) + + +def environment(daemon: str | None = None) -> dict[str, str]: + """Compose the child environment without clobbering explicit user config. + + An endpoint already present in the ambient environment always wins: the + harness treats ``BU_CDP_URL``/``BU_CDP_WS`` as an explicit override, and a + saved Code Puppy setting must not silently replace it. + """ + env = dict(os.environ) + endpoint = policy.settings_store.endpoint() + if endpoint and ambient_endpoint(env) is None: + env["BU_CDP_URL" if endpoint.startswith("http") else "BU_CDP_WS"] = endpoint + if daemon: + if not _DAEMON_NAME_RE.match(daemon): + raise BrowserHarnessError( + f"Invalid browser name {daemon!r}: use 1-64 letters, digits, '-' " + "or '_'." + ) + env["BU_NAME"] = daemon + return env + + +def ambient_endpoint(env: dict[str, str] | None = None) -> str | None: + """Return an endpoint the user exported, which outranks the saved setting.""" + env = dict(os.environ) if env is None else env + return env.get("BU_CDP_WS") or env.get("BU_CDP_URL") or None + + +def run_script( + script: str, daemon: str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS +) -> HarnessResult: + """Execute a helper script with the harness helpers pre-imported.""" + return _invoke( + [require_executable()], + timeout=timeout, + env=environment(daemon), + input_text=script, + ) + + +def run_command( + args: list[str], timeout: float = DEFAULT_TIMEOUT_SECONDS +) -> HarnessResult: + """Run a ``browser-harness`` sub-command such as ``--doctor``.""" + return _invoke([require_executable(), *args], timeout=timeout, env=environment()) + + +def _invoke( + argv: list[str], + timeout: float, + env: dict[str, str], + input_text: str | None = None, +) -> HarnessResult: + # ``input=`` implies a pipe; sub-commands get DEVNULL so an interactive + # prompt inside the harness can never hang Code Puppy's UI. + streams: dict[str, object] = ( + {} if input_text is not None else {"stdin": subprocess.DEVNULL} + ) + try: + completed = subprocess.run( # noqa: S603 - argv is our resolved entry point + argv, + input=input_text, + capture_output=True, + text=True, + env=env, + timeout=timeout, + check=False, + **streams, + ) + except subprocess.TimeoutExpired as exc: + return HarnessResult( + ok=False, + exit_code=-1, + stdout=_cap(_text(exc.stdout)), + stderr=_cap(_text(exc.stderr)) or "browser-harness timed out", + timed_out=True, + ) + except FileNotFoundError as exc: # pragma: no cover - resolved path vanished + raise BrowserHarnessError(f"browser-harness could not be run: {exc}") from exc + return HarnessResult( + ok=completed.returncode == 0, + exit_code=completed.returncode, + stdout=_cap(completed.stdout), + stderr=_cap(completed.stderr), + ) + + +def _cap(value: str | None) -> str: + text = value or "" + return text if len(text) <= MAX_CAPTURED_CHARS else text[:MAX_CAPTURED_CHARS] + "…" + + +def _text(value: str | bytes | None) -> str: + if value is None: + return "" + return value.decode(errors="replace") if isinstance(value, bytes) else value diff --git a/code_puppy_core_plugins/browser_harness/commands.py b/code_puppy_core_plugins/browser_harness/commands.py new file mode 100644 index 0000000..98b05c3 --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/commands.py @@ -0,0 +1,208 @@ +"""/browser - connect Code Puppy to a real browser through browser-harness.""" + +from __future__ import annotations + +import shlex + +from code_puppy.messaging import emit_error, emit_info, emit_success, emit_warning + +from . import browser, cli +from . import policy +from .policy import BrowserHarnessError + +USAGE = ( + "Usage: /browser [status|enable|disable|doctor|connect |" + "disconnect|install|recordings [on|off]]" +) + + +def command_help() -> list[tuple[str, str]]: + return [ + ( + "browser", + "Connect, enable, or diagnose browser-harness browser control", + ) + ] + + +def handle_command(command: str, name: str) -> bool | None: + if name != "browser": + return None + try: + tokens = shlex.split(command) + except ValueError as exc: + emit_error(f"Invalid /browser command: {exc}") + return True + subcommand = tokens[1].casefold() if len(tokens) > 1 else "status" + + if subcommand == "status": + _emit_status() + elif subcommand == "doctor": + _emit_doctor() + elif subcommand == "install": + _emit_install_help() + elif subcommand == "enable": + policy.settings_store.set_enabled(True) + _refresh_tool_registry() + emit_success( + "Browser control enabled. Code Puppy can now drive your real, " + "signed-in browser; the browser tools are registered for this session." + ) + elif subcommand == "disable": + policy.settings_store.set_enabled(False) + emit_warning("Browser control disabled; running browser tools is blocked.") + elif subcommand == "connect": + _connect(tokens[2] if len(tokens) > 2 else None) + elif subcommand == "disconnect": + policy.settings_store.clear_endpoint() + emit_success( + "Cleared the saved endpoint. The harness will auto-discover your " + "running Chromium browser again." + ) + elif subcommand == "recordings": + _recordings(tokens[2] if len(tokens) > 2 else None) + else: + emit_error(USAGE) + return True + + +def _refresh_tool_registry() -> None: + """Merge the newly consented tools in without asking the user to restart. + + Asking for the available tool names re-runs Code Puppy's ``register_tools`` + hook, which is exactly how plugin tools reach ``TOOL_REGISTRY``. Disabling + needs no such dance: the tools refuse to run on their own and vanish from + the registry on the next start. + """ + try: + from code_puppy.tools import get_available_tool_names + + get_available_tool_names() + except Exception: # pragma: no cover - only costs a restart, not a session + pass + + +def _emit_status() -> None: + _emit_install_state() + consent = policy.settings_store.consent_state() + if consent == "unset": + emit_warning( + "Browser control is awaiting your one-time consent. Run `/browser " + "enable` to allow it, or `/browser disable` to keep it off." + ) + else: + emit_info(f"Browser control consent: {consent}") + _emit_connection() + _emit_browsers() + + +def _emit_install_state() -> None: + try: + path = cli.executable() + except BrowserHarnessError as exc: + emit_error(str(exc)) + return + if path is None: + emit_warning( + f"browser-harness is not installed. Install it with:\n {cli.INSTALL_COMMAND}" + ) + return + emit_info(f"browser-harness {cli.version() or 'unknown version'} ({path})") + + +def _emit_connection() -> None: + ambient, saved = cli.ambient_endpoint(), policy.settings_store.endpoint() + if ambient: + emit_info(f"CDP endpoint: {ambient} (from BU_CDP_WS/BU_CDP_URL)") + elif saved: + emit_info(f"CDP endpoint: {saved} (saved by /browser connect)") + else: + emit_info( + "CDP endpoint: auto-discovery - the harness attaches to whichever " + "Chromium browser is running" + ) + endpoints = browser.reachable_endpoints( + [ambient or saved] if ambient or saved else [] + ) + live = [endpoint for endpoint in endpoints if endpoint.reachable] + for endpoint in live: + emit_info(f" answering on {endpoint.url}: {endpoint.product}") + if not live: + emit_info(" no local DevTools endpoint is answering yet") + + +def _emit_browsers() -> None: + installed = browser.detect_browsers() + drivable = browser.drivable_browsers(installed) + if not drivable: + emit_warning(f"No drivable browser is installed. {browser.UNSUPPORTED_NOTE}") + _emit_install_help() + for item in drivable: + emit_info(f" {item.name}: {'running' if item.running else 'installed'}") + blocked = [item.name for item in browser.undrivable_browsers(installed)] + if blocked: + emit_info(f" present but not drivable: {', '.join(blocked)}") + + +def _emit_install_help() -> None: + emit_info( + "Install a Chromium-family browser, then ask me to retry:\n " + + "\n ".join(browser.install_suggestions()) + ) + + +def _connect(target: str | None) -> None: + if not target: + emit_error( + "Usage: /browser connect e.g. " + "http://127.0.0.1:9222 or wss://your-browser.example/cdp" + ) + return + try: + policy.settings_store.set_endpoint(target) + except BrowserHarnessError as exc: + emit_error(str(exc)) + return + endpoint = browser.probe_endpoint(target) + if endpoint.reachable: + emit_success(f"Saved {endpoint.url} - {endpoint.product}") + elif endpoint.reachable is None: + emit_success( + f"Saved {endpoint.url}. The harness resolves WebSocket endpoints on " + "first use, so it is not probed here." + ) + else: + emit_warning( + f"Saved {endpoint.url}, but nothing answered just now. Start that " + "browser, or run `/browser doctor`." + ) + + +def _emit_doctor() -> None: + try: + result = cli.run_command(["--doctor"], timeout=60.0) + except BrowserHarnessError as exc: + emit_error(str(exc)) + return + emit_info(result.stdout.strip() or "(no report)") + if not result.ok: + emit_warning(result.failure()) + return + emit_success("browser-harness reports a healthy connection.") + + +def _recordings(action: str | None) -> None: + args = { + "on": ["recordings", "enable"], + "off": ["recordings", "disable"], + None: ["recordings"], + }.get(action.casefold() if action else None) + if args is None: + emit_error("Usage: /browser recordings [on|off]") + return + try: + result = cli.run_command(args, timeout=30.0) + except BrowserHarnessError as exc: + emit_error(str(exc)) + return + emit_info(result.stdout.strip() if result.ok else result.failure()) diff --git a/code_puppy_core_plugins/browser_harness/policy.py b/code_puppy_core_plugins/browser_harness/policy.py new file mode 100644 index 0000000..1b6afe9 --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/policy.py @@ -0,0 +1,122 @@ +"""Persisted consent and connection target for the browser-harness plugin.""" + +from __future__ import annotations + +import json +import os +import threading +from pathlib import Path +from typing import Any + +from code_puppy.config import CONFIG_DIR + +POLICY_PATH = Path(CONFIG_DIR) / "browser_harness_policy.json" +ENDPOINT_SCHEMES = ("http://", "https://", "ws://", "wss://") + + +class BrowserHarnessError(RuntimeError): + """Raised when the harness is unusable or refused a request.""" + + +class SettingsStore: + """Remember the one-time opt-in and an optional explicit CDP endpoint. + + The endpoint mirrors browser-harness's own escape hatch: an HTTP DevTools + URL (``BU_CDP_URL``) or a WebSocket URL (``BU_CDP_WS``). Leaving it unset + defers browser discovery entirely to the harness, which auto-detects the + running Chromium-family browser. + """ + + def __init__(self, path: Path = POLICY_PATH) -> None: + self.path = path + self._lock = threading.RLock() + + def _load(self) -> dict[str, Any]: + if not self.path.is_file(): + return {"enabled": None, "endpoint": None} + try: + payload = json.loads(self.path.read_text()) + except (OSError, json.JSONDecodeError): + return {"enabled": None, "endpoint": None} + enabled = payload.get("enabled") + endpoint = payload.get("endpoint") + return { + "enabled": enabled if isinstance(enabled, bool) else None, + "endpoint": endpoint if isinstance(endpoint, str) and endpoint else None, + } + + def _save(self, payload: dict[str, Any]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + temporary = self.path.with_suffix(".tmp") + temporary.write_text(json.dumps(payload, indent=2, sort_keys=True)) + os.chmod(temporary, 0o600) + temporary.replace(self.path) + + # --- consent --------------------------------------------------------- + def set_enabled(self, enabled: bool) -> None: + with self._lock: + payload = self._load() + payload["enabled"] = enabled + self._save(payload) + + def is_enabled(self) -> bool: + """Return whether the user explicitly opted in to browser control.""" + with self._lock: + return self._load()["enabled"] is True + + def consent_state(self) -> str: + with self._lock: + enabled = self._load()["enabled"] + return "unset" if enabled is None else "enabled" if enabled else "disabled" + + def require_enabled(self) -> None: + with self._lock: + state = self.consent_state() + if state == "unset": + raise BrowserHarnessError( + "Browser Harness needs your one-time permission before its first " + "use. It drives your real, signed-in browser: it can read pages, " + "click, type, and download. Run `/browser enable` to allow it, or " + "`/browser disable` to keep it off. You can change this later with " + "the same commands." + ) + if state == "disabled": + raise BrowserHarnessError( + "Browser Harness is disabled in settings. Run `/browser enable` to " + "turn it on." + ) + + # --- connection target ---------------------------------------------- + def endpoint(self) -> str | None: + with self._lock: + return self._load()["endpoint"] + + def set_endpoint(self, endpoint: str) -> None: + normalized = endpoint.strip().rstrip("/") + if not normalized.lower().startswith(ENDPOINT_SCHEMES): + raise BrowserHarnessError( + f"Invalid CDP endpoint {endpoint!r}: expected an http(s):// DevTools " + "URL (for example http://127.0.0.1:9222) or a ws(s):// URL." + ) + with self._lock: + payload = self._load() + payload["endpoint"] = normalized + self._save(payload) + + def clear_endpoint(self) -> None: + with self._lock: + payload = self._load() + payload["endpoint"] = None + self._save(payload) + + def status(self) -> dict[str, Any]: + with self._lock: + payload = self._load() + return { + "consent": payload["enabled"], + "endpoint": payload["endpoint"], + "endpoint_source": "saved" if payload["endpoint"] else "auto-discovery", + } + + +settings_store = SettingsStore() diff --git a/code_puppy_core_plugins/browser_harness/register_callbacks.py b/code_puppy_core_plugins/browser_harness/register_callbacks.py new file mode 100644 index 0000000..3ce71fd --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/register_callbacks.py @@ -0,0 +1,97 @@ +"""Register browser-harness browser control as an opt-in builtin plugin. + +The plugin stays silent until the user consents with ``/browser enable``: +driving a real, signed-in browser deserves the same explicit opt-in that macOS +Computer Use requires. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from code_puppy.callbacks import register_callback +from code_puppy.messaging import emit_info + +from . import cli +from . import policy + +_SKILL_PATH = Path(__file__).with_name("SKILL.md") +_TOOL_NAMES = ("browser_harness", "browser_screenshot", "browser_doctor") + + +def _installed() -> bool: + try: + return cli.installed() + except Exception: # pragma: no cover - a bad override must not break startup + return False + + +def _register_tools() -> list[dict[str, Any]]: + if not policy.settings_store.is_enabled(): + return [] + + from .tools import REGISTRARS + + return [{"name": name, "register_func": REGISTRARS[name]} for name in _TOOL_NAMES] + + +def _register_agent_tools(agent_name: str | None = None) -> list[str]: + del agent_name + return list(_TOOL_NAMES) if policy.settings_store.is_enabled() else [] + + +def _register_skills() -> list[dict[str, str]]: + if not policy.settings_store.is_enabled(): + return [] + return [{"name": "browser-harness", "skill_md_path": str(_SKILL_PATH)}] + + +def _load_prompt() -> str | None: + if not policy.settings_store.is_enabled(): + return None + return ( + "Browser control runs inside the user's real, signed-in browser through " + "browser-harness. Use the browser_harness tool when a task needs those " + "sessions; use plain HTTP for public pages and the web-retriever agent " + "for sandboxed scraping. The first navigation of a task is new_tab(url): " + "the harness keeps one tab attached, so do not reopen tabs per call or " + "close tabs you did not create. Locate elements in the accessibility " + "tree and verify each action with page_info() or js() rather than " + "screenshots. Never submit passwords, MFA codes, payments, deletions, or " + "published content without asking first - being signed in is not consent. " + "On a connection error, call browser_doctor() and apply the fix it " + "prints instead of retrying blindly. Firefox and Safari cannot be driven " + "at all (they expose no CDP endpoint); Chrome, Chromium, Brave, Edge, " + "Arc, and Helium can be." + ) + + +def _startup() -> None: + if _installed() and policy.settings_store.consent_state() == "unset": + emit_info( + "browser-harness is installed but browser control is off. Run " + "`/browser status` to see connection health, or `/browser enable` to " + "let Code Puppy drive this machine's Chromium browser." + ) + + +def _custom_help(): + from .commands import command_help + + return command_help() + + +def _custom_command(command: str, name: str): + from .commands import handle_command + + return handle_command(command, name) + + +register_callback("startup", _startup) +register_callback("register_tools", _register_tools) +register_callback("register_agent_tools", _register_agent_tools) +register_callback("register_skills", _register_skills) +register_callback("load_prompt", _load_prompt) +register_callback("custom_command_help", _custom_help) +register_callback("custom_command", _custom_command) diff --git a/code_puppy_core_plugins/browser_harness/tools.py b/code_puppy_core_plugins/browser_harness/tools.py new file mode 100644 index 0000000..c16f73d --- /dev/null +++ b/code_puppy_core_plugins/browser_harness/tools.py @@ -0,0 +1,182 @@ +"""Pydantic-AI tools that drive a real browser through browser-harness.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any, Callable + +from pydantic_ai import BinaryContent, RunContext, ToolReturn + +from . import cli +from . import policy +from .policy import BrowserHarnessError + +try: # Same distribution as the computer-use plugin; reuse its terminal renderer. + from code_puppy_core_plugins.computer_use.inline_image import emit_inline_image +except ImportError: # pragma: no cover - the bundle always ships both + + def emit_inline_image(path: str | Path) -> bool: + del path + return False + + +MIN_TIMEOUT_SECONDS = 5.0 +MAX_TIMEOUT_SECONDS = 900.0 +#: Keeps captures inside the long-edge limit image-aware models resample to. +DEFAULT_SCREENSHOT_MAX_DIM = 1568 +_SCREENSHOT_SCRIPT = "print(capture_screenshot(full={full}, max_dim={max_dim}))" + + +def _clamp_timeout(timeout: float) -> float: + return min(max(float(timeout), MIN_TIMEOUT_SECONDS), MAX_TIMEOUT_SECONDS) + + +async def _attempt(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + """Run a blocking harness call, turning errors into model-readable results.""" + try: + return await asyncio.to_thread(func, *args, **kwargs) + except BrowserHarnessError as exc: + return {"success": False, "error": str(exc)} + except Exception as exc: # pragma: no cover - never leak a traceback + return {"success": False, "error": f"browser-harness failed: {exc}"} + + +def _consent_blocked() -> dict[str, Any] | None: + try: + policy.settings_store.require_enabled() + except BrowserHarnessError as exc: + return {"success": False, "error": str(exc)} + return None + + +def _last_line(text: str) -> str: + lines = [line.strip() for line in text.splitlines() if line.strip()] + return lines[-1] if lines else "" + + +def register_browser_harness(agent): + @agent.tool + async def browser_harness( + context: RunContext, + script: str, + browser_name: str | None = None, + timeout: float = cli.DEFAULT_TIMEOUT_SECONDS, + ) -> Any: + """Run Python against the browser using browser-harness helpers. + + Helpers arrive pre-imported: new_tab, goto_url, page_info, js, cdp, + click_at_xy, type_text, fill_input, press_key, scroll, capture_screenshot, + list_tabs, current_tab, switch_tab, close_tab, wait_for_load, + wait_for_element, wait_for_network_idle, ensure_real_tab, upload_file. + Print whatever you want back - stdout is the only channel. + + The first navigation of a task is new_tab(url); the attached tab is + remembered between calls, so do not open another one per script. Prefer + the accessibility tree (cdp("Accessibility.getFullAXTree")) and js() + over screenshots to locate elements, then click by coordinate. Stop and + ask before entering passwords, MFA codes, or completing purchases. + + browser_name selects a named harness daemon (a cloud browser started with + start_remote_daemon); leave it unset for your local browser. + """ + del context + blocked = _consent_blocked() + if blocked is not None: + return blocked + result = await _attempt( + cli.run_script, script, browser_name, _clamp_timeout(timeout) + ) + if not isinstance(result, cli.HarnessResult): + return result + if result.ok: + return {"success": True, "output": result.stdout.strip() or "(no output)"} + return {"success": False, "error": result.failure(), "output": result.stdout} + + return browser_harness + + +def register_browser_screenshot(agent): + @agent.tool + async def browser_screenshot( + context: RunContext, + full: bool = False, + max_dim: int | None = DEFAULT_SCREENSHOT_MAX_DIM, + ) -> Any: + """Capture the attached tab and show it inline to the user. + + Pixels alone cannot tell you what is clickable: pair a capture with + page_info() or js() before acting. Set full=True for the whole scroll + height, and max_dim=None to keep native resolution. + """ + del context + blocked = _consent_blocked() + if blocked is not None: + return blocked + script = _SCREENSHOT_SCRIPT.format(full=bool(full), max_dim=max_dim) + result = await _attempt(cli.run_script, script) + if not isinstance(result, cli.HarnessResult): + return result + if not result.ok: + return {"success": False, "error": result.failure()} + path = _last_line(result.stdout) + image = Path(path) + if not image.is_file(): + return { + "success": False, + "error": f"browser-harness reported no screenshot file at {path!r}", + "output": result.stdout, + } + metadata = {"success": True, "path": path, "displayed_inline": False} + content = [ + f"Here is the browser screenshot ({'full page' if full else 'viewport'}):" + ] + try: + metadata["displayed_inline"] = emit_inline_image(path) + content.append( + BinaryContent(data=image.read_bytes(), media_type="image/png") + ) + except OSError as exc: + return {"success": False, "error": f"Could not read screenshot: {exc}"} + return ToolReturn(return_value=metadata, content=content, metadata=metadata) + + return browser_screenshot + + +def register_browser_doctor(agent): + @agent.tool + async def browser_doctor(context: RunContext) -> Any: + """Report browser-harness install, daemon, and browser connection health. + + Use it before blaming a script for a connection problem: the report names + the exact fix (remote-debugging toggle, mac-approve approval, or starting + a browser). + """ + del context + blocked = _consent_blocked() + if blocked is not None: + return blocked + result = await _attempt(cli.run_command, ["--doctor"], timeout=60.0) + if not isinstance(result, cli.HarnessResult): + return result + report = result.stdout.strip() + if result.ok: + return {"success": True, "healthy": True, "report": report or "(no output)"} + payload: dict[str, Any] = { + "success": True, + "healthy": False, + "report": report or result.failure(), + } + fixup = cli.fixup_for(f"{result.stderr}\n{result.stdout}") + if fixup: + payload["fix"] = fixup + return payload + + return browser_doctor + + +REGISTRARS = { + "browser_harness": register_browser_harness, + "browser_screenshot": register_browser_screenshot, + "browser_doctor": register_browser_doctor, +} diff --git a/plugin-names.txt b/plugin-names.txt index dabd13f..6295982 100644 --- a/plugin-names.txt +++ b/plugin-names.txt @@ -4,6 +4,7 @@ agent_creator_skill aws_bedrock azure_foundry auto_continue +browser_harness btw chatgpt_oauth claude_code_hooks diff --git a/pyproject.toml b/pyproject.toml index f54fe05..14056f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ agent_creator_skill = "code_puppy_core_plugins.agent_creator_skill.register_call aws_bedrock = "code_puppy_core_plugins.aws_bedrock.register_callbacks" azure_foundry = "code_puppy_core_plugins.azure_foundry.register_callbacks" auto_continue = "code_puppy_core_plugins.auto_continue.register_callbacks" +browser_harness = "code_puppy_core_plugins.browser_harness.register_callbacks" btw = "code_puppy_core_plugins.btw.register_callbacks" chatgpt_oauth = "code_puppy_core_plugins.chatgpt_oauth.register_callbacks" claude_code_hooks = "code_puppy_core_plugins.claude_code_hooks.register_callbacks" diff --git a/tests/browser_harness/__init__.py b/tests/browser_harness/__init__.py new file mode 100644 index 0000000..1b56e5b --- /dev/null +++ b/tests/browser_harness/__init__.py @@ -0,0 +1 @@ +"""Tests for the browser-harness plugin.""" diff --git a/tests/browser_harness/_helpers.py b/tests/browser_harness/_helpers.py new file mode 100644 index 0000000..db9bbb5 --- /dev/null +++ b/tests/browser_harness/_helpers.py @@ -0,0 +1,53 @@ +"""Shared fakes for the browser-harness plugin tests.""" + +from __future__ import annotations + +import subprocess + + +class FakeAgent: + """Stand-in for a pydantic-ai agent that captures ``@agent.tool`` calls.""" + + def __init__(self) -> None: + self.registered = {} + + def tool(self, function): + self.registered[function.__name__] = function + return function + + +class FakeSubprocess: + """Replay one canned ``subprocess.run`` result and record the calls.""" + + def __init__( + self, + stdout: str = "", + stderr: str = "", + returncode: int = 0, + error: BaseException | None = None, + ) -> None: + self.stdout = stdout + self.stderr = stderr + self.returncode = returncode + self.error = error + self.calls: list[tuple[list[str], dict]] = [] + + def __call__(self, argv, **kwargs): + self.calls.append((list(argv), kwargs)) + if self.error is not None: + raise self.error + return subprocess.CompletedProcess( + argv, self.returncode, self.stdout, self.stderr + ) + + @property + def env(self) -> dict: + return self.calls[0][1]["env"] + + @property + def stdin(self): + return self.calls[0][1].get("stdin") + + @property + def argv(self) -> list[str]: + return self.calls[0][0] diff --git a/tests/browser_harness/conftest.py b/tests/browser_harness/conftest.py new file mode 100644 index 0000000..3302fe3 --- /dev/null +++ b/tests/browser_harness/conftest.py @@ -0,0 +1,42 @@ +"""Shared fixtures for the browser-harness plugin tests.""" + +from __future__ import annotations + +import pytest + +from code_puppy_core_plugins.browser_harness import cli +from code_puppy_core_plugins.browser_harness import policy +from code_puppy_core_plugins.browser_harness.policy import SettingsStore + + +@pytest.fixture(autouse=True) +def clean_harness_env(monkeypatch): + """Never let the developer's real harness config leak into a test.""" + for name in ("BU_CDP_URL", "BU_CDP_WS", "BU_NAME", cli.EXECUTABLE_ENV_VAR): + monkeypatch.delenv(name, raising=False) + + +@pytest.fixture(autouse=True) +def _quiet_registry_refresh(monkeypatch): + """Stop /browser enable from mutating Code Puppy's real tool registry.""" + from code_puppy import tools as core_tools + + monkeypatch.setattr(core_tools, "get_available_tool_names", lambda: []) + + +@pytest.fixture +def store(tmp_path, monkeypatch): + """Point the whole plugin at one throwaway settings file.""" + isolated = SettingsStore(tmp_path / "browser_harness_policy.json") + monkeypatch.setattr(policy, "settings_store", isolated) + return isolated + + +@pytest.fixture +def harness_bin(tmp_path, monkeypatch): + """Pretend ``browser-harness`` is installed, via the override env var.""" + binary = tmp_path / "browser-harness" + binary.write_text("#!/bin/sh\n") + binary.chmod(0o755) + monkeypatch.setenv(cli.EXECUTABLE_ENV_VAR, str(binary)) + return binary diff --git a/tests/browser_harness/test_browser.py b/tests/browser_harness/test_browser.py new file mode 100644 index 0000000..9a1d9a7 --- /dev/null +++ b/tests/browser_harness/test_browser.py @@ -0,0 +1,152 @@ +"""Browser discovery and DevTools endpoint probing.""" + +from __future__ import annotations + +import io +import json + +import pytest + +from code_puppy_core_plugins.browser_harness import browser + + +def test_only_drivable_browsers_are_recommended(monkeypatch): + monkeypatch.setattr(browser.platform, "system", lambda: "Darwin") + suggestions = browser.install_suggestions() + + assert "brew install --cask google-chrome" in suggestions + assert all("firefox" not in s and "safari" not in s for s in suggestions) + + +@pytest.mark.parametrize( + "system, expected", + [ + ("Linux", "chromium"), + ("Windows", "winget install -e --id Google.Chrome"), + ], +) +def test_install_suggestions_follow_the_platform(system, expected, monkeypatch): + monkeypatch.setattr(browser.platform, "system", lambda: system) + assert any(expected in suggestion for suggestion in browser.install_suggestions()) + + +def test_filters_split_browsers_by_drivability(): + found = [ + browser.Browser("Chrome", "/x", True, False), + browser.Browser("Firefox", "/y", False, True), + ] + + assert [item.name for item in browser.drivable_browsers(found)] == ["Chrome"] + assert [item.name for item in browser.undrivable_browsers(found)] == ["Firefox"] + + +def test_mac_detection_trusts_the_filesystem(tmp_path, monkeypatch): + apps = tmp_path / "Applications" + (apps / "Firefox.app/Contents/MacOS/firefox").parent.mkdir(parents=True) + (apps / "Firefox.app/Contents/MacOS/firefox").write_text("") + chrome = apps / "Google Chrome.app/Contents/MacOS/Google Chrome" + chrome.parent.mkdir(parents=True) + chrome.write_text("") + monkeypatch.setattr(browser.platform, "system", lambda: "Darwin") + monkeypatch.setattr(browser, "_MAC_APPLICATION_DIRS", (apps,)) + monkeypatch.setattr(browser, "_running_commands", lambda: {"google chrome"}) + + found = browser.detect_browsers() + + assert {item.name for item in browser.drivable_browsers(found)} == {"Chrome"} + assert [item.name for item in found if item.running] == ["Chrome"] + assert [item.name for item in browser.undrivable_browsers(found)] == ["Firefox"] + + +def test_linux_detection_uses_path_lookup(tmp_path, monkeypatch): + binaries = {"brave-browser": str(tmp_path / "brave-browser")} + monkeypatch.setattr(browser.platform, "system", lambda: "Linux") + monkeypatch.setattr(browser.shutil, "which", binaries.get) + monkeypatch.setattr(browser, "_running_commands", lambda: {"brave-browser"}) + + found = browser.detect_browsers() + + assert [item.name for item in found] == ["Brave"] + assert found[0].running is True + + +def test_a_browser_that_is_not_running_stays_quiet(tmp_path, monkeypatch): + apps = tmp_path / "Applications" + chromium = apps / "Chromium.app/Contents/MacOS/Chromium" + chromium.parent.mkdir(parents=True) + chromium.write_text("") + monkeypatch.setattr(browser.platform, "system", lambda: "Darwin") + monkeypatch.setattr(browser, "_MAC_APPLICATION_DIRS", (apps,)) + monkeypatch.setattr(browser, "_running_commands", lambda: set()) + + assert browser.detect_browsers()[0].running is False + + +def test_running_process_probe_survives_a_missing_ps(monkeypatch): + monkeypatch.setattr( + browser.subprocess, + "run", + lambda *a, **k: (_ for _ in ()).throw(OSError("no ps")), + ) + assert browser._running_commands() == set() + + +def test_websocket_endpoints_are_reported_not_probed(): + endpoint = browser.probe_endpoint("wss://host.example/cdp") + + assert endpoint.reachable is None + assert endpoint.product == "websocket endpoint" + + +def test_a_refused_port_is_unreachable(): + assert browser.probe_endpoint("http://127.0.0.1:1").reachable is False + + +def test_a_live_endpoint_names_itself(monkeypatch): + payload = json.dumps({"Browser": "Chrome/144.0.0.0"}).encode() + + class Response(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + monkeypatch.setattr( + browser.urllib.request, "urlopen", lambda url, timeout: Response(payload) + ) + + endpoint = browser.probe_endpoint("http://127.0.0.1:9222/") + + assert (endpoint.reachable, endpoint.product) == (True, "Chrome/144.0.0.0") + + +def test_reachable_endpoints_probe_the_defaults_plus_the_saved_url(monkeypatch): + seen: list[str] = [] + + def record(url): + seen.append(url) + return browser.Endpoint(url, reachable=False, product="unreachable") + + monkeypatch.setattr(browser, "probe_endpoint", record) + + browser.reachable_endpoints(["http://10.0.0.5:9222"]) + + assert seen == [ + "http://127.0.0.1:9222", + "http://127.0.0.1:9223", + "http://10.0.0.5:9222", + ] + + +def test_a_duplicated_endpoint_is_only_probed_once(monkeypatch): + seen: list[str] = [] + monkeypatch.setattr( + browser, + "probe_endpoint", + lambda url: seen.append(url) or browser.Endpoint(url, False, "x"), + ) + + browser.reachable_endpoints(["http://127.0.0.1:9222", ""]) + + assert seen == ["http://127.0.0.1:9222", "http://127.0.0.1:9223"] diff --git a/tests/browser_harness/test_cli.py b/tests/browser_harness/test_cli.py new file mode 100644 index 0000000..4d413c7 --- /dev/null +++ b/tests/browser_harness/test_cli.py @@ -0,0 +1,154 @@ +"""Subprocess plumbing for the browser-harness plugin.""" + +from __future__ import annotations + +import subprocess + +import pytest + +from code_puppy_core_plugins.browser_harness import cli +from code_puppy_core_plugins.browser_harness.policy import BrowserHarnessError + +from ._helpers import FakeSubprocess + + +def test_executable_prefers_the_environment_override(tmp_path, monkeypatch): + override = tmp_path / "bh" + override.write_text("#!/bin/sh\n") + monkeypatch.setenv(cli.EXECUTABLE_ENV_VAR, str(override)) + assert cli.executable() == str(override) + + +def test_executable_shouts_about_a_broken_override(monkeypatch): + monkeypatch.setenv(cli.EXECUTABLE_ENV_VAR, "definitely-not-here") + with pytest.raises(BrowserHarnessError, match="not an executable"): + cli.executable() + + +def test_executable_finds_the_path_and_the_uv_tool_install(monkeypatch, tmp_path): + monkeypatch.setattr( + cli.shutil, "which", lambda name: "/usr/local/bin/browser-harness" + ) + assert cli.executable() == "/usr/local/bin/browser-harness" + + missing = tmp_path / "uv-tool-bin" / "browser-harness" + missing.parent.mkdir() + missing.write_text("#!/bin/sh\n") + monkeypatch.setattr(cli.shutil, "which", lambda name: None) + monkeypatch.setattr(cli, "_UV_TOOL_BIN", missing) + assert cli.executable() == str(missing) + assert cli.installed() is True + + +def test_missing_install_reports_the_install_command(monkeypatch): + monkeypatch.setattr(cli, "executable", lambda: None) + with pytest.raises(BrowserHarnessError, match="uv tool install"): + cli.require_executable() + assert cli.version() is None + + +def test_environment_maps_an_http_endpoint(store, monkeypatch): + store.set_endpoint("http://127.0.0.1:9222") + env = cli.environment() + assert env["BU_CDP_URL"] == "http://127.0.0.1:9222" + assert "BU_CDP_WS" not in env + + +def test_environment_maps_a_websocket_endpoint(store): + store.set_endpoint("wss://browser.example/cdp") + assert cli.environment()["BU_CDP_WS"] == "wss://browser.example/cdp" + + +def test_an_ambient_endpoint_outranks_the_saved_one(store, monkeypatch): + store.set_endpoint("http://127.0.0.1:9222") + monkeypatch.setenv("BU_CDP_WS", "wss://somewhere.else/cdp") + env = cli.environment() + assert "BU_CDP_URL" not in env + assert env["BU_CDP_WS"] == "wss://somewhere.else/cdp" + assert cli.ambient_endpoint() == "wss://somewhere.else/cdp" + + +def test_environment_names_a_daemon_and_rejects_rubbish(store): + assert cli.environment("r7k2")["BU_NAME"] == "r7k2" + with pytest.raises(BrowserHarnessError, match="Invalid browser name"): + cli.environment("../escape") + + +def test_run_script_hands_the_script_over_stdin(harness_bin, monkeypatch): + fake = FakeSubprocess(stdout="page title\n") + monkeypatch.setattr(cli.subprocess, "run", fake) + result = cli.run_script("print(page_info())", "r7k2", 30.0) + + assert result.ok and result.stdout == "page title\n" + assert fake.argv == [str(harness_bin)] + assert fake.calls[0][1]["input"] == "print(page_info())" + # Passing stdin= alongside input= is a ValueError in subprocess.run. + assert "stdin" not in fake.calls[0][1] + assert fake.env["BU_NAME"] == "r7k2" + assert fake.calls[0][1]["timeout"] == 30.0 + + +def test_run_command_never_blocks_on_stdin(harness_bin, monkeypatch): + fake = FakeSubprocess(stdout="healthy\n") + monkeypatch.setattr(cli.subprocess, "run", fake) + result = cli.run_command(["--doctor"]) + + assert result.ok + assert fake.argv[-1] == "--doctor" + assert fake.stdin == subprocess.DEVNULL + + +def test_timeouts_become_a_readable_failure(harness_bin, monkeypatch): + error = subprocess.TimeoutExpired( + cmd="browser-harness", timeout=1, output=b"partial" + ) + fake = FakeSubprocess(error=error) + monkeypatch.setattr(cli.subprocess, "run", fake) + result = cli.run_script("wait(999)") + + assert result.ok is False and result.timed_out is True + assert "partial" in result.stdout + assert "timed out" in result.failure() + + +def test_output_is_capped_so_one_page_cannot_flood_the_context( + harness_bin, monkeypatch +): + fake = FakeSubprocess(stdout="x" * (cli.MAX_CAPTURED_CHARS + 500)) + monkeypatch.setattr(cli.subprocess, "run", fake) + assert cli.run_script("print(1)").stdout.endswith("…") + + +def test_version_is_read_from_the_cli(monkeypatch): + monkeypatch.setattr(cli, "executable", lambda: "/tools/browser-harness") + fake = FakeSubprocess(stdout="0.1.10\n") + monkeypatch.setattr(cli.subprocess, "run", fake) + assert cli.version() == "0.1.10" + assert fake.argv == ["/tools/browser-harness", "--version"] + + +@pytest.mark.parametrize( + "stderr, needle", + [ + ( + "RuntimeError: permission-blocked: Chrome is reachable", + "mac-approve", + ), + ( + "remote debugging is turned off for this browser instance", + "chrome://inspect", + ), + ( + "DevToolsActivePort not found in [...]", + "/browser connect", + ), + ("chrome-not-running: no supported Chromium-family browser", "start one"), + ], +) +def test_known_harness_errors_carry_their_documented_fix(stderr, needle): + assert cli.fixup_for(stderr) is not None + assert needle.lower() in cli.fixup_for(stderr).lower() + + +def test_unknown_errors_get_no_invented_fix(): + assert cli.fixup_for("RuntimeError: selector matched nothing") is None diff --git a/tests/browser_harness/test_plugin.py b/tests/browser_harness/test_plugin.py new file mode 100644 index 0000000..e3cf28e --- /dev/null +++ b/tests/browser_harness/test_plugin.py @@ -0,0 +1,429 @@ +"""Consent gating, model tools, and the /browser slash command.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +import pytest +from pydantic_ai import BinaryContent, ToolReturn + +from code_puppy_core_plugins.browser_harness import browser +from code_puppy_core_plugins.browser_harness import cli, commands, tools +from code_puppy_core_plugins.browser_harness import register_callbacks as rc + +from ._helpers import FakeAgent, FakeSubprocess + +EMIT = "code_puppy_core_plugins.browser_harness.commands.emit_" +RC_EMIT = "code_puppy_core_plugins.browser_harness.register_callbacks.emit_info" + + +def _registered(register, name): + agent = FakeAgent() + register(agent) + return agent.registered[name] + + +def _installed(monkeypatch, fake=None): + monkeypatch.setattr(cli, "executable", lambda: "/tools/browser-harness") + monkeypatch.setattr(cli.subprocess, "run", fake or FakeSubprocess()) + return fake + + +# ── registration gating ───────────────────────────────────────── + + +def test_plugin_is_inert_until_the_user_opts_in(store): + assert rc._register_tools() == [] + assert rc._register_agent_tools("main") == [] + assert rc._register_skills() == [] + assert rc._load_prompt() is None + + store.set_enabled(True) + assert {item["name"] for item in rc._register_tools()} == set(tools.REGISTRARS) + assert set(rc._register_agent_tools()) == set(tools.REGISTRARS) + assert all(callable(item["register_func"]) for item in rc._register_tools()) + + skills = rc._register_skills() + assert [skill["name"] for skill in skills] == ["browser-harness"] + assert Path(skills[0]["skill_md_path"]).is_file() + assert "Firefox" in rc._load_prompt() + + +def test_startup_nudges_only_once_and_only_when_usable(store, monkeypatch): + monkeypatch.setattr(cli, "installed", lambda: False) + with patch(RC_EMIT) as info: + rc._startup() + assert info.call_args_list == [] + + monkeypatch.setattr(cli, "installed", lambda: True) + with patch(RC_EMIT) as info: + rc._startup() + assert "/browser enable" in info.call_args[0][0] + + store.set_enabled(True) + with patch(RC_EMIT) as info: + rc._startup() + assert info.call_args_list == [] + + +def test_a_broken_install_never_breaks_startup(monkeypatch): + monkeypatch.setattr( + cli, "installed", lambda: (_ for _ in ()).throw(RuntimeError("boom")) + ) + with patch(RC_EMIT) as info: + rc._startup() + assert info.call_args_list == [] + + +# ── tools ─────────────────────────────────────────────────────── + +TOOLS = "code_puppy_core_plugins.browser_harness.tools." + + +async def test_script_tool_returns_helper_stdout(store, monkeypatch): + store.set_enabled(True) + fake = _installed(monkeypatch, FakeSubprocess(stdout="Example Domain\n")) + tool = _registered(tools.register_browser_harness, "browser_harness") + + result = await tool(None, "print(page_info()['title'])") + + assert result == {"success": True, "output": "Example Domain"} + assert fake.calls[0][1]["input"] == "print(page_info()['title'])" + + +async def test_script_tool_refuses_without_consent(store, monkeypatch): + _installed(monkeypatch) + tool = _registered(tools.register_browser_harness, "browser_harness") + + result = await tool(None, "print(page_info())") + + assert result["success"] is False + assert "/browser enable" in result["error"] + + +async def test_script_tool_attaches_the_documented_fix(store, monkeypatch): + store.set_enabled(True) + _installed( + monkeypatch, + FakeSubprocess( + returncode=1, + stderr="RuntimeError: permission-blocked: Chrome is reachable", + ), + ) + tool = _registered(tools.register_browser_harness, "browser_harness") + + result = await tool(None, "print(page_info())") + + assert result["success"] is False + assert "mac-approve" in result["error"] + + +async def test_script_tool_teaches_the_install_when_it_is_missing(store, monkeypatch): + store.set_enabled(True) + monkeypatch.setattr(cli, "executable", lambda: None) + tool = _registered(tools.register_browser_harness, "browser_harness") + + assert "uv tool install" in (await tool(None, "print(1)"))["error"] + + +async def test_script_tool_clamps_a_silly_timeout(store, monkeypatch): + store.set_enabled(True) + fake = _installed(monkeypatch, FakeSubprocess(stdout="ok")) + tool = _registered(tools.register_browser_harness, "browser_harness") + + await tool(None, "print(1)", None, 1_000_000) + + assert fake.calls[0][1]["timeout"] == tools.MAX_TIMEOUT_SECONDS + + +async def test_screenshot_tool_shows_the_png(store, monkeypatch, tmp_path): + store.set_enabled(True) + png = tmp_path / "shot.png" + png.write_bytes(b"\x89PNG\r\n\x1a\n fake") + fake = _installed(monkeypatch, FakeSubprocess(stdout=f" {png} \n")) + monkeypatch.setattr(tools, "emit_inline_image", lambda path: True) + tool = _registered(tools.register_browser_screenshot, "browser_screenshot") + + result = await tool(None, True, None) + + assert isinstance(result, ToolReturn) + assert result.metadata["path"] == str(png) + assert result.metadata["displayed_inline"] is True + assert any(isinstance(part, BinaryContent) for part in result.content) + assert "full=True" in fake.calls[0][1]["input"] + assert "max_dim=None" in fake.calls[0][1]["input"] + + +async def test_screenshot_tool_defaults_to_a_resized_viewport( + store, monkeypatch, tmp_path +): + store.set_enabled(True) + png = tmp_path / "shot.png" + png.write_bytes(b"png") + fake = _installed(monkeypatch, FakeSubprocess(stdout=f"{png}\n")) + tool = _registered(tools.register_browser_screenshot, "browser_screenshot") + + result = await tool(None, False, tools.DEFAULT_SCREENSHOT_MAX_DIM) + + assert f"max_dim={tools.DEFAULT_SCREENSHOT_MAX_DIM}" in fake.calls[0][1]["input"] + assert result.metadata["displayed_inline"] is False + + +async def test_screenshot_tool_reports_a_missing_file(store, monkeypatch): + store.set_enabled(True) + _installed(monkeypatch, FakeSubprocess(stdout="/nowhere/shot.png\n")) + tool = _registered(tools.register_browser_screenshot, "browser_screenshot") + + result = await tool(None, False, None) + + assert "no screenshot file" in result["error"] + + +async def test_doctor_tool_reports_health(store, monkeypatch): + store.set_enabled(True) + fake = _installed(monkeypatch, FakeSubprocess(stdout="chrome running PASS\n")) + tool = _registered(tools.register_browser_doctor, "browser_doctor") + + result = await tool(None) + + assert result["healthy"] is True + assert fake.argv[-1] == "--doctor" + + +async def test_doctor_tool_flags_an_unhealthy_connection(store, monkeypatch): + store.set_enabled(True) + _installed( + monkeypatch, + FakeSubprocess( + returncode=1, + stdout="daemon alive FAIL", + stderr="permission-blocked: awaiting the Allow remote debugging popup", + ), + ) + tool = _registered(tools.register_browser_doctor, "browser_doctor") + + result = await tool(None) + + assert result["healthy"] is False + assert "daemon alive" in result["report"] + assert "mac-approve" in result["fix"] + + +# ── /browser ──────────────────────────────────────────────────── + + +def test_command_help_advertises_browser(): + assert [name for name, _ in commands.command_help()] == ["browser"] + + +def test_other_slash_commands_pass_through(): + assert commands.handle_command("/computer-use status", "computer-use") is None + + +def test_enable_and_disable_persist(store): + with patch(EMIT + "success"): + assert commands.handle_command("/browser enable", "browser") is True + assert store.is_enabled() is True + + with patch(EMIT + "warning"): + commands.handle_command("/browser disable", "browser") + assert store.is_enabled() is False + + +def test_enable_registers_the_tools_without_a_restart(store): + with ( + patch("code_puppy.tools.get_available_tool_names") as refresh, + patch(EMIT + "success"), + ): + commands.handle_command("/browser enable", "browser") + + assert refresh.call_args_list != [] + assert store.is_enabled() is True + + +def test_a_failed_refresh_still_records_consent(store): + broken = RuntimeError("registry unavailable") + with ( + patch("code_puppy.tools.get_available_tool_names", side_effect=broken), + patch(EMIT + "success"), + ): + commands.handle_command("/browser enable", "browser") + + assert store.is_enabled() is True + + +def test_unknown_subcommand_prints_usage(store): + with patch(EMIT + "error") as error: + commands.handle_command("/browser nope", "browser") + assert "/browser" in error.call_args[0][0] + + +def test_connect_requires_a_target(store): + with patch(EMIT + "error") as error: + commands.handle_command("/browser connect", "browser") + assert "devtools-url" in error.call_args[0][0] + + +def test_connect_rejects_a_non_devtools_url(store): + with patch(EMIT + "error") as error: + commands.handle_command("/browser connect localhost:9222", "browser") + assert "Invalid CDP endpoint" in error.call_args[0][0] + assert store.endpoint() is None + + +def test_connect_saves_even_while_the_browser_is_asleep(store, monkeypatch): + monkeypatch.setattr( + commands.browser, + "probe_endpoint", + lambda url: browser.Endpoint(url, reachable=False, product="unreachable"), + ) + with patch(EMIT + "warning") as warning: + commands.handle_command("/browser connect http://127.0.0.1:9222", "browser") + + assert store.endpoint() == "http://127.0.0.1:9222" + assert "nothing answered" in warning.call_args[0][0] + + +def test_connect_celebrates_a_live_endpoint(store, monkeypatch): + monkeypatch.setattr( + commands.browser, + "probe_endpoint", + lambda url: browser.Endpoint(url, reachable=True, product="Chrome/144"), + ) + with patch(EMIT + "success") as success: + commands.handle_command("/browser connect http://127.0.0.1:9222", "browser") + + assert "Chrome/144" in success.call_args[0][0] + + +def test_disconnect_returns_to_auto_discovery(store): + store.set_endpoint("http://127.0.0.1:9222") + with patch(EMIT + "success"): + commands.handle_command("/browser disconnect", "browser") + assert store.endpoint() is None + + +def test_recordings_rejects_an_unknown_action(store): + with patch(EMIT + "error") as error: + commands.handle_command("/browser recordings sideways", "browser") + assert "recordings" in error.call_args[0][0] + + +def test_recordings_passes_through_to_the_harness(store, monkeypatch): + fake = _installed(monkeypatch, FakeSubprocess(stdout="recordings: on\n")) + with patch(EMIT + "info") as info: + commands.handle_command("/browser recordings on", "browser") + + assert fake.argv[-2:] == ["recordings", "enable"] + assert "on" in info.call_args[0][0] + + +def test_doctor_command_prints_the_report(store, monkeypatch): + fake = _installed(monkeypatch, FakeSubprocess(stdout="all good\n")) + with patch(EMIT + "info"), patch(EMIT + "success") as success: + commands.handle_command("/browser doctor", "browser") + + assert fake.argv[-1] == "--doctor" + assert "healthy" in success.call_args[0][0] + + +def test_doctor_command_without_an_install(store, monkeypatch): + monkeypatch.setattr(cli, "executable", lambda: None) + with patch(EMIT + "error") as error: + commands.handle_command("/browser doctor", "browser") + assert "uv tool install" in error.call_args[0][0] + + +def test_status_is_honest_about_a_firefox_only_machine(store, monkeypatch): + monkeypatch.setattr(cli, "executable", lambda: None) + monkeypatch.setattr( + commands.browser, + "detect_browsers", + lambda: [ + browser.Browser("Firefox", "/Applications/Firefox.app", False, True), + ], + ) + monkeypatch.setattr(commands.browser, "reachable_endpoints", lambda extra=(): []) + + with patch(EMIT + "info") as info, patch(EMIT + "warning") as warning: + commands.handle_command("/browser status", "browser") + + messages = [call[0][0] for call in info.call_args_list] + [ + call[0][0] for call in warning.call_args_list + ] + joined = "\n".join(messages) + assert "not installed" in joined + assert "No drivable browser" in joined + assert "not drivable: Firefox" in joined + assert "awaiting your one-time consent" in joined + + +def test_status_lists_a_live_endpoint_and_a_running_browser(store, monkeypatch): + store.set_enabled(True) + monkeypatch.setattr(cli, "executable", lambda: "/tools/browser-harness") + monkeypatch.setattr(cli, "version", lambda: "0.1.10") + monkeypatch.setattr( + commands.browser, + "reachable_endpoints", + lambda extra=(): [ + browser.Endpoint("http://127.0.0.1:9222", True, "Chrome/144") + ], + ) + monkeypatch.setattr( + commands.browser, + "detect_browsers", + lambda: [browser.Browser("Chrome", "/Applications/Chrome", True, True)], + ) + + with patch(EMIT + "info") as info: + commands.handle_command("/browser status", "browser") + + joined = "\n".join(call[0][0] for call in info.call_args_list) + assert "0.1.10" in joined + assert "Chrome/144" in joined + assert "Chrome: running" in joined + assert "consent: enabled" in joined + + +def test_status_reports_an_ambient_endpoint_as_the_winning_one(store, monkeypatch): + monkeypatch.setattr(cli, "executable", lambda: "/tools/browser-harness") + monkeypatch.setattr(cli, "version", lambda: "0.1.10") + monkeypatch.setattr(cli, "ambient_endpoint", lambda: "wss://cloud.example/cdp") + monkeypatch.setattr(commands.browser, "reachable_endpoints", lambda extra=(): []) + monkeypatch.setattr(commands.browser, "detect_browsers", lambda: []) + + with patch(EMIT + "info") as info: + commands.handle_command("/browser status", "browser") + + joined = "\n".join(call[0][0] for call in info.call_args_list) + assert "wss://cloud.example/cdp (from BU_CDP_WS/BU_CDP_URL)" in joined + + +def test_status_surfaces_a_broken_override(monkeypatch): + monkeypatch.setattr( + cli, + "executable", + lambda: (_ for _ in ()).throw( + cli.policy.BrowserHarnessError("not an executable") + ), + ) + with patch(EMIT + "error") as error: + commands.handle_command("/browser status", "browser") + assert "not an executable" in error.call_args[0][0] + + +def test_install_help_lists_drivable_browsers(store, monkeypatch): + monkeypatch.setattr(browser.platform, "system", lambda: "Darwin") + with patch(EMIT + "info") as info: + commands.handle_command("/browser install", "browser") + + joined = info.call_args[0][0] + assert "brew install --cask google-chrome" in joined + assert "firefox" not in joined.casefold() + + +@pytest.mark.parametrize("name", sorted(tools.REGISTRARS)) +def test_every_tool_name_is_registered_once_and_matches_its_function(name): + registered = _registered(tools.REGISTRARS[name], name) + assert registered.__name__ == name diff --git a/tests/browser_harness/test_policy.py b/tests/browser_harness/test_policy.py new file mode 100644 index 0000000..0f21812 --- /dev/null +++ b/tests/browser_harness/test_policy.py @@ -0,0 +1,75 @@ +"""Consent and connection settings for the browser-harness plugin.""" + +from __future__ import annotations + +import json + +import pytest + +from code_puppy_core_plugins.browser_harness.policy import ( + BrowserHarnessError, + SettingsStore, +) + + +def test_consent_starts_unset_and_blocks(store): + assert store.is_enabled() is False + assert store.consent_state() == "unset" + with pytest.raises(BrowserHarnessError, match="one-time permission"): + store.require_enabled() + + +def test_unset_consent_offers_the_enable_command(store): + with pytest.raises(BrowserHarnessError, match=r"/browser enable"): + store.require_enabled() + + +def test_declined_consent_gets_its_own_message(store): + store.set_enabled(False) + assert store.consent_state() == "disabled" + assert store.is_enabled() is False + with pytest.raises(BrowserHarnessError, match="disabled in settings"): + store.require_enabled() + + +def test_opted_in_consent_persists(store): + store.set_enabled(True) + assert store.is_enabled() is True + store.require_enabled() # must not raise + + +def test_corrupt_settings_revert_to_asking_consent(tmp_path): + path = tmp_path / "policy.json" + path.write_text("{not json") + assert SettingsStore(path).consent_state() == "unset" + + +def test_values_of_the_wrong_type_are_ignored(tmp_path): + path = tmp_path / "policy.json" + path.write_text(json.dumps({"enabled": "yes please", "endpoint": 42})) + store = SettingsStore(path) + assert store.consent_state() == "unset" + assert store.endpoint() is None + + +def test_endpoint_must_be_a_devtools_url(store): + with pytest.raises(BrowserHarnessError, match="Invalid CDP endpoint"): + store.set_endpoint("localhost:9222") + assert store.endpoint() is None + + +def test_endpoint_is_normalised_and_attributed(store): + store.set_endpoint(" http://127.0.0.1:9222/ ") + assert store.endpoint() == "http://127.0.0.1:9222" + assert store.status()["endpoint_source"] == "saved" + + store.set_endpoint("wss://browser.example/devtools") + assert store.endpoint() == "wss://browser.example/devtools" + + +def test_endpoint_defaults_to_auto_discovery(store): + assert store.status()["endpoint_source"] == "auto-discovery" + store.set_endpoint("http://127.0.0.1:9222") + store.clear_endpoint() + assert store.endpoint() is None + assert store.status()["endpoint_source"] == "auto-discovery"