"""Restricted JSONL worker for the fast-browser prototype.

The worker deliberately delegates browser ownership to browser-harness.  It does
not contain a second CDP client, navigation API, JavaScript execution API, or
model-facing browser handle.  Only the small observe/click/fill surface below
is exposed to the TypeScript adapter.  The session runtime below adds only
fixed, typed read/assertion, tab, navigation, and bounded evidence operations.
"""
from __future__ import annotations

import argparse
import ipaddress
import signal
import subprocess
import json
import os
import secrets
import sys
import tempfile
import time
import uuid
import base64
import re
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Iterable, Protocol
from urllib.parse import urlsplit, urlunsplit

# Protocol limits are intentionally small.  In particular, page text never
# becomes an unbounded JSONL response.
# Runtime evidence carries bounded base64 bytes; keep JSONL framing bounded
# above the published 8 MiB screenshot ceiling while retaining one line/request.
MAX_LINE_BYTES = 12 * 1024 * 1024
MAX_OUTPUT_BYTES = 12 * 1024 * 1024
MAX_CANDIDATES = 60
MAX_METADATA_BATCH = 16
MAX_OBSERVATIONS = 8
MAX_STRING = 4_000
MAX_NAME = 240
MAX_FILL_TEXT = 8_000
MAX_READ_TEXT = 16_000
# JSONL transport remains bounded below the existing bridge line ceiling. The
# parent runtime writes these bytes to its own PrivateArtifactStore.
MAX_SCREENSHOT_BYTES = 8 * 1024 * 1024
MAX_DIAGNOSTICS_BYTES = 256 * 1024
MAX_EVENTS = 256
MAX_CONSOLE_FAILURES = 64
MAX_NETWORK_FAILURES = 64
MAX_EVENT_MESSAGE = 1_000
ACTION_TIMEOUT_SECONDS = 3.0
READ_ASSERTION_TIMEOUT_SECONDS = 3.0
NAVIGATION_TIMEOUT_SECONDS = 10.0
# Leave a small IPC margin below the published 5s popup budget so a normal
# timeout returns a structured result instead of racing the parent deadline.
POPUP_WAIT_SECONDS = 4.75
EXPECTED_BROWSER_HARNESS_VERSION = "0.1.13"
_INTERNAL_TARGET_PREFIXES = (
    "chrome://", "chrome-untrusted://", "devtools://", "chrome-extension://", "about:",
)
_TEXT_INPUT_TYPES = {"", "text", "search", "email", "url", "tel", "number", "password"}

_FILL_ROLES = {"textbox", "searchbox", "spinbutton"}
_CLICK_ROLES = {
    "button", "link", "checkbox", "radio", "switch", "tab", "menuitem",
    "option", "combobox", "listbox", "treeitem",
}


class WorkerError(RuntimeError):
    """A safe, non-sensitive error suitable for the JSONL response."""


class BrowserOperationError(WorkerError):
    """Structured operation failure with no raw browser exception attached."""

    def __init__(self, details: dict[str, Any]):
        self.details = details
        super().__init__(str(details.get("message", "browser operation failed")))


def _operation_error(
    operation: str,
    code: str,
    phase: str,
    message: str,
    *,
    mutation_state: str = "not_attempted",
    retry: str = "never",
) -> BrowserOperationError:
    # Keep this shape aligned with BrowserErrorDetails.  Messages are selected
    # from fixed, non-sensitive strings by callers; browser exception text is
    # deliberately never inserted here.
    return BrowserOperationError({
        "code": code,
        "operation": operation,
        "phase": phase,
        "message": message[:MAX_STRING],
        "mutationState": mutation_state,
        "retry": retry,
    })


@dataclass(frozen=True)
class ElementMetadata:
    backend_id: int
    tag: str
    input_type: str
    role: str
    name: str
    visible: bool
    disabled: bool
    connected: bool = True
    main_frame: bool = True
    contenteditable: bool = False
    ax_role: str = ""
    ax_name: str = ""


@dataclass(frozen=True)
class CandidateRecord:
    id: str
    backend_id: int
    role: str
    name: str
    actions: tuple[str, ...]
    metadata: ElementMetadata


@dataclass(frozen=True)
class ObservationRecord:
    observation_id: str
    target_id: str
    frame_id: str
    loader_id: str
    url: str
    title: str
    candidates: tuple[CandidateRecord, ...]


def _safe_event_message(value: Any) -> str:
    if not isinstance(value, str):
        return "browser event"
    value = " ".join(value.split())
    # Console data is untrusted page content. It remains local evidence, but
    # public shaping must not carry URL query/fragment tokens or obvious
    # credential-bearing key/value diagnostics.
    value = re.sub(r"https?://[^\s?#]+(?:[?#][^\s]*)?", lambda match: (_origin(match.group(0)) or "[url]"), value)
    value = re.sub(r"(?i)(password|passwd|secret|token|authorization|cookie)(\s*[:=]\s*)[^\s,;]+", r"\1\2[redacted]", value)
    value = "".join(ch for ch in value if ord(ch) >= 0x20 or ch in "\t\n")
    return value[:MAX_EVENT_MESSAGE] or "browser event"


def _policy_event_message(value: str, policy: OriginPolicy) -> str:
    return re.sub(
        r"https?://[^\s?#]+(?:[?#][^\s]*)?",
        lambda match: (_origin(match.group(0)) if _approved_origin(_origin(match.group(0)), policy) else "[url]"),
        value,
    )


def _safe_component(value: Any, label: str) -> str:
    if not isinstance(value, str) or not value or len(value) > MAX_STRING or "\x00" in value:
        raise WorkerError(f"invalid {label}")
    # IDs may be opaque but must not become path traversal or separator input.
    if value in {".", ".."} or "/" in value or "\\" in value:
        raise WorkerError(f"invalid {label}")
    return value


class BrowserBackend(Protocol):
    """The deliberately tiny seam used by offline tests and the real adapter."""

    def current_tab(self) -> dict[str, Any]: ...
    def document_identity(self) -> tuple[str, str]: ...
    def accessibility_nodes(self) -> Iterable[dict[str, Any]]: ...
    def metadata(self, backend_id: int) -> ElementMetadata: ...
    def click(self, backend_id: int) -> None: ...
    def fill(self, backend_id: int, text: str) -> None: ...
    def value_matches(self, backend_id: int, text: str) -> bool: ...
    def visible_text(self, text: str) -> bool: ...
    def close(self) -> None: ...


def _origin(url: str) -> str | None:
    """Return a normalized web origin, rejecting non-web and malformed URLs."""
    try:
        parsed = urlsplit(url)
        if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname or "*" in parsed.hostname:
            return None
        if parsed.username is not None or parsed.password is not None:
            return None
        port = parsed.port
        host = parsed.hostname.lower()
        if "*" in host:
            return None
        try:
            address = ipaddress.ip_address(host)
            host = f"[{host}]" if address.version == 6 else host
        except ValueError:
            pass
        if (parsed.scheme.lower() == "http" and port == 80) or (
            parsed.scheme.lower() == "https" and port == 443
        ):
            port = None
        netloc = host if port is None else f"{host}:{port}"
        return urlunsplit((parsed.scheme.lower(), netloc, "", "", ""))
    except ValueError:
        return None


OriginPolicy = str | frozenset[str]


def normalize_allowed_origins(origins: str | Iterable[str]) -> OriginPolicy:
    if origins == "*":
        return "*"
    if isinstance(origins, str):
        raise ValueError("allowed origins must be '*' or a JSON list")
    normalized: set[str] = set()
    for raw in origins:
        if not isinstance(raw, str) or not raw or len(raw) > MAX_STRING:
            raise ValueError("allowed origins must be short origin strings")
        value = _origin(raw)
        if value is None or raw not in {value, value + "/"}:
            # Requiring origin-only input avoids silently broadening an
            # allowlist containing a path, query, or credential.
            raise ValueError(f"allowed origin is not an exact web origin: {raw!r}")
        normalized.add(value)
    if not normalized:
        raise ValueError("at least one allowed origin is required")
    return frozenset(normalized)


def _approved_origin(origin: str | None, policy: OriginPolicy) -> bool:
    return origin is not None and (policy == "*" or origin in policy)


def validate_cdp_url(value: str) -> str:
    if not isinstance(value, str) or len(value) > MAX_STRING:
        raise ValueError("cdpUrl must be a short URL")
    try:
        parsed = urlsplit(value)
        host = (parsed.hostname or "").lower()
        try:
            address = ipaddress.ip_address(host) if host else None
        except ValueError:
            address = None
        loopback = host == "localhost" or (address is not None and address.is_loopback)
        if parsed.scheme.lower() != "http" or not loopback or parsed.username is not None or parsed.password is not None:
            raise ValueError
        if parsed.port is None or not (1 <= parsed.port <= 65535):
            raise ValueError
        if parsed.fragment or parsed.query or parsed.path not in {"", "/"}:
            raise ValueError
    except (ValueError, TypeError):
        raise ValueError("cdpUrl must be an http loopback endpoint") from None
    return value


def _short(value: Any, limit: int = MAX_STRING) -> str:
    if not isinstance(value, str):
        raise WorkerError("expected string")
    if len(value) > limit:
        raise WorkerError("string exceeds protocol limit")
    return value


def _ax_value(value: Any) -> str:
    if isinstance(value, dict):
        value = value.get("value", "")
    return value if isinstance(value, str) else str(value or "")


def _safe_name(value: str, limit: int = MAX_NAME) -> str:
    # The worker never reads or returns an input's current value.  Strip
    # controls so an AX label cannot forge a JSONL/log line in a downstream UI.
    value = " ".join(value.split())
    return value[:limit] or "(unnamed)"


def _candidate_actions(meta: ElementMetadata, ax_role: str) -> tuple[str, ...]:
    role = (ax_role or meta.role or "").lower()
    tag = meta.tag.lower()
    input_type = meta.input_type.lower()
    if input_type in {"hidden", "file"} or not meta.visible or meta.disabled or not meta.main_frame or meta.contenteditable:
        return ()
    actions: list[str] = []
    if (tag == "textarea" and role not in {"combobox"}) or (tag == "input" and input_type in _TEXT_INPUT_TYPES):
        actions.append("fill")
    if role in _CLICK_ROLES or tag in {"button", "a", "summary"} or input_type in {"button", "submit", "reset", "checkbox", "radio"}:
        actions.append("click")
    # A combobox is exposed for filling only unless its AX role explicitly
    # describes a clickable widget.  Keep the list stable and duplicate-free.
    return tuple(dict.fromkeys(actions))


def _is_visible_text(value: str) -> bool:
    return bool(value and len(value) <= MAX_STRING)


class BrowserWorker:
    """Stateful, serialized implementation of the restricted bridge protocol."""

    def __init__(self, backend: BrowserBackend, allowed_origins: OriginPolicy):
        self.backend = backend
        self.allowed_origins = normalize_allowed_origins(allowed_origins)
        self._counter = 0
        self._observations: dict[str, ObservationRecord] = {}
        self._closed = False
        self._blocked = False
        # Explicit mutation boundary consumed by the session runtime. It is
        # set before invoking the backend and never inferred from exception
        # wording, which may be arbitrary browser text.
        self._mutation_dispatched = False
        self._mutation_verified = False

    def _require_open(self) -> None:
        if self._closed:
            raise WorkerError("worker is closed")
        if self._blocked:
            raise WorkerError("browser left the approved origin; bridge stopped")

    def _approved_url(self, url: str) -> str:
        origin = _origin(url)
        if not _approved_origin(origin, self.allowed_origins):
            self._blocked = True
            raise WorkerError("browser URL is outside the approved origins")
        return url

    def _state(self) -> tuple[dict[str, Any], str, str]:
        tab = self.backend.current_tab()
        if not isinstance(tab, dict):
            raise WorkerError("invalid browser tab state")
        target_id = tab.get("targetId") or tab.get("target_id")
        url = tab.get("url")
        title = tab.get("title") or ""
        if not isinstance(target_id, str) or not target_id or not isinstance(url, str):
            raise WorkerError("invalid browser tab identity")
        self._approved_url(url)
        frame_id, loader_id = self.backend.document_identity()
        if not isinstance(frame_id, str) or not isinstance(loader_id, str):
            raise WorkerError("invalid main-frame identity")
        return {"target_id": target_id, "url": url, "title": _short(title, MAX_STRING)}, frame_id, loader_id

    def observe(self) -> dict[str, Any]:
        self._require_open()
        tab, frame_id, loader_id = self._state()
        self._counter += 1
        observation_id = f"obs-{self._counter}-{secrets.token_hex(4)}"
        records: list[CandidateRecord] = []
        seen: set[int] = set()
        batch_metadata = getattr(self.backend, "metadata_many", None)
        pending: list[tuple[int, str, str]] = []

        def consume(items: list[tuple[int, str, str]]) -> None:
            if not items:
                return
            metadata_by_id: dict[int, ElementMetadata] = {}
            if callable(batch_metadata):
                try:
                    result = batch_metadata([backend_id for backend_id, _role, _name in items])
                except Exception:
                    result = None
                if isinstance(result, dict):
                    for backend_id, meta in result.items():
                        if type(backend_id) is int and isinstance(meta, ElementMetadata):
                            metadata_by_id[backend_id] = meta
            for backend_id, ax_role, ax_name in items:
                meta = metadata_by_id.get(backend_id)
                if meta is None or meta.backend_id != backend_id:
                    try:
                        meta = self.backend.metadata(backend_id)
                    except Exception:
                        continue
                if not meta.connected or meta.backend_id != backend_id:
                    continue
                # Full AX discovery is authoritative for observation roles and
                # click names. DOM metadata remains the only source for fill
                # labels, so a field's current value cannot enter an observation.
                observed_meta = replace(meta, ax_role=ax_role, ax_name=ax_name)
                candidate_role = (ax_role or observed_meta.role).lower()
                actions = _candidate_actions(observed_meta, candidate_role)
                if not actions:
                    continue
                name = _safe_name(
                    observed_meta.name if "fill" in actions
                    else (ax_name or observed_meta.ax_name or observed_meta.name)
                )
                candidate_id = f"c-{len(records)}"
                records.append(CandidateRecord(
                    candidate_id, backend_id, candidate_role, name, actions, observed_meta
                ))
                if len(records) >= MAX_CANDIDATES:
                    break

        for node in self.backend.accessibility_nodes():
            if len(records) >= MAX_CANDIDATES:
                break
            if not isinstance(node, dict) or node.get("ignored") is True:
                continue
            backend_id = node.get("backendDOMNodeId")
            if type(backend_id) is not int or backend_id <= 0 or backend_id in seen:
                continue
            seen.add(backend_id)
            ax_role = _ax_value(node.get("role", "")).lower()
            ax_name = _ax_value(node.get("name", ""))
            # Do not spend several CDP round trips on static text/layout nodes.
            if ax_role not in _FILL_ROLES | _CLICK_ROLES:
                continue
            pending.append((backend_id, ax_role, ax_name))
            if len(pending) >= MAX_METADATA_BATCH:
                consume(pending)
                pending.clear()
        consume(pending)
        snapshot = ObservationRecord(
            observation_id, tab["target_id"], frame_id, loader_id, tab["url"], tab["title"], tuple(records)
        )
        self._observations[observation_id] = snapshot
        while len(self._observations) > MAX_OBSERVATIONS:
            self._observations.pop(next(iter(self._observations)))
        return {
            "observationId": observation_id,
            "url": tab["url"],
            "title": tab["title"],
            "candidates": [
                {"id": c.id, "role": c.role, "name": c.name, "actions": list(c.actions)}
                for c in records
            ],
        }

    def _find_snapshot(self, request: dict[str, Any]) -> tuple[ObservationRecord, CandidateRecord]:
        observation_id = request.get("observationId")
        candidate_id = request.get("candidateId")
        if not isinstance(observation_id, str) or not isinstance(candidate_id, str):
            raise WorkerError("invalid action target")
        snapshot = self._observations.pop(observation_id, None)
        # Every attempted mutation invalidates the entire prior observation set,
        # not only the particular handle chosen for this action.
        self._observations.clear()
        if snapshot is None:
            raise WorkerError("observation is missing or already consumed")
        candidate = next((item for item in snapshot.candidates if item.id == candidate_id), None)
        if candidate is None:
            raise WorkerError("candidate is not in the observation")
        return snapshot, candidate

    def _revalidate(self, snapshot: ObservationRecord, candidate: CandidateRecord, action: str) -> None:
        tab, frame_id, loader_id = self._state()
        if (
            tab["target_id"] != snapshot.target_id
            or tab["url"] != snapshot.url
            or frame_id != snapshot.frame_id
            or loader_id != snapshot.loader_id
        ):
            raise WorkerError("observation is stale")
        if action not in candidate.actions:
            raise WorkerError("action is not allowed for candidate")
        current = self.backend.metadata(candidate.backend_id)
        current_role = (current.ax_role or current.role).lower()
        current_name = current.name if action == "fill" else (current.ax_name or current.name)
        current_actions = _candidate_actions(current, current_role)
        if (
            not current.connected
            or not current.visible
            or current.disabled
            or current.tag.lower() != candidate.metadata.tag.lower()
            or current.input_type.lower() != candidate.metadata.input_type.lower()
            or current.main_frame != candidate.metadata.main_frame
            or current.contenteditable != candidate.metadata.contenteditable
            or current_role != candidate.role.lower()
            or action not in current_actions
            or _safe_name(current_name) != candidate.name
        ):
            raise WorkerError("candidate changed or is unavailable")

    def _post_action_state(self, snapshot: ObservationRecord, same_document: bool = False) -> dict[str, Any]:
        tab, frame_id, loader_id = self._state()
        if tab["target_id"] != snapshot.target_id:
            raise WorkerError("browser tab changed during action")
        if same_document and (frame_id != snapshot.frame_id or loader_id != snapshot.loader_id):
            raise WorkerError("document changed during fill")
        return tab

    def act(self, request: dict[str, Any]) -> dict[str, Any]:
        self._require_open()
        self._mutation_dispatched = False
        self._mutation_verified = False
        if not isinstance(request, dict):
            raise WorkerError("invalid action request")
        action = request.get("action")
        if action not in {"click", "fill"}:
            raise WorkerError("unsupported action")
        snapshot, candidate = self._find_snapshot(request)
        # Popping the snapshot happens before any revalidation or mutation:
        # attempted mutations can never be replayed with the same observation.
        if action == "click":
            if set(request) - {"observationId", "candidateId", "action", "postcondition"}:
                raise WorkerError("malformed click request")
            postcondition = request.get("postcondition")
            if not isinstance(postcondition, dict) or set(postcondition) != {"kind", "value"}:
                raise WorkerError("click requires one postcondition")
            kind = postcondition.get("kind")
            value = postcondition.get("value")
            if kind not in {"url", "text"} or not isinstance(value, str) or not value or len(value) > MAX_STRING:
                raise WorkerError("invalid click postcondition")
            if kind == "url" and not _approved_origin(_origin(value), self.allowed_origins):
                raise WorkerError("postcondition URL is outside the approved origins")
        else:
            if set(request) - {"observationId", "candidateId", "action", "text", "postcondition"}:
                raise WorkerError("malformed fill request")
            text = request.get("text")
            if not isinstance(text, str) or len(text) > MAX_FILL_TEXT:
                raise WorkerError("fill text exceeds protocol limit")
            postcondition = request.get("postcondition")
            if postcondition is not None:
                if not isinstance(postcondition, dict) or set(postcondition) != {"kind", "value"}:
                    raise WorkerError("invalid fill postcondition")
                if postcondition.get("kind") not in {"url", "text"} or not isinstance(postcondition.get("value"), str):
                    raise WorkerError("invalid fill postcondition")
                if not postcondition["value"] or len(postcondition["value"]) > MAX_STRING:
                    raise WorkerError("invalid fill postcondition")
                if postcondition["kind"] == "url" and not _approved_origin(_origin(postcondition["value"]), self.allowed_origins):
                    raise WorkerError("fill postcondition URL is outside the approved origins")
        self._revalidate(snapshot, candidate, action)
        if action == "fill":
            self._mutation_dispatched = True
            self.backend.fill(candidate.backend_id, text)
            deadline = time.monotonic() + ACTION_TIMEOUT_SECONDS
            while time.monotonic() < deadline:
                tab = self._post_action_state(snapshot, same_document=True)
                value_verified = self.backend.value_matches(candidate.backend_id, text)
                condition_verified = True
                if postcondition is not None:
                    if postcondition["kind"] == "url":
                        condition_verified = tab.get("url") == postcondition["value"]
                    else:
                        condition_verified = self.backend.visible_text(postcondition["value"])
                if value_verified and condition_verified:
                    self._mutation_verified = True
                    return {"verified": True, "evidence": "field value and postcondition verified" if postcondition else "field value verified"}
                time.sleep(0.025)
            return {"verified": False, "evidence": "field value or postcondition could not be verified"}

        # A postcondition that already holds cannot prove that this click did
        # anything.  Refuse it before the one and only mutation.
        tab = self._post_action_state(snapshot)
        if postcondition["kind"] == "url":
            already_satisfied = tab["url"] == postcondition["value"]
        else:
            already_satisfied = _is_visible_text(postcondition["value"]) and self.backend.visible_text(postcondition["value"])
        if already_satisfied:
            raise WorkerError("click postcondition was already satisfied")

        # Exactly one click is ever issued.  Verification polling below never
        # retries the mutation, and checks origin before examining each result.
        self._mutation_dispatched = True
        self.backend.click(candidate.backend_id)
        deadline = time.monotonic() + ACTION_TIMEOUT_SECONDS
        while time.monotonic() < deadline:
            tab = self._post_action_state(snapshot)
            if postcondition["kind"] == "url":
                if tab["url"] == postcondition["value"]:
                    self._mutation_verified = True
                    return {"verified": True, "evidence": "URL postcondition verified"}
            elif _is_visible_text(postcondition["value"]) and self.backend.visible_text(postcondition["value"]):
                self._mutation_verified = True
                return {"verified": True, "evidence": "visible-text postcondition verified"}
            time.sleep(0.025)
        # _post_action_state performs the final approved-origin check too.
        self._post_action_state(snapshot)
        return {"verified": False, "evidence": "click postcondition was not verified"}

    def close(self) -> None:
        if self._closed:
            return
        self._closed = True
        self._observations.clear()
        self.backend.close()


_METADATA_FUNCTION = """function () {
    const e = this;
    const style = getComputedStyle(e);
    const type = (e.getAttribute('type') || '').toLowerCase();
    const text = node => node ? String(node.innerText || node.textContent || '').trim() : '';
    const labelledBy = (e.getAttribute('aria-labelledby') || '').split(/\\s+/)
        .map(id => id ? text(document.getElementById(id)) : '').filter(Boolean).join(' ');
    const labels = e.labels ? [...e.labels].map(text).filter(Boolean).join(' ') : '';
    const visibleControlText = ['BUTTON', 'A', 'SUMMARY'].includes(e.tagName) ? text(e) : '';
    // Never read .value: for fill candidates this is user data.
    const label = e.getAttribute('aria-label') || labelledBy || labels ||
        (visibleControlText || e.getAttribute('placeholder') ||
            e.getAttribute('name') || e.getAttribute('id') || '');
    return {
        tag: e.tagName.toLowerCase(), type, role: e.getAttribute('role') || '',
        name: String(label).slice(0, 240), connected: !!e.isConnected,
        mainFrame: window.top === window,
        contenteditable: e.isContentEditable,
        visible: !!(e.getClientRects().length && style.display !== 'none' &&
            style.visibility !== 'hidden' && style.opacity !== '0' &&
            e.getAttribute('aria-hidden') !== 'true'),
        disabled: !!(e.disabled || e.getAttribute('aria-disabled') === 'true' ||
            e.matches(':disabled'))
    };
}"""
_METADATA_MANY_FUNCTION = (
    "function (...nodes) {\n"
    f"    const read = {_METADATA_FUNCTION};\n"
    "    return nodes.map(node => node && typeof node === 'object' ? read.call(node) : null);\n"
    "}"
)


class RealBrowserBackend:
    """Adapter over browser-harness; no CDP implementation is duplicated here."""

    def __init__(self, cdp_url: str, allowed_origins: OriginPolicy):
        validate_cdp_url(cdp_url)
        self.allowed_origins = normalize_allowed_origins(allowed_origins)
        self._sandbox = Path(tempfile.mkdtemp(prefix="fast-browser-worker-"))
        self._screenshot_dir = self._sandbox / "screenshots"
        self._screenshot_dir.mkdir(mode=0o700)
        try:
            os.chmod(self._screenshot_dir, 0o700)
        except OSError:
            pass
        self._daemon_ready = False
        self._process: subprocess.Popen | None = None
        self._attached_target: str | None = None
        self._select_all_modifier: int | None = None
        self._configure_environment(cdp_url, self._sandbox)
        try:
            self._load_harness()
            self._start_daemon()
            self._attach_initial_tab()
        except BaseException:
            self.close()
            raise

    def _start_daemon(self) -> None:
        # Own the actual child handle. admin.ensure_daemon detaches/retries its
        # child and can launch Chrome; neither behavior belongs in this adapter.
        self._process = subprocess.Popen(
            [sys.executable, "-m", "browser_harness.daemon"],
            env=dict(os.environ), stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if self._process.poll() is not None:
                raise WorkerError("browser-harness daemon failed to connect to the configured endpoint")
            try:
                self._helpers.cdp("Target.getTargets", _response_timeout=0.2)
                self._daemon_ready = True
                return
            except Exception:
                time.sleep(0.025)
        raise WorkerError("configured browser endpoint was not ready within five seconds")

    @staticmethod
    def _configure_environment(cdp_url: str, root: Path) -> None:
        # The directory is retained by the instance through the environment;
        # setting it before importing browser-harness prevents its .env and
        # agent-helper discovery from reaching the user's workspace.
        os.environ.update({
            "BU_CDP_URL": cdp_url,
            "BU_NAME": f"fast-{uuid.uuid4().hex[:20]}",
            "BH_HOME": str(root / "home"),
            "BH_CONFIG_DIR": str(root / "config"),
            "BH_RUNTIME_DIR": str(root / "runtime"),
            "BH_RUNTIME_DIR_SHARED": "0",
            "BH_TMP_DIR": str(root / "tmp"),
            "BH_TMP_DIR_SHARED": "0",
            "BH_AGENT_WORKSPACE": str(root / "workspace"),
            "BH_TELEMETRY": "0",
            "BH_UPDATE_CHECK": "0",
            "BH_RECORD": "0",
            "BH_TAB_MARKER": "0",
            "PYTHONNOUSERSITE": "1",
            # Empty, rather than absent: browser-harness's .env loader uses
            # setdefault and must not reintroduce cloud credentials or spawn.
            "BROWSER_USE_API_KEY": "",
            "BU_CDP_WS": "",
            "BU_AUTOSPAWN": "",
            "BU_BROWSER_ID": "",
            "BU_PROFILE_ID": "",
            "BU_PROXY_URL": "",
            "BU_PROXY_USERNAME": "",
            "BU_PROXY_PASSWORD": "",
        })

    def _load_harness(self) -> None:
        # browser-harness must come from the configured Python environment's
        # installed package.  In particular, never import a temporary checkout
        # via sys.path: that would make the runtime transport supply-chain
        # mutable and would bypass the pinned environment chosen by the caller.
        from importlib.metadata import PackageNotFoundError, version
        try:
            installed = version("browser-harness")
        except PackageNotFoundError:
            raise WorkerError(
                "browser-harness is not installed in this Python environment; "
                "install the pinned 0.1.13 package or set FAST_BROWSER_PYTHON"
            ) from None
        if installed != EXPECTED_BROWSER_HARNESS_VERSION:
            raise WorkerError(
                f"browser-harness {EXPECTED_BROWSER_HARNESS_VERSION} is required, found {installed}; "
                "use a dedicated pinned Python environment (no auto-upgrade is attempted)"
            )
        from browser_harness import helpers  # type: ignore
        self._helpers = helpers

    def _attach_initial_tab(self) -> None:
        """Attach one eligible target when unambiguous, otherwise stay idle.

        Runtime startup must remain usable with multiple tabs so callers can
        list and explicitly select a target.  A zero-tab startup still fails
        closed because no operation can be served until a target exists.
        """
        tabs = self._eligible_tabs()
        if not tabs:
            raise WorkerError("no approved-origin browser tab is available")
        if len(tabs) > 1:
            self._attached_target = None
            return
        target = tabs[0].get("targetId") or tabs[0].get("target_id")
        if not isinstance(target, str) or not target:
            raise WorkerError("approved browser tab has no target identity")
        self.switch_tab(target)

    def _eligible_tabs(self) -> list[dict[str, Any]]:
        tabs = self._helpers.list_tabs(include_chrome=False)
        return [
            tab for tab in tabs
            if isinstance(tab, dict)
            and isinstance(tab.get("url"), str)
            and _approved_origin(_origin(tab["url"]), self.allowed_origins)
            and isinstance(tab.get("targetId") or tab.get("target_id"), str)
        ]

    def _attach_approved_tab(self) -> None:
        # A named browser-harness daemon owns a blank dedicated tab.  It is
        # intentionally not navigated: attach only to the one already-open
        # approved tab and leave the user's selected tab/visibility untouched.
        tabs = self._helpers.list_tabs(include_chrome=False)
        approved = [
            tab for tab in tabs
            if isinstance(tab, dict)
            and isinstance(tab.get("url"), str)
            and _approved_origin(_origin(tab["url"]), self.allowed_origins)
        ]
        if len(approved) != 1:
            # Diagnose redirects without exposing titles, paths, query tokens,
            # field values, or any content from outside the allowed origins.
            web_origins = sorted({
                origin for tab in tabs if isinstance(tab, dict)
                and isinstance(tab.get("url"), str)
                if (origin := _origin(tab["url"])) is not None
            })
            summary = json.dumps([origin[:240] for origin in web_origins[:10]])
            raise WorkerError(
                "expected exactly one existing approved-origin browser tab; "
                f"found {len(approved)}; open web origins: {summary}"
            )
        target = approved[0].get("targetId") or approved[0].get("target_id")
        if not isinstance(target, str) or not target:
            raise WorkerError("approved browser tab has no target identity")
        self._helpers.switch_tab(target, activate=False)

    def switch_tab(self, target_id: str) -> None:
        if not isinstance(target_id, str) or not target_id:
            raise WorkerError("tab identity is unavailable")
        # browser-harness 0.1.13 enables Page/DOM/Runtime/Network as part of
        # its switch_tab set_session path. Re-enabling them here duplicates
        # four CDP calls and can race the daemon's session setup.
        self._session_id = self._helpers.switch_tab(target_id, activate=False)
        self._attached_target = target_id

    def list_tabs(self) -> list[dict[str, Any]]:
        # helpers.list_tabs() and this opener lookup used to issue two identical
        # Target.getTargets requests. Keep its page/internal filtering locally
        # while deriving opener metadata from the same target inventory.
        raw = self._helpers.cdp("Target.getTargets")
        target_infos = raw.get("targetInfos", []) if isinstance(raw, dict) else []
        if not isinstance(target_infos, list):
            return []
        result: list[dict[str, Any]] = []
        for info in target_infos:
            if not isinstance(info, dict) or info.get("type") != "page":
                continue
            url = info.get("url", "")
            title = info.get("title", "")
            if not isinstance(url, str) or not isinstance(title, str):
                continue
            # Match browser-harness helpers.list_tabs(include_chrome=False),
            # including its startup-placeholder suppression.
            if title.startswith("Starting agent ") and (
                url in ("", "about:blank") or url.startswith("about:blank#")
            ):
                continue
            if url.startswith(_INTERNAL_TARGET_PREFIXES):
                continue
            target = info.get("targetId")
            if not isinstance(target, str) or not target:
                continue
            opener = info.get("openerId")
            result.append({
                "targetId": target,
                "target_id": target,
                "url": url,
                "title": title,
                "openerTargetId": opener if isinstance(opener, str) and opener else None,
            })
        return result

    def navigate(self, url: str) -> dict[str, Any]:
        result = self._helpers.goto_url(url)
        if isinstance(result, dict):
            return result
        return {}

    def reload(self) -> dict[str, Any]:
        return self._helpers.cdp("Page.reload")

    def wait_ready(self, timeout: float = NAVIGATION_TIMEOUT_SECONDS) -> bool:
        # Keep the harness' exact readyState query, but avoid its coarse 300ms
        # sleep. This is only a readiness read; it never dispatches navigation.
        if timeout <= 0:
            return False
        deadline = time.monotonic() + timeout
        while True:
            if self._helpers.js("document.readyState") == "complete":
                return True
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return False
            time.sleep(min(0.025, remaining))

    def visible_text_content(self) -> str:
        # This is a fixed, worker-owned read operation.  No caller expression,
        # selector, HTML, form value, frame, or script output is accepted.
        expression = """(() => {
            const root = document.body;
            if (!root) return '';
            const blocked = new Set(['SCRIPT','STYLE','NOSCRIPT','TEMPLATE',
                'INPUT','TEXTAREA','SELECT','OPTION','DATALIST','OUTPUT']);
            const visible = node => {
                for (let e = node.nodeType === Node.ELEMENT_NODE ? node : node.parentElement;
                     e && e !== root.parentElement; e = e.parentElement) {
                    const s = getComputedStyle(e);
                    if (e.getAttribute('aria-hidden') === 'true' ||
                        s.display === 'none' || s.visibility === 'hidden' ||
                        s.opacity === '0') return false;
                }
                return true;
            };
            const walker = document.createTreeWalker(root, NodeFilter.SHOW_TEXT);
            const parts = [];
            let size = 0;
            let node;
            while ((node = walker.nextNode())) {
                const parent = node.parentElement;
                if (!parent || blocked.has(parent.tagName) || !visible(parent)) continue;
                const text = String(node.nodeValue || '').replace(/[\\t\\n\\r ]+/g, ' ').trim();
                if (text) {
                    parts.push(text);
                    size += text.length + 1;
                    if (size > 16001) break;
                }
            }
            return parts.join(' ').replace(/ +/g, ' ').trim().slice(0, 16001);
        })()"""
        result = self._helpers.cdp("Runtime.evaluate", expression=expression,
                                   returnByValue=True, awaitPromise=False)
        remote = result.get("result") or {}
        value = remote.get("value")
        if not isinstance(value, str):
            raise WorkerError("visible page text is unavailable")
        return value

    def drain_events(self) -> list[dict[str, Any]]:
        try:
            events = self._helpers.drain_events()
        except Exception:
            return []
        return events if isinstance(events, list) else []

    def screenshot_bytes(self) -> bytes:
        # NamedTemporaryFile creates the file with O_EXCL inside the daemon's
        # private 0700 sandbox; the helper only receives this worker-owned path.
        handle = tempfile.NamedTemporaryFile(
            prefix="shot-", suffix=".png", dir=self._screenshot_dir,
            mode="w+b", delete=False,
        )
        path = Path(handle.name)
        handle.close()
        try:
            os.chmod(path, 0o600)
            self._helpers.capture_screenshot(str(path), full=False, max_dim=2000)
            with open(path, "rb") as stream:
                return stream.read(MAX_SCREENSHOT_BYTES + 1)
        finally:
            try:
                path.unlink()
            except FileNotFoundError:
                pass

    def close_tab(self, target_id: str) -> None:
        self._helpers.close_tab(target_id)

    def current_tab(self) -> dict[str, Any]:
        return self._helpers.current_tab()

    def document_identity(self) -> tuple[str, str]:
        tree = self._helpers.cdp("Page.getFrameTree")
        frame = (tree.get("frameTree") or {}).get("frame") or {}
        frame_id = frame.get("id")
        loader_id = frame.get("loaderId") or "no-loader"
        if not isinstance(frame_id, str):
            raise WorkerError("main frame is unavailable")
        return frame_id, str(loader_id)

    def accessibility_nodes(self) -> Iterable[dict[str, Any]]:
        tree = self._helpers.cdp("Accessibility.getFullAXTree", depth=10)
        nodes = tree.get("nodes")
        return nodes if isinstance(nodes, list) else ()

    def _call(self, backend_id: int, function: str, arguments: list[dict[str, Any]] | None = None) -> Any:
        resolved = self._helpers.cdp("DOM.resolveNode", backendNodeId=backend_id)
        object_id = (resolved.get("object") or {}).get("objectId")
        if not isinstance(object_id, str):
            raise WorkerError("element is unavailable")
        try:
            result = self._helpers.cdp(
                "Runtime.callFunctionOn", objectId=object_id, functionDeclaration=function,
                arguments=arguments or [], returnByValue=True, awaitPromise=False,
            )
            details = result.get("exceptionDetails")
            remote = result.get("result") or {}
            if details or remote.get("subtype") == "error" or "value" not in remote:
                raise WorkerError("browser evaluation failed")
            value = remote.get("value")
            if not isinstance(value, (dict, bool, str, int, float, type(None))):
                raise WorkerError("invalid browser metadata")
            return value
        finally:
            try:
                self._helpers.cdp("Runtime.releaseObject", objectId=object_id)
            except Exception:
                pass

    @staticmethod
    def _metadata_from_dom_value(backend_id: int, value: Any) -> ElementMetadata:
        if not isinstance(value, dict):
            raise WorkerError("element metadata unavailable")
        return ElementMetadata(
            backend_id, str(value.get("tag", "")), str(value.get("type", "")),
            str(value.get("role", "")), " ".join(str(value.get("name", "")).split())[:MAX_NAME],
            bool(value.get("visible")), bool(value.get("disabled")), bool(value.get("connected")),
            bool(value.get("mainFrame", True)), bool(value.get("contenteditable", False)),
        )

    def metadata(self, backend_id: int) -> ElementMetadata:
        value = self._call(backend_id, _METADATA_FUNCTION)
        metadata = self._metadata_from_dom_value(backend_id, value)
        ax_role, ax_name = "", ""
        try:
            partial = self._helpers.cdp(
                "Accessibility.getPartialAXTree", backendNodeId=backend_id, fetchRelatives=False
            )
            for node in partial.get("nodes", []) if isinstance(partial, dict) else []:
                if not isinstance(node, dict):
                    continue
                ax_role = _ax_value(node.get("role", "")).lower()
                ax_name = _ax_value(node.get("name", ""))
                if ax_role or ax_name:
                    break
        except Exception:
            # DOM metadata still provides safe labels; accessibility is a
            # refinement and not a reason to expose a broad CDP surface.
            pass
        return replace(metadata, ax_role=ax_role, ax_name=ax_name)

    def metadata_many(self, backend_ids: Iterable[int]) -> dict[int, ElementMetadata]:
        """Read value-free DOM metadata for one bounded observation chunk.

        The full AX tree is already the observation's source for role/name. This
        seam only batches the DOM extraction; action revalidation continues to
        use metadata(), including its fresh partial AX read.
        """
        ids = [backend_id for backend_id in backend_ids if type(backend_id) is int and backend_id > 0][:MAX_METADATA_BATCH]
        if not ids:
            return {}
        object_group = f"fast-browser-metadata-{uuid.uuid4().hex}"
        object_ids: list[str | None] = []
        try:
            for backend_id in ids:
                try:
                    resolved = self._helpers.cdp(
                        "DOM.resolveNode", backendNodeId=backend_id, objectGroup=object_group
                    )
                except Exception:
                    object_ids.append(None)
                    continue
                object_id = (resolved.get("object") or {}).get("objectId") if isinstance(resolved, dict) else None
                object_ids.append(object_id if isinstance(object_id, str) else None)

            target_object = next((object_id for object_id in object_ids if object_id is not None), None)
            if target_object is None:
                return {}
            arguments = [
                {"objectId": object_id} if object_id is not None else {"value": None}
                for object_id in object_ids
            ]
            try:
                response = self._helpers.cdp(
                    "Runtime.callFunctionOn", objectId=target_object,
                    functionDeclaration=_METADATA_MANY_FUNCTION, arguments=arguments,
                    returnByValue=True, awaitPromise=False, objectGroup=object_group,
                )
            except Exception:
                return {}
            details = response.get("exceptionDetails") if isinstance(response, dict) else None
            remote = response.get("result") if isinstance(response, dict) else None
            values = remote.get("value") if isinstance(remote, dict) else None
            if details or not isinstance(remote, dict) or remote.get("subtype") == "error" or not isinstance(values, list):
                return {}
            result: dict[int, ElementMetadata] = {}
            for backend_id, value in zip(ids, values):
                try:
                    result[backend_id] = self._metadata_from_dom_value(backend_id, value)
                except WorkerError:
                    # Missing or malformed entries use the safe legacy path in
                    # BrowserWorker.observe rather than being treated as absent.
                    continue
            return result
        finally:
            try:
                self._helpers.cdp("Runtime.releaseObjectGroup", objectGroup=object_group)
            except Exception:
                pass

    def click(self, backend_id: int) -> None:
        box = self._helpers.cdp("DOM.getBoxModel", backendNodeId=backend_id)
        content = (box.get("model") or {}).get("content")
        if not isinstance(content, list) or len(content) < 8:
            raise WorkerError("candidate has no clickable box")
        xs = [float(content[i]) for i in range(0, len(content), 2)]
        ys = [float(content[i]) for i in range(1, len(content), 2)]
        x, y = sum(xs) / len(xs), sum(ys) / len(ys)
        hit = self._call(backend_id, """function (x, y) {
            const hit = document.elementFromPoint(x, y);
            return !!hit && (hit === this || this.contains(hit));
        }""", [{"value": x}, {"value": y}])
        if hit is not True:
            raise WorkerError("candidate is covered or no longer at its observed position")
        self._helpers.cdp("Input.dispatchMouseEvent", type="mousePressed", x=x, y=y, button="left", clickCount=1)
        self._helpers.cdp("Input.dispatchMouseEvent", type="mouseReleased", x=x, y=y, button="left", clickCount=1)

    def fill(self, backend_id: int, text: str) -> None:
        self._helpers.cdp("DOM.focus", backendNodeId=backend_id)
        focused = self._call(backend_id, """function () {
            return document.activeElement === this;
        }""")
        if focused is not True:
            raise WorkerError("candidate could not be focused")
        # Select all without an arbitrary page script. Input.insertText is a
        # single bounded browser operation and is substantially faster than
        # dispatching up to 8,000 individual key events. Browser.getVersion is
        # stable for this backend, so cache the platform modifier for its life.
        if getattr(self, "_select_all_modifier", None) is None:
            user_agent = self._helpers.cdp("Browser.getVersion").get("userAgent", "")
            self._select_all_modifier = 4 if "Mac OS X" in user_agent or "Macintosh" in user_agent else 2
        modifiers = self._select_all_modifier
        select_all = {"key": "a", "code": "KeyA", "modifiers": modifiers,
                      "windowsVirtualKeyCode": 65, "nativeVirtualKeyCode": 65,
                      "commands": ["SelectAll"]}
        self._helpers.cdp("Input.dispatchKeyEvent", type="rawKeyDown", **select_all)
        self._helpers.cdp("Input.dispatchKeyEvent", type="keyUp", **{k: v for k, v in select_all.items() if k != "commands"})
        self._helpers.press_key("Backspace")
        if text:
            self._helpers.cdp("Input.insertText", text=text)
        self._call(backend_id, """function () {
            this.dispatchEvent(new Event('input', {bubbles: true}));
            this.dispatchEvent(new Event('change', {bubbles: true}));
            return true;
        }""")

    def value_matches(self, backend_id: int, text: str) -> bool:
        value = self._call(backend_id, """function () {
            return 'value' in this ? String(this.value) : String(this.textContent || '');
        }""")
        return isinstance(value, str) and value == text

    def visible_text(self, text: str) -> bool:
        # The literal is JSON-encoded, never concatenated as executable code.
        result = self._helpers.cdp(
            "Runtime.evaluate", expression="""(() => {
                const wanted = __TEXT__;
                const visible = e => {
                    const s = getComputedStyle(e);
                    return !!(e.getClientRects().length && s.display !== 'none' &&
                        s.visibility !== 'hidden' && s.opacity !== '0');
                };
                return [...document.querySelectorAll('body *')].some(e =>
                    visible(e) && (e.innerText || '').includes(wanted));
            })()""".replace("__TEXT__", json.dumps(text)), returnByValue=True,
        )
        return (result.get("result") or {}).get("value") is True

    def close(self) -> None:
        if getattr(self, "_closed", False):
            return
        self._closed = True
        if getattr(self, "_daemon_ready", False):
            try:
                # This is the browser-harness daemon's ownership-aware
                # shutdown; it closes only this named daemon's dedicated tab.
                self._helpers._send({"meta": "shutdown"}, response_timeout=0.3)
            except Exception:
                pass
            self._daemon_ready = False
        process = getattr(self, "_process", None)
        if process is not None:
            try:
                process.wait(timeout=0.3)
            except subprocess.TimeoutExpired:
                # asyncio's SIGINT handling runs the harness ownership cleanup.
                try:
                    process.send_signal(signal.SIGINT)
                except ProcessLookupError:
                    pass
                try:
                    process.wait(timeout=0.3)
                except subprocess.TimeoutExpired:
                    try:
                        process.kill()
                    except ProcessLookupError:
                        pass
                    try:
                        process.wait(timeout=0.3)
                    except subprocess.TimeoutExpired:
                        pass
        self._cleanup_sandbox()

    def _cleanup_sandbox(self) -> None:
        import shutil
        shutil.rmtree(self._sandbox, ignore_errors=True)


@dataclass
class _RuntimeTab:
    tab_id: str
    target_id: str
    kind: str
    opener_target_id: str | None
    url: str
    title: str
    active: bool = True


class BrowserRuntimeWorker(BrowserWorker):
    """Persistent session runtime for the published M2/M3/M4 operations."""

    def __init__(self, backend: BrowserBackend, allowed_origins: OriginPolicy,
                 *, session_id: str | None = None):
        super().__init__(backend, allowed_origins)
        self.session_id = _safe_component(session_id or f"session-{uuid.uuid4().hex}", "session id")
        self._base_policy = self.allowed_origins
        self._tabs: dict[str, _RuntimeTab] = {}
        self._target_to_tab: dict[str, str] = {}
        self._selected_tab_id: str | None = None
        self._tab_counter = 0
        self._event_buffer: list[dict[str, Any]] = []
        self._popup_candidates: set[str] = set()
        self._claimed_popups: set[str] = set()
        self._initializing = True
        self._closed_runtime = False
        self._refresh_tabs()
        # Existing popup/page targets are baseline state, not newly-owned
        # popup candidates for a later waitPopup call.
        self._popup_candidates.clear()
        self._initializing = False
        # Exactly one initial target is safe to select.  With multiple tabs no
        # arbitrary page is selected; listTabs remains available.
        if len(self._tabs) == 1:
            self._selected_tab_id = next(iter(self._tabs))
            self._attach(self._tabs[self._selected_tab_id].target_id, "selectTab")

    def _fail(self, operation: str, code: str, phase: str, message: str,
              *, mutation_state: str = "not_attempted", retry: str = "never") -> BrowserOperationError:
        return _operation_error(operation, code, phase, message,
                                mutation_state=mutation_state, retry=retry)

    def _backend_tabs(self) -> list[dict[str, Any]]:
        listing = getattr(self.backend, "list_tabs", None)
        if callable(listing):
            try:
                value = listing()
            except Exception:
                raise self._fail("listTabs", "worker_failure", "list", "browser tabs are unavailable")
            return value if isinstance(value, list) else []
        try:
            current = self.backend.current_tab()
        except Exception:
            return []
        return [current] if isinstance(current, dict) else []

    def _refresh_tabs(self) -> None:
        seen: set[str] = set()
        raw_tabs = self._backend_tabs()
        # Build the eligible target index before classifying any row. CDP target
        # ordering is not an ownership signal: a popup may precede its opener.
        entries: list[tuple[dict[str, Any], str, str]] = []
        target_index: set[str] = set()
        for raw in raw_tabs:
            if not isinstance(raw, dict):
                continue
            target = raw.get("targetId") or raw.get("target_id")
            url = raw.get("url")
            if not isinstance(target, str) or not target or not isinstance(url, str):
                continue
            origin = _origin(url)
            # Discovery is bounded by the immutable session base. A per-call
            # context only filters/authorizes an operation; it must not make a
            # tab handle disappear when the next call restores the base.
            if not _approved_origin(origin, self._base_policy):
                continue
            entries.append((raw, target, url))
            target_index.add(target)
        for raw, target, url in entries:
            seen.add(target)
            tab_id = self._target_to_tab.get(target)
            opener = raw.get("openerTargetId") or raw.get("openerId") or raw.get("opener_target_id")
            opener_target = opener if isinstance(opener, str) and opener else None
            owned_opener = opener_target if opener_target in target_index or opener_target in self._target_to_tab else None
            if tab_id is None:
                self._tab_counter += 1
                tab_id = f"tab-{self._tab_counter}-{secrets.token_hex(6)}"
                self._target_to_tab[target] = tab_id
                kind = "popup" if owned_opener else "page"
                self._tabs[tab_id] = _RuntimeTab(tab_id, target, kind, owned_opener,
                                                   url, _short(raw.get("title") or ""))
                if kind == "popup" and not self._initializing:
                    self._popup_candidates.add(target)
            else:
                tab = self._tabs[tab_id]
                tab.url = url
                tab.title = _short(raw.get("title") or "")
                if owned_opener:
                    was_popup = tab.kind == "popup"
                    tab.opener_target_id = owned_opener
                    tab.kind = "popup"
                    if not was_popup and not self._initializing:
                        self._popup_candidates.add(target)
        # Handles to closed tabs are invalidated.  Do not invent a replacement
        # selection when a selected tab disappears.
        for target, tab_id in list(self._target_to_tab.items()):
            if target not in seen:
                self._target_to_tab.pop(target, None)
                self._tabs.pop(tab_id, None)
                self._popup_candidates.discard(target)
                self._claimed_popups.discard(target)
                if self._selected_tab_id == tab_id:
                    self._selected_tab_id = None

    def _attach(self, target_id: str, operation: str = "selectTab") -> None:
        attached = getattr(self.backend, "_attached_target", None)
        if attached is not None and attached == target_id:
            return
        switch = getattr(self.backend, "switch_tab", None)
        if callable(switch):
            try:
                switch(target_id)
            except Exception:
                raise self._fail(operation, "no_tab", "attach", "tab is unavailable")
            return
        try:
            current = self.backend.current_tab()
        except Exception:
            raise self._fail(operation, "no_tab", "attach", "tab is unavailable")
        if current.get("targetId") != target_id and current.get("target_id") != target_id:
            raise self._fail(operation, "no_tab", "attach", "tab is unavailable")

    def _resolve_tab(self, request: dict[str, Any] | None, operation: str) -> _RuntimeTab:
        self._refresh_tabs()
        tab_id = request.get("tabId") if isinstance(request, dict) else None
        if tab_id is not None:
            if not isinstance(tab_id, str) or len(tab_id) > 200:
                raise self._fail(operation, "invalid_request", "target", "tab handle is invalid")
            tab = self._tabs.get(tab_id)
            if tab is None:
                raise self._fail(operation, "no_tab", "target", "tab handle is stale or unavailable")
        else:
            if self._selected_tab_id is None:
                if not self._tabs:
                    raise self._fail(operation, "no_tab", "target", "no approved tab is available")
                raise self._fail(operation, "ambiguous_tab", "target", "select an approved tab explicitly")
            tab = self._tabs.get(self._selected_tab_id)
            if tab is None:
                self._selected_tab_id = None
                raise self._fail(operation, "no_tab", "target", "selected tab is unavailable")
        self._attach(tab.target_id, operation)
        # The inherited worker uses the current backend target and current
        # policy.  Verify it once more without returning raw browser details.
        try:
            current = self.backend.current_tab()
            url = current.get("url") if isinstance(current, dict) else None
        except Exception:
            raise self._fail(operation, "no_tab", "target", "tab is unavailable")
        if not isinstance(url, str) or not _approved_origin(_origin(url), self.allowed_origins):
            raise self._fail(operation, "policy_denied", "policy", "tab is outside the approved origins")
        return tab

    def _document_id(self, operation: str) -> str:
        try:
            frame, loader = self.backend.document_identity()
        except Exception:
            raise self._fail(operation, "worker_failure", "document", "document identity is unavailable")
        if not isinstance(frame, str) or not isinstance(loader, str):
            raise self._fail(operation, "worker_failure", "document", "document identity is unavailable")
        return f"{frame}:{loader}"[:MAX_STRING]

    def _public_tab(self, tab: _RuntimeTab) -> dict[str, Any]:
        opener_tab = self._target_to_tab.get(tab.opener_target_id or "")
        return {
            "tabId": tab.tab_id,
            "kind": "popup" if tab.kind == "popup" else "page",
            **({"openerTabId": opener_tab} if opener_tab else {}),
            "selected": tab.tab_id == self._selected_tab_id,
            "owned": True,
            "url": tab.url,
            "origin": _origin(tab.url) or "",
            "title": tab.title,
        }

    def _collect_events(self) -> None:
        drain = getattr(self.backend, "drain_events", None)
        if not callable(drain):
            return
        try:
            events = drain()
        except Exception:
            return
        if isinstance(events, list):
            active_session = getattr(self.backend, "_session_id", None)
            supported = {"Runtime.consoleAPICalled", "Network.requestWillBeSent",
                         "Network.loadingFailed", "Network.responseReceived"}
            for event in events[:MAX_EVENTS]:
                if not isinstance(event, dict) or event.get("method") not in supported:
                    continue
                event_session = event.get("session_id")
                # The pinned daemon includes the CDP session on every tapped
                # event. Unattributed events are fail-closed: retaining them
                # could mix a prior tab's private diagnostics into this target.
                if not active_session or not isinstance(event_session, str) or event_session != active_session:
                    continue
                method = event.get("method")
                params = event.get("params") if isinstance(event.get("params"), dict) else {}
                safe_params: dict[str, Any] = {}
                if method == "Runtime.consoleAPICalled":
                    level = params.get("type")
                    if level not in {"error", "warning", "warn"}:
                        continue
                    safe_args: list[dict[str, str]] = []
                    args = params.get("args")
                    if isinstance(args, list):
                        for arg in args[:8]:
                            if not isinstance(arg, dict):
                                continue
                            value = arg.get("value")
                            description = arg.get("description")
                            if isinstance(value, (str, int, float, bool)):
                                safe_args.append({"value": _safe_event_message(str(value))})
                            elif isinstance(description, str):
                                safe_args.append({"value": _safe_event_message(description)})
                    safe_params = {"type": level, "args": safe_args}
                elif method == "Network.requestWillBeSent":
                    request_id = params.get("requestId")
                    request = params.get("request") if isinstance(params.get("request"), dict) else {}
                    url = request.get("url")
                    origin = _origin(url) if isinstance(url, str) else None
                    if not isinstance(request_id, str) or not origin:
                        continue
                    safe_params = {"requestId": request_id,
                                   "request": {"url": origin, "method": str(request.get("method") or "GET")[:32]}}
                elif method == "Network.loadingFailed":
                    url = params.get("url")
                    origin = _origin(url) if isinstance(url, str) else None
                    safe_params = {"requestId": str(params.get("requestId") or "")[:200],
                                   "url": origin or "", "blockedReason": bool(params.get("blockedReason"))}
                else:
                    response = params.get("response") if isinstance(params.get("response"), dict) else {}
                    safe_params = {"requestId": str(params.get("requestId") or "")[:200],
                                   "response": {"status": response.get("status")}}
                # Preserve provenance with the sanitized event. The daemon's
                # drain buffer is persistent across target switches; without
                # retaining this session ID, evidence on a later tab could
                # replay private diagnostics collected from an earlier tab.
                self._event_buffer.append({
                    "method": method, "params": safe_params,
                    "session_id": active_session,
                })
            if len(self._event_buffer) > MAX_EVENTS:
                del self._event_buffer[:-MAX_EVENTS]

    def _read_text(self) -> tuple[str, bool]:
        getter = getattr(self.backend, "visible_text_content", None)
        if not callable(getter):
            # Offline backends may expose the safe content under this explicit
            # seam; never fall back to a caller-provided predicate or script.
            getter = getattr(self.backend, "read_visible_text", None)
        if not callable(getter):
            raise self._fail("read", "unsupported", "visibleText", "visible page text is unavailable")
        try:
            value = getter()
        except BrowserOperationError:
            raise
        except Exception:
            raise self._fail("read", "worker_failure", "visibleText", "visible page text is unavailable")
        if not isinstance(value, str):
            raise self._fail("read", "worker_failure", "visibleText", "visible page text is unavailable")
        # Remove control characters that can forge protocol lines, preserving
        # ordinary Unicode and the rendered line structure.
        clean = "".join(ch for ch in value if ch in "\t\n" or ord(ch) >= 0x20 and ch != "\x7f")
        truncated = len(clean) > MAX_READ_TEXT
        return clean[:MAX_READ_TEXT], truncated

    def list_tabs(self) -> list[dict[str, Any]]:
        self._refresh_tabs()
        # Sort by safe public fields; raw target IDs never determine ordering.
        return [self._public_tab(tab) for tab in sorted(
            (tab for tab in self._tabs.values() if _approved_origin(_origin(tab.url), self.allowed_origins)),
            key=lambda item: (item.kind, _origin(item.url) or "", item.url, item.title)
        )]

    def apply_context(self, context: dict[str, Any] | None, operation: str = "narrowOriginPolicy") -> None:
        """Set one effective policy for a single serialized dispatch."""
        if context is None:
            self.allowed_origins = self._base_policy
            return
        if not isinstance(context, dict) or set(context) != {"allowedOrigins"}:
            raise self._fail(operation, "invalid_request", "policy", "invalid operation context")
        try:
            policy = normalize_allowed_origins(context["allowedOrigins"])
        except (ValueError, TypeError):
            raise self._fail(operation, "invalid_request", "policy", "origin policy is invalid")
        if self._base_policy != "*" and (policy == "*" or not policy.issubset(self._base_policy)):
            raise self._fail(operation, "policy_denied", "policy", "origin policy cannot broaden the session policy")
        self.allowed_origins = policy

    def narrow_origin_policy(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) != {"allowedOrigins"}:
            raise self._fail("narrowOriginPolicy", "invalid_request", "policy", "invalid origin policy request")
        try:
            requested = normalize_allowed_origins(request["allowedOrigins"])
        except (ValueError, TypeError):
            raise self._fail("narrowOriginPolicy", "invalid_request", "policy", "origin policy is invalid")
        if self._base_policy != "*" and (requested == "*" or not requested.issubset(self._base_policy)):
            raise self._fail("narrowOriginPolicy", "policy_denied", "policy", "origin policy cannot broaden the session policy")
        # If this request itself is context-scoped, the returned context must
        # remain within that effective policy too. It is never persisted.
        if self.allowed_origins != "*" and (requested == "*" or not requested.issubset(self.allowed_origins)):
            raise self._fail("narrowOriginPolicy", "policy_denied", "policy", "origin policy cannot broaden the operation policy")
        return {"allowedOrigins": "*" if requested == "*" else sorted(requested)}

    def observe(self) -> dict[str, Any]:
        self._resolve_tab({}, "observe")
        try:
            return super().observe()
        except WorkerError as exc:
            if isinstance(exc, BrowserOperationError):
                raise
            message = str(exc).lower()
            code = "stale" if "stale" in message else "policy_denied" if "outside" in message else "worker_failure"
            raise self._fail("observe", code, "observe", "browser observation failed")

    def act(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict):
            raise self._fail("act", "invalid_request", "request", "invalid action request")
        self._resolve_tab({}, "act")
        try:
            return super().act(request)
        except BrowserOperationError:
            raise
        except WorkerError as exc:
            # The compatibility worker's policy fuse is scoped to its current
            # effective policy. A narrower context must not poison the
            # immutable base when a page redirects outside that context.
            if self._blocked:
                try:
                    current = self.backend.current_tab()
                    current_url = current.get("url") if isinstance(current, dict) else None
                    if _approved_origin(_origin(current_url), self._base_policy):
                        self._blocked = False
                except Exception:
                    pass
            # The explicit backend boundary, not arbitrary exception text,
            # decides whether a dispatched mutation may have executed.
            message = str(exc)
            lowered = message.lower()
            state = "unknown" if self._mutation_dispatched and not self._mutation_verified else "not_attempted"
            code = "stale" if "stale" in lowered and not self._mutation_dispatched else "policy_denied" if "outside" in lowered else "worker_failure"
            raise self._fail("act", code, "act", "browser action failed",
                             mutation_state=state, retry="never")
        except Exception:
            # Backend exceptions are opaque by design; the dispatch flag still
            # preserves the conservative unknown-mutation classification.
            state = "unknown" if self._mutation_dispatched and not self._mutation_verified else "not_attempted"
            raise self._fail("act", "worker_failure", "act", "browser action failed",
                             mutation_state=state, retry="never")
        except BaseException as exc:
            # Cancellation can be raised as a BaseException by an embedding
            # host. Preserve process signals, but classify post-dispatch
            # cancellation exactly like any other uncertain mutation.
            if isinstance(exc, (KeyboardInterrupt, SystemExit)):
                raise
            state = "unknown" if self._mutation_dispatched and not self._mutation_verified else "not_attempted"
            raise self._fail("act", "aborted", "act", "browser action was cancelled",
                             mutation_state=state, retry="never")

    def read(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) - {"tabId"}:
            raise self._fail("read", "invalid_request", "request", "invalid read request")
        tab = self._resolve_tab(request, "read")
        self._collect_events()
        try:
            current = self.backend.current_tab()
            url = current.get("url")
            title = current.get("title") or ""
        except Exception:
            raise self._fail("read", "worker_failure", "state", "page state is unavailable")
        if not isinstance(url, str) or not _approved_origin(_origin(url), self.allowed_origins):
            raise self._fail("read", "policy_denied", "policy", "page is outside the approved origins")
        text, truncated = self._read_text()
        return {"tab": self._public_tab(tab), "documentId": self._document_id("read"),
                "url": url, "title": _short(title), "visibleText": text,
                "visibleTextTruncated": truncated}

    def assert_page(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) - {"tabId", "assertion", "wait"} or "assertion" not in request or request.get("wait") not in {"none", "bounded"}:
            raise self._fail("assert", "invalid_request", "request", "invalid assertion request")
        assertion = request["assertion"]
        if not isinstance(assertion, dict) or set(assertion) != {"kind", "match", "value"}:
            raise self._fail("assert", "invalid_request", "assertion", "invalid assertion")
        kind, match, value = assertion.get("kind"), assertion.get("match"), assertion.get("value")
        valid = ((kind == "url" and match in {"exact", "origin"}) or
                 (kind == "title" and match in {"exact", "contains"}) or
                 (kind == "visibleText" and match == "contains"))
        if not valid or not isinstance(value, str) or not value or len(value) > MAX_STRING:
            raise self._fail("assert", "invalid_request", "assertion", "invalid assertion")
        if kind == "url" and (_origin(value) is None or (match == "exact" and not _approved_origin(_origin(value), self.allowed_origins))):
            raise self._fail("assert", "policy_denied", "assertion", "assertion URL is outside the approved origins")
        deadline = time.monotonic() + (READ_ASSERTION_TIMEOUT_SECONDS if request["wait"] == "bounded" else 0)
        last: dict[str, Any] | None = None
        while True:
            last = self.read({"tabId": request.get("tabId")} if "tabId" in request else {})
            actual = last["url"] if kind == "url" else last["title"] if kind == "title" else last["visibleText"]
            if kind == "url" and match == "origin":
                satisfied = _origin(actual) == _origin(value)
            elif match == "contains":
                satisfied = value in actual
            else:
                satisfied = actual == value
            if satisfied:
                status = "satisfied"
                break
            if request["wait"] == "none":
                status = "not_satisfied"
                break
            if time.monotonic() >= deadline:
                status = "timed_out"
                break
            time.sleep(0.025)
        return {"status": status, "readOnly": True, "tab": last["tab"], "documentId": last["documentId"]}

    def _wait_navigation(self, operation: str) -> None:
        wait_ready = getattr(self.backend, "wait_ready", None)
        if callable(wait_ready):
            try:
                ready = wait_ready(NAVIGATION_TIMEOUT_SECONDS)
            except Exception:
                raise self._fail(operation, "worker_failure", "wait", "navigation readiness is unavailable", mutation_state="unknown")
            except BaseException as exc:
                if isinstance(exc, (KeyboardInterrupt, SystemExit)):
                    raise
                raise self._fail(operation, "aborted", "wait", "navigation readiness was cancelled", mutation_state="unknown")
            if ready is False:
                raise self._fail(operation, "timeout", "wait", "navigation did not complete", mutation_state="unknown")

    def navigate(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) - {"tabId", "url"} or not isinstance(request.get("url"), str):
            raise self._fail("navigate", "invalid_request", "request", "invalid navigation request")
        url = request["url"]
        if len(url) > MAX_STRING or _origin(url) is None or not _approved_origin(_origin(url), self.allowed_origins):
            raise self._fail("navigate", "policy_denied", "policy", "navigation URL is outside the approved origins")
        tab = self._resolve_tab(request, "navigate")
        nav = getattr(self.backend, "navigate", None)
        if not callable(nav):
            raise self._fail("navigate", "unsupported", "navigate", "navigation is unavailable")
        self._mutation_dispatched = False
        self._mutation_verified = False
        try:
            self._mutation_dispatched = True
            nav(url)
        except Exception:
            raise self._fail("navigate", "worker_failure", "navigate", "navigation failed",
                             mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
        except BaseException as exc:
            if isinstance(exc, (KeyboardInterrupt, SystemExit)):
                raise
            raise self._fail("navigate", "aborted", "navigate", "navigation was cancelled",
                             mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
        self._wait_navigation("navigate")
        self._refresh_tabs()
        current = self._tabs.get(tab.tab_id)
        if current is None or not _approved_origin(_origin(current.url), self.allowed_origins):
            raise self._fail("navigate", "policy_denied", "redirect", "navigation left the approved origins", mutation_state="unknown")
        self._attach(tab.target_id, "navigate")
        self._mutation_verified = True
        return {"status": "committed", "mutationState": "verified", "tab": self._public_tab(current),
                "documentId": self._document_id("navigate")}

    def reload(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) - {"tabId"}:
            raise self._fail("reload", "invalid_request", "request", "invalid reload request")
        tab = self._resolve_tab(request, "reload")
        reload_page = getattr(self.backend, "reload", None)
        if not callable(reload_page):
            raise self._fail("reload", "unsupported", "reload", "reload is unavailable")
        self._mutation_dispatched = False
        self._mutation_verified = False
        try:
            self._mutation_dispatched = True
            reload_page()
        except Exception:
            raise self._fail("reload", "worker_failure", "reload", "reload failed",
                             mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
        except BaseException as exc:
            if isinstance(exc, (KeyboardInterrupt, SystemExit)):
                raise
            raise self._fail("reload", "aborted", "reload", "reload was cancelled",
                             mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
        self._wait_navigation("reload")
        self._refresh_tabs()
        current = self._tabs.get(tab.tab_id)
        if current is None:
            raise self._fail("reload", "no_tab", "reload", "tab closed during reload", mutation_state="unknown")
        self._attach(tab.target_id, "reload")
        self._mutation_verified = True
        return {"status": "committed", "mutationState": "verified", "tab": self._public_tab(current),
                "documentId": self._document_id("reload")}

    def select_tab(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) != {"tabId"} or not isinstance(request.get("tabId"), str):
            raise self._fail("selectTab", "invalid_request", "request", "invalid tab selection request")
        self._refresh_tabs()
        tab = self._tabs.get(request["tabId"])
        if tab is None:
            raise self._fail("selectTab", "no_tab", "target", "tab handle is stale or unavailable")
        if not _approved_origin(_origin(tab.url), self.allowed_origins):
            raise self._fail("selectTab", "policy_denied", "policy", "tab is outside the approved origins")
        self._attach(tab.target_id, "selectTab")
        self._selected_tab_id = tab.tab_id
        return {"selected": True, "tab": self._public_tab(tab)}

    def wait_popup(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) != {"openerTabId"} or not isinstance(request.get("openerTabId"), str):
            raise self._fail("waitPopup", "invalid_request", "request", "invalid popup request")
        # Refresh first: an immediate popup may already be in the target
        # inventory before waitPopup starts. Newly discovered popup candidates
        # are tracked separately from baseline/previously claimed targets.
        self._refresh_tabs()
        opener = self._tabs.get(request["openerTabId"])
        if opener is None:
            raise self._fail("waitPopup", "no_tab", "target", "opener tab is stale or unavailable")
        if not _approved_origin(_origin(opener.url), self.allowed_origins):
            raise self._fail("waitPopup", "policy_denied", "policy", "opener tab is outside the approved origins")
        deadline = time.monotonic() + POPUP_WAIT_SECONDS
        while True:
            self._refresh_tabs()
            for target in list(self._popup_candidates):
                tab_id = self._target_to_tab.get(target)
                tab = self._tabs.get(tab_id) if tab_id else None
                if (tab is not None and target not in self._claimed_popups
                        and tab.opener_target_id == opener.target_id
                        and _approved_origin(_origin(tab.url), self.allowed_origins)):
                    self._popup_candidates.discard(target)
                    self._claimed_popups.add(target)
                    return {"status": "found", "tab": self._public_tab(tab), "autoSwitched": False}
            if time.monotonic() >= deadline:
                return {"status": "timed_out", "autoSwitched": False}
            time.sleep(0.025)

    def close_tab(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) != {"tabId"} or not isinstance(request.get("tabId"), str):
            raise self._fail("closeTab", "invalid_request", "request", "invalid tab close request")
        self._refresh_tabs()
        tab = self._tabs.get(request["tabId"])
        if tab is None:
            raise self._fail("closeTab", "no_tab", "target", "tab handle is stale or unavailable")
        if not _approved_origin(_origin(tab.url), self.allowed_origins):
            raise self._fail("closeTab", "policy_denied", "policy", "tab is outside the approved origins")
        close = getattr(self.backend, "close_tab", None)
        if not callable(close):
            raise self._fail("closeTab", "unsupported", "close", "tab closing is unavailable")
        self._mutation_dispatched = False
        self._mutation_verified = False
        try:
            self._mutation_dispatched = True
            close(tab.target_id)
        except Exception:
            raise self._fail("closeTab", "worker_failure", "close", "tab close failed",
                             mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
        except BaseException as exc:
            if isinstance(exc, (KeyboardInterrupt, SystemExit)):
                raise
            raise self._fail("closeTab", "aborted", "close", "tab close was cancelled",
                             mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
        # Do not assume a close acknowledgement means the target disappeared;
        # verify inventories until the target is gone or the existing action
        # budget expires. The close mutation itself is dispatched exactly once.
        deadline = time.monotonic() + ACTION_TIMEOUT_SECONDS
        while True:
            try:
                self._refresh_tabs()
            except BrowserOperationError:
                raise self._fail("closeTab", "worker_failure", "verify", "tab close could not be verified", mutation_state="unknown")
            if tab.target_id not in self._target_to_tab:
                break
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise self._fail("closeTab", "worker_failure", "verify", "tab close could not be verified", mutation_state="unknown")
            time.sleep(min(0.025, remaining))
        self._target_to_tab.pop(tab.target_id, None)
        self._tabs.pop(tab.tab_id, None)
        if self._selected_tab_id == tab.tab_id:
            self._selected_tab_id = None
        self._mutation_verified = True
        return {"closed": True, "tabId": tab.tab_id, "mutationState": "verified"}

    def _summarize_events(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
        consoles: list[dict[str, Any]] = []
        failures: list[dict[str, Any]] = []
        request_by_id: dict[str, tuple[str, str]] = {}
        active_session = getattr(self.backend, "_session_id", None)
        if not isinstance(active_session, str) or not active_session:
            return consoles, failures
        for event in self._event_buffer[-MAX_EVENTS:]:
            if not isinstance(event, dict) or event.get("session_id") != active_session:
                continue
            method = event.get("method") if isinstance(event, dict) else None
            params = event.get("params") if isinstance(event, dict) else None
            if not isinstance(params, dict):
                continue
            if method == "Runtime.consoleAPICalled":
                level = params.get("type")
                if level not in {"error", "warning", "warn"}:
                    continue
                args = params.get("args")
                pieces: list[str] = []
                if isinstance(args, list):
                    for arg in args[:8]:
                        if isinstance(arg, dict):
                            value = arg.get("value")
                            if isinstance(value, (str, int, float, bool)):
                                pieces.append(str(value))
                            elif isinstance(arg.get("description"), str):
                                pieces.append(arg["description"])
                message = _policy_event_message(
                    _safe_event_message(" ".join(pieces) or "console event"),
                    self.allowed_origins,
                )
                existing = next((item for item in consoles if item["level"] == ("warning" if level == "warn" else level) and item["message"] == message), None)
                if existing:
                    existing["count"] += 1
                elif len(consoles) < MAX_CONSOLE_FAILURES:
                    consoles.append({"level": "warning" if level == "warn" else level,
                                     "message": message, "count": 1, "truncated": len(message) >= MAX_EVENT_MESSAGE})
            elif method == "Network.requestWillBeSent":
                request_id = params.get("requestId")
                request = params.get("request")
                if isinstance(request_id, str) and isinstance(request, dict):
                    url = request.get("url")
                    origin = _origin(url) if isinstance(url, str) else None
                    if origin and _approved_origin(origin, self.allowed_origins):
                        request_by_id[request_id] = (str(request.get("method") or "GET")[:32], origin)
            elif method == "Network.loadingFailed":
                request_id = params.get("requestId")
                method_origin = request_by_id.get(request_id) if isinstance(request_id, str) else None
                if method_origin is None and isinstance(params.get("url"), str):
                    origin = _origin(params["url"])
                    if origin and _approved_origin(origin, self.allowed_origins):
                        method_origin = ("GET", origin)
                if method_origin and len(failures) < MAX_NETWORK_FAILURES:
                    failures.append({"method": method_origin[0], "origin": method_origin[1],
                                     "category": "blocked" if params.get("blockedReason") else "network"})
            elif method == "Network.responseReceived":
                request_id = params.get("requestId")
                response = params.get("response")
                method_origin = request_by_id.get(request_id) if isinstance(request_id, str) else None
                if method_origin and isinstance(response, dict):
                    status = response.get("status")
                    if isinstance(status, (int, float)) and status >= 400 and len(failures) < MAX_NETWORK_FAILURES:
                        failures.append({"method": method_origin[0], "origin": method_origin[1],
                                         "status": int(status), "category": "http"})
        return consoles, failures

    def capture_evidence(self, request: dict[str, Any]) -> dict[str, Any]:
        if not isinstance(request, dict) or set(request) - {"runId", "reason", "operation"} or not isinstance(request.get("runId"), str) or request.get("reason") not in {"failure", "requested"}:
            raise self._fail("captureEvidence", "invalid_request", "request", "invalid evidence request")
        run_id = request["runId"]
        try:
            _safe_component(run_id, "run id")
        except WorkerError:
            raise self._fail("captureEvidence", "invalid_request", "request", "invalid evidence request")
        operations = {"observe", "act", "read", "assert", "navigate", "reload", "listTabs", "selectTab", "waitPopup", "closeTab", "captureEvidence", "narrowOriginPolicy", "close"}
        if request.get("operation") is not None and request.get("operation") not in operations:
            raise self._fail("captureEvidence", "invalid_request", "request", "invalid evidence operation")
        self._collect_events()
        consoles, failures = self._summarize_events()
        screenshot_payload: dict[str, Any] | None = None
        diagnostics_payload: dict[str, Any] | None = None
        status = "complete"
        target_available = self._selected_tab_id is not None
        if target_available:
            try:
                self._resolve_tab({}, "captureEvidence")
            except BrowserOperationError:
                target_available = False
        else:
            status = "partial"
        shot = getattr(self.backend, "screenshot_bytes", None)
        if callable(shot) and target_available:
            try:
                data = shot()
                if not isinstance(data, (bytes, bytearray)):
                    raise WorkerError("screenshot is unavailable")
                raw = bytes(data)
                clipped = raw[:MAX_SCREENSHOT_BYTES]
                screenshot_payload = {
                    "kind": "screenshot", "contentType": "image/png",
                    "encoding": "base64", "dataBase64": base64.b64encode(clipped).decode("ascii"),
                    "byteLength": len(clipped), "truncated": len(raw) > len(clipped),
                }
                if len(raw) > len(clipped):
                    status = "partial"
            except Exception:
                status = "partial"
        else:
            status = "partial"
        try:
            payload = json.dumps({"console": consoles, "failedRequests": failures},
                                 ensure_ascii=False, separators=(",", ":")).encode("utf-8")
            # Keep the diagnostics artifact valid JSON while enforcing its
            # transport ceiling; event summaries are evidence, never a reason
            # to fail the original browser operation.
            while len(payload) > MAX_DIAGNOSTICS_BYTES and (consoles or failures):
                if len(consoles) >= len(failures) and consoles:
                    consoles.pop()
                elif failures:
                    failures.pop()
                payload = json.dumps({"console": consoles, "failedRequests": failures},
                                     ensure_ascii=False, separators=(",", ":")).encode("utf-8")
            if len(payload) > MAX_DIAGNOSTICS_BYTES:
                raise WorkerError("diagnostics exceed their size limit")
            diagnostics_payload = {
                "kind": "diagnostics", "contentType": "application/json",
                "encoding": "base64", "dataBase64": base64.b64encode(payload).decode("ascii"),
                "byteLength": len(payload), "truncated": False,
            }
        except Exception:
            status = "partial" if status == "complete" else status
        if screenshot_payload is None and diagnostics_payload is None:
            status = "unavailable"
        return {"status": status,
                **({"screenshot": screenshot_payload} if screenshot_payload else {}),
                **({"diagnostics": diagnostics_payload} if diagnostics_payload else {}),
                "console": consoles, "failedRequests": failures}

    def close(self) -> None:
        if self._closed_runtime:
            return
        self._closed_runtime = True
        super().close()


# Descriptive aliases for adapters that refer to the published runtime rather
# than the historical observe/act worker name.
BrowserRuntime = BrowserRuntimeWorker
RuntimeWorker = BrowserRuntimeWorker


def _response(request_id: Any, *, result: Any = None, error: str | None = None,
              details: dict[str, Any] | None = None) -> dict[str, Any]:
    if type(request_id) is not int or request_id < 0:
        raise WorkerError("invalid request id")
    if error is not None:
        response = {"id": request_id, "ok": False, "error": str(error)[:MAX_STRING]}
        if details is not None:
            response["details"] = details
        return response
    return {"id": request_id, "ok": True, "result": result}


def process_request(worker: BrowserWorker, request: Any) -> dict[str, Any]:
    runtime_worker = isinstance(worker, BrowserRuntimeWorker)
    startup_worker = isinstance(worker, StartupFailureWorker)
    if not isinstance(request, dict) or type(request.get("id")) is not int:
        raise WorkerError("malformed request")
    if runtime_worker or (startup_worker and "operation" in request):
        if "operation" not in request:
            raise WorkerError("malformed transport request")
        op = request.get("operation", {}).get("op") if isinstance(request.get("operation"), dict) else None
    else:
        if "op" not in request:
            raise WorkerError("malformed request")
        op = request.get("op")
    request_id = request["id"]
    if isinstance(worker, BrowserRuntimeWorker):
        # Published transport shape: operation is the typed BrowserOperation
        # union and context is an optional one-dispatch policy narrowing.
        if (set(request) - {"id", "operation", "context"} or "operation" not in request
                or not isinstance(request["operation"], dict)
                or ("context" in request and request["context"] is None)):
            raise WorkerError("malformed transport request")
        operation = request["operation"]
        if set(operation) - {"op", "request"} or not isinstance(operation.get("op"), str):
            raise WorkerError("malformed transport operation")
        operation_name = operation["op"]
        if operation_name in {"observe", "listTabs", "close"} and set(operation) != {"op"}:
            raise WorkerError("malformed transport operation")
        if operation_name not in {"observe", "listTabs", "close"} and (set(operation) != {"op", "request"} or not isinstance(operation.get("request"), dict)):
            raise WorkerError("malformed transport operation")
        worker.apply_context(request.get("context"), operation_name)
        try:
            body = operation.get("request")
            if operation_name == "observe": result = worker.observe()
            elif operation_name == "act": result = worker.act(body)
            elif operation_name == "read": result = worker.read(body)
            elif operation_name == "assert": result = worker.assert_page(body)
            elif operation_name == "navigate": result = worker.navigate(body)
            elif operation_name == "reload": result = worker.reload(body)
            elif operation_name == "listTabs": result = worker.list_tabs()
            elif operation_name == "selectTab": result = worker.select_tab(body)
            elif operation_name == "waitPopup": result = worker.wait_popup(body)
            elif operation_name == "closeTab": result = worker.close_tab(body)
            elif operation_name == "captureEvidence": result = worker.capture_evidence(body)
            elif operation_name == "narrowOriginPolicy": result = worker.narrow_origin_policy(body)
            elif operation_name == "close":
                worker.close()
                result = {"closed": True}
            else:
                raise WorkerError("unsupported transport operation")
        finally:
            # An omitted context always restores the immutable base, and a
            # supplied context never leaks into the next JSONL request.
            if not worker._closed_runtime:
                worker.apply_context(None)
        return _response(request_id, result={"op": operation_name, "result": result})
    if isinstance(worker, StartupFailureWorker) and "operation" in request:
        if set(request) - {"id", "operation", "context"}:
            raise WorkerError("malformed transport request")
        worker.observe()  # raises the bounded startup detail
    if op == "observe":
        if set(request) != {"id", "op"}:
            raise WorkerError("malformed observe request")
        return _response(request_id, result=worker.observe())
    if op == "act":
        if set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed act request")
        return _response(request_id, result=worker.act(request["request"]))
    if op == "read":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed read request")
        return _response(request_id, result=worker.read(request["request"]))
    if op == "assert":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed assert request")
        return _response(request_id, result=worker.assert_page(request["request"]))
    if op == "navigate":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed navigate request")
        return _response(request_id, result=worker.navigate(request["request"]))
    if op == "reload":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed reload request")
        return _response(request_id, result=worker.reload(request["request"]))
    if op == "listTabs":
        if not runtime_worker or set(request) != {"id", "op"}:
            raise WorkerError("malformed listTabs request")
        return _response(request_id, result=worker.list_tabs())
    if op == "selectTab":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed selectTab request")
        return _response(request_id, result=worker.select_tab(request["request"]))
    if op == "waitPopup":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed waitPopup request")
        return _response(request_id, result=worker.wait_popup(request["request"]))
    if op == "closeTab":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed closeTab request")
        return _response(request_id, result=worker.close_tab(request["request"]))
    if op == "captureEvidence":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed captureEvidence request")
        return _response(request_id, result=worker.capture_evidence(request["request"]))
    if op == "narrowOriginPolicy":
        if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
            raise WorkerError("malformed narrowOriginPolicy request")
        return _response(request_id, result=worker.narrow_origin_policy(request["request"]))
    if op == "close":
        if set(request) != {"id", "op"}:
            raise WorkerError("malformed close request")
        worker.close()
        return _response(request_id, result={"closed": True})
    raise WorkerError("unsupported operation")


def _json_line(value: dict[str, Any]) -> bytes:
    raw = json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
    if len(raw) > MAX_OUTPUT_BYTES:
        raise WorkerError("response exceeds protocol limit")
    return raw


def run_stream(worker: BrowserWorker, reader: Any = None, writer: Any = None) -> int:
    reader = reader or sys.stdin.buffer
    writer = writer or sys.stdout.buffer
    while True:
        raw = reader.readline(MAX_LINE_BYTES + 1)
        if not raw:
            break
        request: Any = None
        if len(raw) > MAX_LINE_BYTES:
            response = _response(0, error="request exceeds protocol limit")
            writer.write(_json_line(response)); writer.flush()
            return 1
        try:
            request = json.loads(raw.decode("utf-8"))
            response = process_request(worker, request)
        except Exception as exc:
            request_id = request.get("id", 0) if isinstance(request, dict) else 0
            safe_error = str(exc) if isinstance(exc, WorkerError) else "browser operation failed; no action was retried"
            response = _response(max(0, request_id) if type(request_id) is int else 0,
                                 error=safe_error[:MAX_STRING],
                                 details=exc.details if isinstance(exc, BrowserOperationError) else None)
        writer.write(_json_line(response)); writer.flush()
        if isinstance(request, dict) and (request.get("op") == "close" or
                isinstance(request.get("operation"), dict) and request["operation"].get("op") == "close"):
            break
    return 0


def _parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--cdp-url", required=True)
    parser.add_argument("--allowed-origins", required=True)
    return parser.parse_args(argv)


class StartupFailureWorker:
    """Keep startup failures inside bounded IPC; never expose raw stderr."""

    def __init__(self, message: str):
        self.message = message

    def observe(self) -> dict[str, Any]:
        raise WorkerError(self.message)

    def act(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def read(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def assert_page(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def navigate(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def reload(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def list_tabs(self) -> list[dict[str, Any]]:
        raise WorkerError(self.message)

    def select_tab(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def wait_popup(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def close_tab(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def capture_evidence(self, request: Any) -> dict[str, Any]:
        raise WorkerError(self.message)

    def narrow_origin_policy(self, request: Any) -> Any:
        raise WorkerError(self.message)

    def close(self) -> None:
        pass


def main(argv: list[str] | None = None) -> int:
    args = _parse_args(argv or sys.argv[1:])
    worker: BrowserWorker | None = None

    def shutdown(signum: int, _frame: Any) -> None:
        # Also interrupts startup: RealBrowserBackend catches BaseException and
        # tears down its owned daemon before this SystemExit can escape.
        raise SystemExit(128 + signum)

    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)
    try:
        cdp_url = validate_cdp_url(args.cdp_url)
        origins = json.loads(args.allowed_origins)
        normalized_origins = normalize_allowed_origins(origins)
        session_id = os.environ.get("FAST_BROWSER_SESSION_ID") or f"session-{uuid.uuid4().hex}"
        # Evidence bytes cross the bounded worker transport; the parent
        # runtime commits them to its caller-owned PrivateArtifactStore.
        worker = BrowserRuntimeWorker(
            RealBrowserBackend(cdp_url, normalized_origins), normalized_origins,
            session_id=session_id,
        )

        return run_stream(worker)
    except Exception as exc:
        detail = str(exc)[:MAX_STRING] if isinstance(exc, WorkerError) else "browser worker initialization failed"
        if isinstance(exc, ModuleNotFoundError) and getattr(exc, "name", "") == "browser_harness":
            detail = "install browser-harness in this Python environment or set FAST_BROWSER_PYTHON to its executable"
        # The bridge deliberately discards raw stderr. Reply to the queued
        # request with a safe failure instead of exiting before IPC starts.
        return run_stream(StartupFailureWorker(detail))
    finally:
        if worker is not None:
            worker.close()


if __name__ == "__main__":
    raise SystemExit(main())