Nexum-Expanded/runtime/src/nexum_runtime/tooling/browser.py

"""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",
]