"""Optional, workspace-scoped browser control for the tool executor.""" from __future__ import annotations import atexit import hashlib import importlib import ipaddress import json import os import queue import secrets import select import shutil import socket import socketserver import threading import time import urllib.parse import warnings from collections import deque from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Final, Mapping from .contracts import ( SourceTrust, ToolCall, ToolExecutionContext, ToolExecutionResult, ToolParameter, ToolSpec, ) from .sandbox import control_root _DEFAULT_VIEWPORT: Final = {"width": 1280, "height": 720} _VISIBLE_ELEMENT_LIMIT: Final = 256 _VISIBLE_ELEMENT_SCAN_LIMIT: Final = 2048 _EVENT_HISTORY_LIMIT: Final = 512 _EVENT_TEXT_LIMIT: Final = 8192 _STORAGE_CHECKPOINT_SCHEMA: Final = "nexum.browser-storage.v1" _INTERACTIVE_SELECTOR: Final = ", ".join( ( "a", "button", "input", "textarea", "select", "summary", "[role]", "h1", "h2", "h3", "h4", "h5", "h6", "p", "li", "table", "form", "nav", "main", "section", ) ) BROWSER_TOOL_SPECS: tuple[ToolSpec, ...] = ( ToolSpec( "BrowserStatus", "browser", "Report whether optional browser control is available.", "BrowserStatus()", ), ToolSpec( "BrowserNavigate", "browser", "Navigate the current browser session to a public URL or a workspace file URL.", "BrowserNavigate(url='https://example.com')", (ToolParameter("url", "string", "Validated destination URL."),), parallel_safe=False, source_trust="untrusted_content", ), ToolSpec( "BrowserSnapshot", "browser", "Return viewport metadata and visible DOM and accessibility state.", "BrowserSnapshot(selector='body')", ( ToolParameter( "selector", "string", "Optional CSS scope for the snapshot.", required=False, ), ), parallel_safe=False, source_trust="untrusted_content", ), ToolSpec( "BrowserClick", "browser", "Click one exact element selected by CSS.", "BrowserClick(selector='#submit')", (ToolParameter("selector", "string", "Exact CSS selector."),), risk="external_effect", parallel_safe=False, idempotent=False, source_trust="untrusted_content", ), ToolSpec( "BrowserFill", "browser", "Replace the value of one exact editable element.", "BrowserFill(selector='#name', value='Ada')", ( ToolParameter("selector", "string", "Exact CSS selector."), ToolParameter("value", "string", "Exact value to fill."), ), risk="external_effect", parallel_safe=False, source_trust="untrusted_content", ), ToolSpec( "BrowserType", "browser", "Type exact text into one exact editable element without clearing it.", "BrowserType(selector='#name', value=' Lovelace')", ( ToolParameter("selector", "string", "Exact CSS selector."), ToolParameter("value", "string", "Exact text to type."), ), risk="external_effect", parallel_safe=False, idempotent=False, source_trust="untrusted_content", ), ToolSpec( "BrowserScreenshot", "browser", "Capture the current page to a PNG contained in the workspace.", "BrowserScreenshot(path='artifacts/page.png')", ( ToolParameter( "path", "string", "Optional workspace-relative PNG path.", required=False, ), ToolParameter( "full_page", "boolean", "Capture the full document instead of only the viewport.", required=False, ), ), risk="workspace_write", parallel_safe=False, source_trust="untrusted_content", ), ToolSpec( "BrowserState", "browser", "Return current page, console, error, and network observations.", "BrowserState(clear=False)", ( ToolParameter( "clear", "boolean", "Clear captured observations after reading them.", required=False, ), ), parallel_safe=False, source_trust="untrusted_content", ), ToolSpec( "BrowserVerify", "browser", "Verify expected visibility and text for one exact CSS selector.", "BrowserVerify(selector='#status', visible=True, text='Saved')", ( ToolParameter("selector", "string", "Exact CSS selector."), ToolParameter( "visible", "boolean", "Expected visibility.", required=False, ), ToolParameter( "text", "string", "Expected text or text fragment.", required=False, ), ToolParameter( "exact", "boolean", "Require exact text equality.", required=False, ), ), parallel_safe=False, source_trust="untrusted_content", ), ToolSpec( "BrowserClose", "browser", "Close the current workspace browser session.", "BrowserClose()", parallel_safe=False, idempotent=True, ), ) BROWSER_TOOL_NAMES: frozenset[str] = frozenset(spec.name for spec in BROWSER_TOOL_SPECS) _BROWSER_TOOL_SPEC_BY_NAME: Final = {spec.name: spec for spec in BROWSER_TOOL_SPECS} class BrowserUnavailableError(RuntimeError): """Raised when the optional browser dependency cannot be used.""" class BrowserTargetError(ValueError): """Raised when a page target leaves the browser security boundary.""" @dataclass(frozen=True) class BrowserAvailability: available: bool package_available: bool browser_available: bool browser: str = "" source: str = "" reason: str = "" executable_path: str = field(default="", repr=False) def public_dict(self) -> dict[str, object]: payload = asdict(self) payload.pop("executable_path", None) return payload @dataclass(frozen=True) class BrowserActionResult: ok: bool payload: Mapping[str, object] error: str = "" source_trust: SourceTrust = "untrusted_content" @dataclass class _BrowserCommand: name: str arguments: dict[str, Any] workspace: Path session_key: str timeout_s: float done: threading.Event = field(default_factory=threading.Event) result: BrowserActionResult | None = None error: Exception | None = None def _load_sync_api() -> Any: try: return importlib.import_module("playwright.sync_api") except (ImportError, ModuleNotFoundError) as exc: raise BrowserUnavailableError( "Playwright is not installed; browser tools are unavailable" ) from exc def _system_browser() -> str: for name in ("chromium", "google-chrome", "chrome", "chromium-browser"): executable = shutil.which(name) if executable: return executable return "" def browser_availability() -> BrowserAvailability: """Inspect the optional dependency without launching a browser context.""" try: sync_api = _load_sync_api() except BrowserUnavailableError as exc: return BrowserAvailability(False, False, False, reason=str(exc)) playwright = None result: BrowserAvailability try: playwright = sync_api.sync_playwright().start() managed = Path(str(playwright.chromium.executable_path)).expanduser() if managed.is_file(): result = BrowserAvailability( True, True, True, browser="chromium", source="playwright-managed", executable_path=str(managed.resolve()), ) else: system = _system_browser() if system: result = BrowserAvailability( True, True, True, browser="chromium", source="system", executable_path=system, ) else: result = BrowserAvailability( False, True, False, reason=( "Playwright is installed but no Chromium browser executable " "is available" ), ) except Exception as exc: # noqa: BLE001 - optional dependency boundary result = BrowserAvailability( False, True, False, reason=f"Playwright could not initialize: {exc}", ) if playwright is not None: try: playwright.stop() except Exception as exc: # noqa: BLE001 - optional dependency cleanup return BrowserAvailability( False, True, False, reason=f"Playwright availability cleanup failed: {exc}", ) return result def browser_status() -> dict[str, object]: """Return release-safe browser availability metadata.""" return browser_availability().public_dict() def _workspace_root(workspace: str | Path) -> Path: root = Path(workspace).expanduser().resolve() if not root.is_dir(): raise ValueError("browser workspace does not exist or is not a directory") return root def _workspace_path( path: str, workspace: Path, *, allow_control: bool = False ) -> Path: candidate = Path(path).expanduser() target = ( candidate.resolve() if candidate.is_absolute() else (workspace / candidate).resolve() ) try: relative = target.relative_to(workspace) except ValueError as exc: raise BrowserTargetError("browser path leaves the workspace") from exc if ( not allow_control and relative.parts and relative.parts[0] == ".nexum" ): raise BrowserTargetError( "workspace control paths require dedicated runtime tools" ) return target def _public_addresses(hostname: str, port: int) -> tuple[str, ...]: try: addresses = {ipaddress.ip_address(hostname)} except ValueError: try: rows = socket.getaddrinfo( hostname, port, type=socket.SOCK_STREAM, ) except socket.gaierror as exc: raise BrowserTargetError( "browser target hostname could not be resolved" ) from exc addresses = { ipaddress.ip_address(str(row[4][0]).split("%", maxsplit=1)[0]) for row in rows } if not addresses or any(not address.is_global for address in addresses): raise BrowserTargetError( "browser HTTP targets must resolve only to public addresses" ) return tuple(sorted(str(address) for address in addresses)) def _proxy_target(target: str) -> tuple[str, int]: if target.startswith("["): end = target.find("]") if end < 0 or end + 2 > len(target) or target[end + 1] != ":": raise BrowserTargetError("browser proxy target is invalid") hostname = target[1:end] raw_port = target[end + 2 :] else: hostname, separator, raw_port = target.rpartition(":") if not separator: raise BrowserTargetError("browser proxy target requires a port") try: port = int(raw_port) except ValueError as exc: raise BrowserTargetError("browser proxy target port is invalid") from exc if not hostname or not 0 < port < 65536: raise BrowserTargetError("browser proxy target is invalid") return hostname, port def _connect_public_target(hostname: str, port: int) -> socket.socket: last_error: OSError | None = None for address in _public_addresses(hostname, port): try: return socket.create_connection((address, port), timeout=30.0) except OSError as exc: last_error = exc raise ConnectionError("browser proxy could not connect to target") from last_error def _proxy_header(request: bytes) -> tuple[bytes, bytes, str, str]: marker = b"\r\n\r\n" if marker not in request: raise BrowserTargetError("browser proxy request header is incomplete") header, remainder = request.split(marker, maxsplit=1) lines = header.decode("iso-8859-1").split("\r\n") if not lines: raise BrowserTargetError("browser proxy request is empty") parts = lines[0].split(" ", maxsplit=2) if len(parts) != 3: raise BrowserTargetError("browser proxy request line is invalid") method, target, version = parts if method.upper() == "CONNECT": return b"", remainder, method.upper(), target parsed = urllib.parse.urlsplit(target) if parsed.scheme.lower() != "http" or not parsed.hostname: raise BrowserTargetError("browser proxy accepts absolute HTTP targets") if parsed.username is not None or parsed.password is not None: raise BrowserTargetError("browser proxy URL credentials are not allowed") try: port = parsed.port or 80 except ValueError as exc: raise BrowserTargetError("browser proxy target port is invalid") from exc _public_addresses(parsed.hostname, port) display_host = parsed.hostname.encode("idna").decode("ascii") if ":" in display_host: display_host = f"[{display_host}]" host_header = display_host if port == 80 else f"{display_host}:{port}" path = parsed.path or "/" if parsed.query: path += "?" + parsed.query retained = [ line for line in lines[1:] if line and line.partition(":")[0].strip().lower() not in {"host", "proxy-connection"} ] rewritten = "\r\n".join( [f"{method} {path} {version}", f"Host: {host_header}", *retained, "", ""] ).encode("iso-8859-1") return rewritten, remainder, method.upper(), f"{parsed.hostname}:{port}" class _PinnedProxyHandler(socketserver.BaseRequestHandler): def handle(self) -> None: client = self.request if not isinstance(client, socket.socket): return header = b"" try: while b"\r\n\r\n" not in header: chunk = client.recv(65536) if not chunk: return header += chunk if len(header) > 262144: raise BrowserTargetError("browser proxy request header is too large") rewritten, remainder, method, target = _proxy_header(header) hostname, port = _proxy_target(target) upstream = _connect_public_target(hostname, port) try: if method == "CONNECT": client.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n") else: upstream.sendall(rewritten) if remainder: upstream.sendall(remainder) self._relay(client, upstream) finally: upstream.close() except (BrowserTargetError, ConnectionError, OSError): try: client.sendall( b"HTTP/1.1 403 Forbidden\r\nConnection: close\r\n" b"Content-Length: 0\r\n\r\n" ) except OSError: return @staticmethod def _relay(client: socket.socket, upstream: socket.socket) -> None: sockets = (client, upstream) for current in sockets: current.settimeout(None) while True: readable, _, exceptional = select.select(sockets, (), sockets) if exceptional: return for source in readable: data = source.recv(65536) if not data: return destination = upstream if source is client else client destination.sendall(data) class _PinnedBrowserProxy: def __init__(self) -> None: server_type = type( "NexumPinnedProxyServer", (socketserver.ThreadingTCPServer,), {"allow_reuse_address": True, "daemon_threads": True}, ) self._server = server_type(("127.0.0.1", 0), _PinnedProxyHandler) self._thread = threading.Thread( target=self._server.serve_forever, name="nexum-browser-proxy", daemon=True, ) self._thread.start() @property def server_url(self) -> str: host, port = self._server.server_address[:2] return f"http://{host}:{port}" def close(self) -> None: self._server.shutdown() self._server.server_close() self._thread.join(timeout=5.0) def validate_browser_target(url: str, workspace: str | Path) -> str: """Validate one navigation or request target without changing it.""" root = _workspace_root(workspace) if not url or url != url.strip(): raise BrowserTargetError("browser target must be a non-empty URL") if "\\" in url or any(ord(character) < 32 for character in url): raise BrowserTargetError("browser target contains invalid characters") parsed = urllib.parse.urlsplit(url) scheme = parsed.scheme.lower() if scheme in {"http", "https"}: if parsed.username is not None or parsed.password is not None: raise BrowserTargetError("browser URL credentials are not allowed") hostname = parsed.hostname if not hostname: raise BrowserTargetError("browser HTTP target requires a hostname") try: port = parsed.port or (443 if scheme == "https" else 80) except ValueError as exc: raise BrowserTargetError("browser target port is invalid") from exc _public_addresses(hostname, port) return url if scheme == "file": if parsed.netloc: raise BrowserTargetError("browser file URLs cannot include a host") if parsed.query: raise BrowserTargetError("browser file URLs cannot include a query") path = Path(urllib.parse.unquote(parsed.path)).expanduser().resolve() try: relative = path.relative_to(root) except ValueError as exc: raise BrowserTargetError( "browser file target must stay inside the workspace" ) from exc if relative.parts and relative.parts[0] == ".nexum": raise BrowserTargetError( "workspace control paths require dedicated runtime tools" ) if not path.is_file(): raise BrowserTargetError("browser file target does not exist") return url raise BrowserTargetError( "browser navigation accepts only public HTTP(S) or workspace file URLs" ) def _display_url(url: str, workspace: Path) -> str: parsed = urllib.parse.urlsplit(url) scheme = parsed.scheme.lower() if scheme == "file": try: path = Path(urllib.parse.unquote(parsed.path)).resolve() relative = path.relative_to(workspace).as_posix() return f"workspace:///{urllib.parse.quote(relative)}" except ValueError: return "blocked://outside-workspace" if scheme in {"http", "https"}: hostname = parsed.hostname or "" netloc = hostname try: port = parsed.port except ValueError: port = None if port is not None: netloc = f"{hostname}:{port}" return urllib.parse.urlunsplit((scheme, netloc, parsed.path or "/", "", "")) return f"{scheme or 'unknown'}:" def _event_text(value: object) -> str: text = str(value).replace("\x00", "") if len(text) <= _EVENT_TEXT_LIMIT: return text return text[:_EVENT_TEXT_LIMIT] + "...[truncated]" def _session_digest(session_id: str) -> str: value = session_id if session_id else "default" return hashlib.sha256(value.encode("utf-8")).hexdigest()[:32] class _BrowserSession: def __init__( self, *, playwright: Any, executable_path: str, proxy_url: str, workspace: Path, session_key: str, timeout_s: float, ) -> None: self.workspace = workspace self.session_key = session_key self.console_events: deque[dict[str, object]] = deque( maxlen=_EVENT_HISTORY_LIMIT ) self.network_events: deque[dict[str, object]] = deque( maxlen=_EVENT_HISTORY_LIMIT ) self.page_errors: deque[dict[str, object]] = deque(maxlen=_EVENT_HISTORY_LIMIT) self._attached_pages: set[int] = set() profile = control_root(workspace) / "browser" / "sessions" / session_key artifacts = _workspace_path( f".nexum-output/browser/{session_key}", workspace, ) self.artifact_root = artifacts profile.mkdir(parents=True, exist_ok=True, mode=0o700) artifacts.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(profile, 0o700) os.chmod(artifacts, 0o700) self._storage_checkpoint = profile / "storage.json" self._local_storage = self._load_local_storage() timeout_ms = self._timeout_ms(timeout_s) self.context = playwright.chromium.launch_persistent_context( profile, executable_path=executable_path or None, headless=True, handle_sigint=False, handle_sigterm=False, handle_sighup=False, timeout=timeout_ms, viewport=dict(_DEFAULT_VIEWPORT), accept_downloads=False, downloads_path=artifacts, artifacts_dir=artifacts, ignore_https_errors=False, java_script_enabled=True, bypass_csp=False, permissions=(), proxy={"server": proxy_url, "bypass": "<-loopback>"}, service_workers="block", strict_selectors=True, ) self._install_local_storage_restore() self.context.clear_permissions() self.context.route("**/*", self._route_request) self.context.on("page", self._attach_page) pages = tuple(self.context.pages) if pages: for page in pages: self._attach_page(page) self.page = pages[-1] else: self.page = self.context.new_page() self._attach_page(self.page) self._apply_timeout(timeout_s) def _load_local_storage(self) -> dict[str, dict[str, str]]: checkpoint = self._storage_checkpoint if not checkpoint.exists(): return {} if checkpoint.is_symlink() or not checkpoint.is_file(): raise BrowserTargetError( "browser storage checkpoint must be a regular file" ) try: payload = json.loads(checkpoint.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise BrowserTargetError( "browser storage checkpoint is unreadable" ) from exc if not isinstance(payload, dict): raise BrowserTargetError("browser storage checkpoint is invalid") if payload.get("schema") != _STORAGE_CHECKPOINT_SCHEMA: raise BrowserTargetError("browser storage checkpoint schema is invalid") raw_entries = payload.get("entries") if not isinstance(raw_entries, dict): raise BrowserTargetError("browser storage checkpoint entries are invalid") entries: dict[str, dict[str, str]] = {} for storage_key, raw_values in raw_entries.items(): if not isinstance(storage_key, str) or not isinstance(raw_values, dict): raise BrowserTargetError( "browser storage checkpoint entry is invalid" ) values: dict[str, str] = {} for name, value in raw_values.items(): if not isinstance(name, str) or not isinstance(value, str): raise BrowserTargetError( "browser storage checkpoint value is invalid" ) values[name] = value entries[storage_key] = values return entries def _install_local_storage_restore(self) -> None: serialized = json.dumps( self._local_storage, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ) encoded = json.dumps(serialized, ensure_ascii=True) self.context.add_init_script( script=f""" (() => {{ const entries = JSON.parse({encoded}); const storageKey = location.protocol === "file:" ? location.href.split(/[?#]/, 1)[0] : location.origin; if (!Object.prototype.hasOwnProperty.call(entries, storageKey)) return; for (const [name, value] of Object.entries(entries[storageKey])) {{ localStorage.setItem(name, value); }} }})(); """ ) def _capture_local_storage(self) -> None: parsed = urllib.parse.urlsplit(str(self.page.url)) if parsed.scheme.lower() not in {"file", "http", "https"}: return payload = self.page.evaluate( """ () => { const storageKey = location.protocol === "file:" ? location.href.split(/[?#]/, 1)[0] : location.origin; return { storageKey, entries: Object.entries(localStorage), }; } """ ) if not isinstance(payload, dict): raise RuntimeError("browser local storage capture returned invalid state") storage_key = payload.get("storageKey") raw_entries = payload.get("entries") if not isinstance(storage_key, str) or not isinstance(raw_entries, list): raise RuntimeError("browser local storage capture returned invalid entries") values: dict[str, str] = {} for row in raw_entries: if ( not isinstance(row, list) or len(row) != 2 or not isinstance(row[0], str) or not isinstance(row[1], str) ): raise RuntimeError( "browser local storage capture returned an invalid value" ) values[row[0]] = row[1] self._local_storage[storage_key] = values def _persist_local_storage(self) -> None: checkpoint = self._storage_checkpoint payload = { "schema": _STORAGE_CHECKPOINT_SCHEMA, "entries": self._local_storage, } temporary = checkpoint.with_name( f".{checkpoint.name}.{secrets.token_hex(8)}.tmp" ) try: with temporary.open("x", encoding="utf-8") as handle: json.dump( payload, handle, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.chmod(temporary, 0o600) os.replace(temporary, checkpoint) directory_fd = os.open(checkpoint.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(directory_fd) finally: os.close(directory_fd) finally: temporary.unlink(missing_ok=True) @staticmethod def _timeout_ms(timeout_s: float) -> float: if timeout_s < 0: raise ValueError("browser timeout must be non-negative") return timeout_s * 1000.0 def _apply_timeout(self, timeout_s: float) -> float: timeout_ms = self._timeout_ms(timeout_s) self.page.set_default_timeout(timeout_ms) self.page.set_default_navigation_timeout(timeout_ms) return timeout_ms def _append_network(self, event: str, request: Any, **extra: object) -> None: payload: dict[str, object] = { "event": event, "method": _event_text(getattr(request, "method", "")), "resource_type": _event_text(getattr(request, "resource_type", "")), "url": _display_url(str(getattr(request, "url", "")), self.workspace), } payload.update(extra) self.network_events.append(payload) def _route_request(self, route: Any, request: Any) -> None: url = str(request.url) parsed = urllib.parse.urlsplit(url) scheme = parsed.scheme.lower() try: if scheme in {"http", "https", "file"}: validate_browser_target(url, self.workspace) elif request.is_navigation_request(): raise BrowserTargetError( "browser navigation left the allowed target schemes" ) elif scheme not in {"about", "blob", "data"}: raise BrowserTargetError("browser subrequest scheme is not allowed") except BrowserTargetError as exc: self._append_network( "blocked", request, reason=_event_text(exc), ) route.abort("blockedbyclient") return route.continue_() def _attach_page(self, page: Any) -> None: self.page = page page_id = id(page) if page_id in self._attached_pages: return self._attached_pages.add(page_id) page.on( "console", lambda message: self.console_events.append( { "type": _event_text(message.type), "text": _event_text(message.text), } ), ) page.on( "pageerror", lambda error: self.page_errors.append( {"type": "pageerror", "text": _event_text(error)} ), ) page.on( "request", lambda request: self._append_network("request", request), ) page.on( "response", lambda response: self._append_network( "response", response.request, status=int(response.status), ), ) page.on( "requestfailed", lambda request: self._append_network( "failed", request, reason=_event_text(request.failure or "request failed"), ), ) def dismiss_dialog(dialog: Any) -> None: self.page_errors.append( { "type": "dialog", "text": _event_text(dialog.message), } ) dialog.dismiss() page.on("dialog", dismiss_dialog) def _page_summary(self) -> dict[str, object]: return { "url": _display_url(str(self.page.url), self.workspace), "title": _event_text(self.page.title()), } def _exact_locator(self, selector: str) -> Any: if not selector: raise ValueError("selector is required") locator = self.page.locator(selector) count = locator.count() if count != 1: raise ValueError( f"selector must match exactly one element; matched {count}" ) return locator def _navigate( self, arguments: Mapping[str, Any], timeout_ms: float ) -> BrowserActionResult: url = str(arguments["url"]) validate_browser_target(url, self.workspace) self._capture_local_storage() response = self.page.goto( url, wait_until="domcontentloaded", timeout=timeout_ms, ) validate_browser_target(str(self.page.url), self.workspace) payload = self._page_summary() payload["status"] = int(response.status) if response is not None else None return BrowserActionResult(True, payload) def _visible_element( self, locator: Any, viewport: Mapping[str, int] ) -> dict[str, object] | None: if not locator.is_visible(): return None box = locator.bounding_box() if box is None: return None width = int(viewport["width"]) height = int(viewport["height"]) if ( float(box["x"]) + float(box["width"]) <= 0 or float(box["y"]) + float(box["height"]) <= 0 or float(box["x"]) >= width or float(box["y"]) >= height ): return None attributes: dict[str, str] = {} for name in ( "id", "role", "name", "type", "placeholder", "title", "href", "aria-label", ): value = locator.get_attribute(name) if value is not None: attributes[name] = _event_text(value) try: text = _event_text(locator.inner_text()) except (RuntimeError, ValueError): text = "" return { "box": { "x": round(float(box["x"]), 2), "y": round(float(box["y"]), 2), "width": round(float(box["width"]), 2), "height": round(float(box["height"]), 2), }, "attributes": attributes, "text": text, "accessibility": _event_text(locator.aria_snapshot()), } def _snapshot(self, arguments: Mapping[str, Any]) -> BrowserActionResult: selector = str(arguments.get("selector") or "body") scope = self._exact_locator(selector) viewport = self.page.viewport_size or dict(_DEFAULT_VIEWPORT) candidates = scope.locator(_INTERACTIVE_SELECTOR) candidate_count = candidates.count() visible: list[dict[str, object]] = [] scanned = 0 for index in range(min(candidate_count, _VISIBLE_ELEMENT_SCAN_LIMIT)): scanned += 1 item = self._visible_element(candidates.nth(index), viewport) if item is not None: item["dom_index"] = index visible.append(item) if len(visible) >= _VISIBLE_ELEMENT_LIMIT: break payload = self._page_summary() payload.update( { "selector": selector, "viewport": { "width": int(viewport["width"]), "height": int(viewport["height"]), }, "candidate_count": candidate_count, "scanned": scanned, "truncated": ( scanned < candidate_count or len(visible) >= _VISIBLE_ELEMENT_LIMIT ), "visible_elements": visible, } ) return BrowserActionResult(True, payload) def _click( self, arguments: Mapping[str, Any], timeout_ms: float ) -> BrowserActionResult: selector = str(arguments["selector"]) self._exact_locator(selector).click(timeout=timeout_ms) payload = self._page_summary() payload["selector"] = selector return BrowserActionResult(True, payload) def _fill( self, arguments: Mapping[str, Any], timeout_ms: float ) -> BrowserActionResult: selector = str(arguments["selector"]) value = str(arguments["value"]) self._exact_locator(selector).fill(value, timeout=timeout_ms) return BrowserActionResult( True, {"selector": selector, "characters": len(value)}, ) def _type( self, arguments: Mapping[str, Any], timeout_ms: float ) -> BrowserActionResult: selector = str(arguments["selector"]) value = str(arguments["value"]) self._exact_locator(selector).press_sequentially( value, delay=0, timeout=timeout_ms, ) return BrowserActionResult( True, {"selector": selector, "characters": len(value)}, ) def _screenshot(self, arguments: Mapping[str, Any]) -> BrowserActionResult: requested = str(arguments.get("path") or "") if requested: target = _workspace_path(requested, self.workspace) else: target = _workspace_path( str( self.artifact_root / f"{time.time_ns()}-{secrets.token_hex(4)}.png" ), self.workspace, allow_control=True, ) if target.suffix.lower() != ".png": raise ValueError("browser screenshots require a .png path") target.parent.mkdir(parents=True, exist_ok=True) data = self.page.screenshot( path=target, full_page=bool(arguments.get("full_page", False)), type="png", ) target.chmod(0o600) relative = target.relative_to(self.workspace).as_posix() return BrowserActionResult( True, { "path": relative, "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest(), }, ) def _state(self, arguments: Mapping[str, Any]) -> BrowserActionResult: payload = self._page_summary() payload.update( { "console": list(self.console_events), "page_errors": list(self.page_errors), "network": list(self.network_events), } ) if bool(arguments.get("clear", False)): self.console_events.clear() self.page_errors.clear() self.network_events.clear() return BrowserActionResult(True, payload) def _verify(self, arguments: Mapping[str, Any]) -> BrowserActionResult: if "visible" not in arguments and "text" not in arguments: raise ValueError("BrowserVerify requires visible or text") selector = str(arguments["selector"]) locator = self.page.locator(selector) count = locator.count() if count > 1: raise ValueError( f"selector must match at most one element; matched {count}" ) visible = count == 1 and locator.is_visible() checks: dict[str, bool] = {} if "visible" in arguments: checks["visible"] = visible is bool(arguments["visible"]) actual_text = "" if "text" in arguments: if count == 1: actual_text = _event_text(locator.inner_text()) expected_text = str(arguments["text"]) checks["text"] = ( actual_text == expected_text if bool(arguments.get("exact", False)) else expected_text in actual_text ) passed = all(checks.values()) payload: dict[str, object] = { "selector": selector, "matched": count, "visible": visible, "checks": checks, } if "text" in arguments: payload["actual_text"] = actual_text return BrowserActionResult( passed, payload, error="" if passed else "browser verification did not match", ) def execute( self, name: str, arguments: Mapping[str, Any], timeout_s: float, ) -> BrowserActionResult: timeout_ms = self._apply_timeout(timeout_s) if name == "BrowserNavigate": return self._navigate(arguments, timeout_ms) if name == "BrowserSnapshot": return self._snapshot(arguments) if name == "BrowserClick": return self._click(arguments, timeout_ms) if name == "BrowserFill": return self._fill(arguments, timeout_ms) if name == "BrowserType": return self._type(arguments, timeout_ms) if name == "BrowserScreenshot": return self._screenshot(arguments) if name == "BrowserState": return self._state(arguments) if name == "BrowserVerify": return self._verify(arguments) raise ValueError(f"unsupported browser action: {name}") def close(self) -> None: checkpoint_error: Exception | None = None try: self._capture_local_storage() self._persist_local_storage() except Exception as exc: # noqa: BLE001 - close still releases browser resources checkpoint_error = exc try: self.context.close() except Exception as exc: # noqa: BLE001 - preserve checkpoint failure if present if checkpoint_error is None: checkpoint_error = exc if checkpoint_error is not None: raise RuntimeError( f"browser session close did not persist cleanly: {checkpoint_error}" ) from checkpoint_error class _BrowserService: """Own Playwright objects on one thread while serving persistent sessions.""" def __init__(self) -> None: self._commands: queue.Queue[_BrowserCommand | None] = queue.Queue() self._thread = threading.Thread( target=self._run, name="nexum-browser-service", daemon=True, ) self._start_lock = threading.Lock() self._started = False self._stopped = False def _ensure_started(self) -> None: with self._start_lock: if self._stopped: raise BrowserUnavailableError("browser service has been closed") if not self._started: self._thread.start() self._started = True def submit(self, command: _BrowserCommand) -> BrowserActionResult: self._ensure_started() self._commands.put(command) command.done.wait() if command.error is not None: raise command.error if command.result is None: raise RuntimeError("browser service returned no result") return command.result def _run(self) -> None: playwright = None proxy: _PinnedBrowserProxy | None = None sessions: dict[tuple[Path, str], _BrowserSession] = {} startup_error: Exception | None = None try: availability = browser_availability() if not availability.available: startup_error = BrowserUnavailableError(availability.reason) else: try: sync_api = _load_sync_api() playwright = sync_api.sync_playwright().start() proxy = _PinnedBrowserProxy() except Exception as exc: # noqa: BLE001 - dependency boundary startup_error = BrowserUnavailableError( f"Playwright could not initialize: {exc}" ) while True: command = self._commands.get() if command is None: break try: if startup_error is not None: raise startup_error key = (command.workspace, command.session_key) if command.name == "BrowserClose": session = sessions.pop(key, None) if session is not None: session.close() command.result = BrowserActionResult( True, {"closed": session is not None}, source_trust="trusted_execution", ) else: session = sessions.get(key) if session is None: if playwright is None: raise BrowserUnavailableError( "Playwright did not initialize" ) session = _BrowserSession( playwright=playwright, executable_path=availability.executable_path, proxy_url=proxy.server_url if proxy is not None else "", workspace=command.workspace, session_key=command.session_key, timeout_s=command.timeout_s, ) sessions[key] = session command.result = session.execute( command.name, command.arguments, command.timeout_s, ) except Exception as exc: # noqa: BLE001 - executor boundary returns failures command.error = exc finally: command.done.set() finally: for session in sessions.values(): try: session.close() except Exception as exc: # noqa: BLE001 - shutdown boundary warnings.warn( f"browser session cleanup failed: {exc}", RuntimeWarning, stacklevel=2, ) if playwright is not None: try: playwright.stop() except Exception as exc: # noqa: BLE001 - shutdown boundary warnings.warn( f"Playwright cleanup failed: {exc}", RuntimeWarning, stacklevel=2, ) if proxy is not None: proxy.close() def shutdown(self) -> None: with self._start_lock: if self._stopped: return self._stopped = True if not self._started: return self._commands.put(None) self._thread.join(timeout=10.0) _SERVICE = _BrowserService() atexit.register(_SERVICE.shutdown) class PlaywrightBrowserConnector: """Workspace/session-bound callable connector for executor integrations.""" def __init__( self, workspace: str | Path, *, session_id: str = "", timeout_s: float = 0.0, ) -> None: self.workspace = _workspace_root(workspace) self.session_key = _session_digest(session_id) if timeout_s < 0: raise ValueError("browser timeout must be non-negative") self.timeout_s = timeout_s def execute( self, name: str, arguments: Mapping[str, Any] | None = None, ) -> BrowserActionResult: if name not in BROWSER_TOOL_NAMES or name == "BrowserStatus": raise ValueError(f"unsupported browser action: {name}") payload = dict(arguments or {}) _BROWSER_TOOL_SPEC_BY_NAME[name].validate_arguments(payload) return _SERVICE.submit( _BrowserCommand( name=name, arguments=payload, workspace=self.workspace, session_key=self.session_key, timeout_s=self.timeout_s, ) ) def close(self) -> BrowserActionResult: return self.execute("BrowserClose") def __call__( self, name: str, arguments: Mapping[str, Any] | None = None, ) -> BrowserActionResult: return self.execute(name, arguments) def execute_browser_tool( call: ToolCall, context: ToolExecutionContext, ) -> ToolExecutionResult: """Execute an exact browser tool call through the executor contract.""" started = time.perf_counter() spec = _BROWSER_TOOL_SPEC_BY_NAME.get(call.name) if spec is None: error = f"unsupported browser tool: {call.name}" return ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, error=error, elapsed_s=round(time.perf_counter() - started, 4), source_trust="trusted_execution", output_sha256=hashlib.sha256(error.encode("utf-8")).hexdigest(), ) try: spec.validate_arguments(call.args) if call.name == "BrowserStatus": status = browser_status() outcome = BrowserActionResult( bool(status["available"]), status, error=str(status["reason"]) if not status["available"] else "", source_trust="trusted_execution", ) else: connector = PlaywrightBrowserConnector( context.workspace, session_id=context.session_id, timeout_s=context.timeout_s, ) outcome = connector.execute(call.name, call.args) output = json.dumps(outcome.payload, sort_keys=True, ensure_ascii=True) rendered = output or outcome.error return ToolExecutionResult( name=call.name, args=call.args, ok=outcome.ok, tool_call_id=call.call_id, output=output, error=outcome.error, elapsed_s=round(time.perf_counter() - started, 4), executed=True, source_trust=outcome.source_trust, output_sha256=hashlib.sha256(rendered.encode("utf-8")).hexdigest(), ) except Exception as exc: # noqa: BLE001 - optional executor boundary error = f"{type(exc).__name__}: {exc}" return ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, error=error, elapsed_s=round(time.perf_counter() - started, 4), executed=True, source_trust=spec.source_trust, output_sha256=hashlib.sha256(error.encode("utf-8")).hexdigest(), ) def close_all_browser_sessions() -> None: """Close all in-process browser sessions during controlled shutdown.""" _SERVICE.shutdown() __all__ = [ "BROWSER_TOOL_NAMES", "BROWSER_TOOL_SPECS", "BrowserActionResult", "BrowserAvailability", "BrowserTargetError", "BrowserUnavailableError", "PlaywrightBrowserConnector", "browser_availability", "browser_status", "close_all_browser_sessions", "execute_browser_tool", "validate_browser_target", ]