okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petcut idle daemon memory: lean vm defaults, sweep gc, no per-tool copies of catalogs, encoded replay events, lazy browser plugin, 10 min kernel idlec90606d2026-10-05

"""Semantic observations and deliberately explicit actions, on a live CDP page."""

from __future__ import annotations

from typing import SupportsIndex

import asyncio
import base64
import contextlib
import copy
import math
import time
import uuid
from collections import OrderedDict
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, TYPE_CHECKING

from . import hooks, javascript as js
from .capture import CaptureLimits, FrameSession, Reference, capture_frames
from .errors import (
    ActionOutcomeUnknown,
    BrowserError,
    ConnectionLost,
    JavaScriptError,
    NotActionable,
    ProtocolError,
    SelectionError,
    StaleReference,
    UnsupportedOperation,
    WaitTimeout,
    TargetGone,
    RecoveryError,
)
from .observation import (
    Node,
    Observation,
    attributes_from_node,
    find,
)
from .events import EventScope
from .recovery import PageCDPSession

if TYPE_CHECKING:
    from .session import Browser
from .transport import CDPSession, JSON, positive


def now() -> str:
    return datetime.now(timezone.utc).isoformat()


class Page:
    def __init__(
        self, browser: Browser, target_id: str, session_id: str, *, owned: bool
    ) -> None:
        self.browser = browser
        self.target_id = target_id
        self.cdp = PageCDPSession(self, session_id)
        self.owned = owned
        self._sessions = {session_id: FrameSession(self.cdp, target_id)}
        self._tasks: set[asyncio.Task[None]] = set()
        self._snapshots: OrderedDict[str, dict[str, Reference]] = OrderedDict()
        self._lock = asyncio.Lock()
        self._settings_lock = asyncio.Lock()
        self._settings: dict[str, JSON] = {}
        self._recovering = False
        self._destroyed = False
        self.last_recovery: JSON | None = None
        self._closed = False
        self._main_frame = ""
        self.last_action: JSON | None = None
        self._frame_sessions: dict[str, str | None] = {}
        self._frame_info: dict[str, JSON] = {}

    async def _initialize(self) -> None:
        session_id = self.cdp.session_id
        assert session_id is not None
        await self._init_session(self._sessions[session_id])
        self._main_frame = self._sessions[session_id].root_frame

    async def _init_session(self, state: FrameSession) -> None:
        try:
            for domain in ("Page", "DOM", "Runtime", "Accessibility"):
                await state.cdp.send(f"{domain}.enable")
            await state.cdp.send("Page.setLifecycleEventsEnabled", {"enabled": True})
            tree = await state.cdp.send("Page.getFrameTree")
            state.root_frame = tree["frameTree"]["frame"]["id"]
            await state.cdp.send(
                "Target.setAutoAttach",
                {
                    "autoAttach": True,
                    "waitForDebuggerOnStart": False,
                    "flatten": True,
                    "filter": [{"type": "iframe", "exclude": False}, {"exclude": True}],
                },
            )
            state.ready = True
        except Exception as exc:
            state.error = type(exc).__name__ + ": " + str(exc)
            if state.cdp.session_id == self.cdp.session_id:
                raise

    def _on_event(self, method: str, params: JSON, session_id: str) -> None:
        state = self._sessions.get(session_id)
        if (
            method
            in {
                "DOM.documentUpdated",
                "Runtime.executionContextsCleared",
                "Page.frameNavigated",
            }
            and state
        ):
            state.generation += 1
        if method == "Page.frameDetached" and params.get("reason") != "swap" and state:
            state.generation += 1
        if (
            method == "Target.attachedToTarget"
            and params.get("targetInfo", {}).get("type") == "iframe"
        ):
            sid = params["sessionId"]
            if sid in self._sessions:
                return
            child = FrameSession(
                CDPSession(self.browser._connection, sid),
                params["targetInfo"]["targetId"],
            )
            self._sessions[sid] = child
            self.browser._routes[sid] = self
            task = asyncio.create_task(
                self._init_session(child), name="browser-iframe-init"
            )
            self._tasks.add(task)
            task.add_done_callback(self._tasks.discard)

    def _detached(self, sid: str) -> None:
        state = self._sessions.pop(sid, None)
        if state is not None:
            state.generation += 1
        if sid == self.cdp.session_id:
            self._invalidate()

    def _invalidate(self, *, destroyed: bool = False) -> None:
        """Invalidate a whole attachment, including child sessions and references."""
        self._closed = True
        self._destroyed = self._destroyed or destroyed
        self._snapshots.clear()
        for task in self._tasks:
            task.cancel()
        for sid, state in tuple(self._sessions.items()):
            state.generation += 1
            state.cdp.connection.detach_session(sid)
            if self.browser._routes.get(sid) is self:
                self.browser._routes.pop(sid, None)
        self._sessions.clear()
        self._frame_sessions.clear()
        self._frame_info.clear()
        self._main_frame = ""

    async def _stop_tasks(self) -> None:
        tasks = list(self._tasks)
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

    @property
    def attached(self) -> bool:
        """Local attachment status, not a network health probe."""
        return (
            not self._closed
            and not self._destroyed
            and not self._recovering
            and not self.cdp.connection.closed
            and self.cdp.connection is self.browser._connection
        )

    @property
    def recovery_settings(self) -> list[JSON]:
        """Copy of acknowledged, replayable root-page settings; not browser state."""
        return copy.deepcopy(list(self._settings.values()))

    def _check(self) -> None:
        if self._destroyed:
            raise TargetGone(
                "The original tab is closed; reattachment cannot recreate it.",
                target_id=self.target_id,
            )
        if not self.attached:
            failure = self.cdp.connection._closed
            raise ConnectionLost(
                "This page handle is detached or disconnected; use await page.reattach().",
                target_id=self.target_id,
                recovery="await page.reattach()",
                connection_error=failure.to_dict() if failure else None,
            )

    async def reattach(
        self,
        *,
        timeout: float = 15,
        restore_settings: bool = True,
        max_message_bytes: int | None = None,
    ) -> Page:
        """Recover this exact target and Python object, without navigating or retrying actions.

        Reopens a lost browser socket when needed, retaining headers and process
        ownership. Other tabs on that socket require their own explicit reattach.
        Healthy handles are no-ops. References and event streams from a replaced
        session remain invalid. Only the documented acknowledged settings replay.
        """
        positive(timeout, "timeout")
        if not isinstance(restore_settings, bool):
            raise TypeError("restore_settings must be a bool")
        if max_message_bytes is not None and (
            isinstance(max_message_bytes, bool)
            or not isinstance(max_message_bytes, int)
            or max_message_bytes < 1024
        ):
            raise ValueError("max_message_bytes must be an integer of at least 1024")
        try:
            async with asyncio.timeout(timeout):
                # The browser lock serializes connection replacement, attaching,
                # disconnecting and closing. Then serialize with actions/settings.
                async with self.browser._lifecycle_lock:
                    async with self._lock:
                        async with self._settings_lock:
                            return await self._reattach_locked(
                                restore_settings=restore_settings,
                                max_message_bytes=max_message_bytes,
                            )
        except TimeoutError as exc:
            raise RecoveryError(
                "Reattachment exceeded its deadline; no action was replayed.",
                target_id=self.target_id,
                timeout=timeout,
            ) from exc

    async def _reattach_locked(
        self, *, restore_settings: bool, max_message_bytes: int | None
    ) -> Page:
        if self._destroyed:
            raise TargetGone(
                "The original tab is closed; no replacement was created.",
                target_id=self.target_id,
            )
        if self.browser._close_task is not None:
            raise ConnectionLost(
                "Browser.close() was called; use a new browser handle."
            )
        connection = self.browser._connection
        if (
            not connection.closed
            and max_message_bytes is not None
            and max_message_bytes != connection.max_message_bytes
        ):
            raise ValueError(
                "Changing max_message_bytes requires a lost or explicitly disconnected browser socket"
            )
        if self.attached:
            return self
        other = self.browser._pages.get(self.target_id)
        if other is not None and other is not self:
            raise RecoveryError(
                "Another page handle now owns this target's attachment; use that handle.",
                target_id=self.target_id,
            )
        receipt: JSON = {
            "status": "recovering",
            "target_id": self.target_id,
            "old_session_id": self.cdp.session_id,
            "restored_methods": [],
            "event_history_gap": True,
            "started_at": now(),
            "transport_reconnected": False,
        }
        self.last_recovery = receipt
        self._recovering = True
        new_sid: str | None = None
        attach_started = False
        # _invalidate() kills the previous session's event streams and references.
        self._invalidate()
        try:
            await self._stop_tasks()
            receipt["transport_reconnected"] = await self.browser._ensure_connection(
                max_message_bytes
            )
            targets = await self.browser.pages()
            if not any(t["targetId"] == self.target_id for t in targets):
                self._destroyed = True
                raise TargetGone(
                    "The original target no longer exists; no new tab was created.",
                    target_id=self.target_id,
                )
            attach_started = True
            try:
                result = await self.browser.cdp.send(
                    "Target.attachToTarget",
                    {
                        "targetId": self.target_id,
                        "flatten": True,
                    },
                )
            except ProtocolError as exc:
                # A protocol rejection acknowledges that no attachment succeeded.
                attach_started = False
                if self._destroyed or not any(
                    t["targetId"] == self.target_id for t in await self.browser.pages()
                ):
                    self._destroyed = True
                    raise TargetGone(
                        "The target closed during reattachment; no replacement was created.",
                        target_id=self.target_id,
                    ) from exc
                raise
            new_sid = result["sessionId"]
            self.cdp = PageCDPSession(self, new_sid)
            self._sessions = {new_sid: FrameSession(self.cdp, self.target_id)}
            self.browser._pages[self.target_id] = self
            self.browser._routes[new_sid] = self
            # _closed stays True until initialization and settings both succeed.
            await self._initialize()
            if restore_settings:
                for item in self.recovery_settings:
                    try:
                        # Bypass recording: restoration is not a new configuration.
                        await self.cdp.connection.send(
                            item["method"], item["params"], session_id=new_sid
                        )
                    except BrowserError as exc:
                        raise RecoveryError(
                            "Emulation restoration failed; this handle remains detached.",
                            target_id=self.target_id,
                            failed_method=item["method"],
                            restored_methods=list(receipt["restored_methods"]),
                            cause=exc.to_dict(),
                        ) from exc
                    receipt["restored_methods"].append(item["method"])
            if (
                self.cdp.connection.closed
                or new_sid not in self._sessions
                or self._destroyed
            ):
                raise ConnectionLost(
                    "Target detached again during reattachment.",
                    target_id=self.target_id,
                )
            self._closed = False
            receipt.update(
                status="reattached",
                new_session_id=new_sid,
                finished_at=now(),
                settings_restored=restore_settings,
            )
            return self
        except BaseException as exc:
            receipt.update(
                status="cancelled"
                if isinstance(exc, asyncio.CancelledError)
                else "failed",
                error=type(exc).__name__,
                finished_at=now(),
            )
            self._invalidate()
            # Bounded cleanup survives repeated interruption. Never close the tab.
            from .launch import _finish_cleanup

            async def cleanup() -> None:
                await self._stop_tasks()
                if new_sid is not None and not self.browser._connection.closed:
                    try:
                        await self.browser.cdp.send(
                            "Target.detachFromTarget", {"sessionId": new_sid}, timeout=2
                        )
                    except BrowserError:
                        # Cannot prove cleanup of the failed attachment. Release
                        # every session on this socket, while preserving tabs.
                        receipt["transport_disconnected_for_cleanup"] = True
                        await self.browser._disconnect()
                elif attach_started and not self.browser._connection.closed:
                    # The attach may have succeeded but its session ID was lost.
                    # Closing the socket releases unknown sessions without closing
                    # their tabs. Other page handles must explicitly reattach too.
                    receipt["transport_disconnected_for_cleanup"] = True
                    await self.browser._disconnect()

            await _finish_cleanup(
                asyncio.create_task(cleanup(), name="browser-reattach-cleanup")
            )
            raise
        finally:
            self._recovering = False

    def descriptor(self) -> JSON:
        """Serializable reconnect data; no sockets, headers or actionable references."""
        return {"endpoint": self.browser.endpoint, "target_id": self.target_id}

    def events(self, method: str, **options: Any) -> EventScope:
        """Root-target events only; for OOPIFs use frame_cdp() with EventScope."""
        return EventScope(self.cdp, method, **options)

    async def frames(self) -> list[JSON]:
        self._check()
        for _ in range(4):
            tasks = list(self._tasks)
            if not tasks:
                break
            await asyncio.gather(*tasks)
        frames: dict[str, JSON] = {}
        routes: dict[str, str | None] = {}
        # Root target first; an OOPIF's own session takes precedence over its placeholder.
        states = sorted(
            self._sessions.values(),
            key=lambda s: s.cdp.session_id != self.cdp.session_id,
        )
        for state in states:
            if not state.ready:
                continue
            try:
                tree = (await state.cdp.send("Page.getFrameTree"))["frameTree"]
            except ProtocolError:
                continue
            stack = [(tree, None)]
            while stack:
                entry, parent = stack.pop()
                f = entry["frame"]
                fid = f["id"]
                old = frames.get(fid, {})
                frames[fid] = {
                    "id": fid,
                    "parent_id": f.get("parentId", parent or old.get("parent_id")),
                    "url": f.get("url", ""),
                    "name": f.get("name", ""),
                    "session_id": state.cdp.session_id,
                    "out_of_process": state.cdp.session_id != self.cdp.session_id,
                }
                routes[fid] = state.cdp.session_id
                for child in reversed(entry.get("childFrames", [])):
                    stack.append((child, fid))
        self._frame_sessions = routes
        self._frame_info = frames
        return list(frames.values())

    async def frame_cdp(self, frame_id: str) -> CDPSession:
        await self.frames()
        sid = self._frame_sessions.get(frame_id)
        if sid is None:
            raise UnsupportedOperation("Frame is not attached.", frame_id=frame_id)
        return self._sessions[sid].cdp

    async def observe(
        self,
        *,
        max_nodes: int = 1500,
        max_text: int = 1500,
        max_depth: int = 50,
        frame_id: str | None = None,
        retain: int = 8,
    ) -> Observation:
        """Capture AX + safe DOM attributes into JSON-compatible records.

        max_nodes bounds retained records, NOT Chromium's full per-frame AX capture.
        max_depth and the connection's message-size ceiling bound other dimensions.
        Coverage reports omissions/races. Capture is not atomic with page scripts.
        """
        if min(max_nodes, max_text, max_depth, retain) < 1 or retain > 64:
            raise ValueError(
                "Capture limits must be positive; retain must be at most 64"
            )
        self._check()
        started = time.monotonic()
        snapshot_id = "s" + uuid.uuid4().hex[:12]
        frames = await self.frames()
        chosen = [f for f in frames if frame_id is None or f["id"] == frame_id]
        if not chosen:
            raise UnsupportedOperation(
                "Requested frame was not found.", frame_id=frame_id
            )
        warnings: list[JSON] = []
        for state in self._sessions.values():
            if not state.ready:
                warnings.append(
                    {
                        "kind": "frame_not_ready",
                        "target_id": state.target_id,
                        "error": state.error,
                    }
                )
        captured = await capture_frames(
            chosen,
            self._sessions,
            CaptureLimits(snapshot_id, max_nodes, max_text, max_depth),
        )
        warnings.extend(captured.warnings)
        rows, refs = captured.rows, captured.references
        self._snapshots[snapshot_id] = refs
        while len(self._snapshots) > retain:
            self._snapshots.popitem(last=False)
        title = await self.evaluate("document.title")
        return {
            "snapshot_id": snapshot_id,
            "target_id": self.target_id,
            "url": self._frame_info.get(self._main_frame, {}).get("url", ""),
            "title": title,
            "captured_at": now(),
            "capture_ms": round((time.monotonic() - started) * 1000, 1),
            "nodes": rows,
            "frames": frames,
            "coverage": {
                "complete": not warnings,
                "warnings": warnings,
                "scope_frame_id": frame_id,
                "atomic": False,
                "max_nodes_is_retention_limit": True,
            },
        }

    def _ref(self, ref: str | Node) -> tuple[str, Reference, FrameSession]:
        self._check()
        key = ref.get("ref") if isinstance(ref, dict) else ref
        if not isinstance(key, str) or ":" not in key:
            raise StaleReference("This node has no actionable reference.")
        record = self._snapshots.get(key.split(":", 1)[0], {}).get(key)
        if record is None:
            raise StaleReference(
                "Reference is unknown, evicted, or from another connection.", ref=key
            )
        state = self._sessions.get(record.session_id)
        if state is None or state.generation != record.generation:
            raise StaleReference(
                "Document or frame changed after observation.", ref=key
            )
        return key, record, state

    async def _call(
        self,
        cdp: CDPSession,
        object_id: str,
        function: str,
        *,
        arguments: list[JSON] | None = None,
    ) -> Any:
        response = await cdp.send(
            "Runtime.callFunctionOn",
            {
                "objectId": object_id,
                "functionDeclaration": function,
                "arguments": arguments or [],
                "returnByValue": True,
                "awaitPromise": True,
            },
        )
        return self._value(response)

    @staticmethod
    def _value(response: JSON) -> Any:
        if "exceptionDetails" in response:
            detail = response["exceptionDetails"]
            raise JavaScriptError(
                "Page JavaScript threw an exception.",
                description=detail.get("exception", {}).get(
                    "description", detail.get("text")
                ),
            )
        result = response.get("result", {})
        if "unserializableValue" in result:
            raise JavaScriptError(
                "Result is not JSON-serializable; convert it explicitly in JavaScript.",
                value=result["unserializableValue"],
            )
        if "objectId" in result:
            raise JavaScriptError(
                "Result did not serialize by value; return plain JSON data."
            )
        return result.get("value")

    @asynccontextmanager
    async def _resolved(self, ref: str | Node):
        key, record, state = self._ref(ref)
        object_id = None
        try:
            try:
                node = (
                    await state.cdp.send(
                        "DOM.describeNode", {"backendNodeId": record.backend_id}
                    )
                )["node"]
                if attributes_from_node(node) != record.attributes:
                    raise StaleReference(
                        "Element identity attributes changed.", ref=key
                    )
                ax = (
                    await state.cdp.send(
                        "Accessibility.getPartialAXTree",
                        {
                            "backendNodeId": record.backend_id,
                            "fetchRelatives": False,
                        },
                    )
                )["nodes"]
                current = next(
                    (n for n in ax if n.get("backendDOMNodeId") == record.backend_id),
                    None,
                )
                if (
                    current is None
                    or current.get("ignored")
                    or current.get("role", {}).get("value") != record.role
                    or current.get("name", {}).get("value", "") != record.name
                ):
                    raise StaleReference(
                        "Element role or accessible name changed.", ref=key
                    )
                obj = (
                    await state.cdp.send(
                        "DOM.resolveNode", {"backendNodeId": record.backend_id}
                    )
                )["object"]
                object_id = obj.get("objectId")
                if not object_id:
                    raise StaleReference("Element no longer resolves.", ref=key)
                status = await self._call(state.cdp, object_id, js.STATE)
                self._ref(key)
                if not status.get("connected"):
                    raise StaleReference("Element was detached.", ref=key)
            except ProtocolError as exc:
                raise StaleReference(
                    "Element no longer resolves in its original document.", ref=key
                ) from exc
            yield key, record, state, object_id, status
        finally:
            if object_id and not self.browser._connection.closed:
                with contextlib.suppress(BrowserError):
                    await state.cdp.send(
                        "Runtime.releaseObject", {"objectId": object_id}, timeout=2
                    )

    @asynccontextmanager
    async def _action(self, action: str, ref: str | Node | None = None):
        async with self._lock:
            self._check()
            receipt: JSON = {
                "action": action,
                "ref": ref.get("ref") if isinstance(ref, dict) else ref,
                "status": "preparing",
                "may_have_executed": False,
                "started_at": now(),
            }
            self.last_action = receipt
            try:
                yield receipt
                receipt["status"] = "input_dispatched"
                receipt["finished_at"] = now()
            except BaseException as exc:
                receipt["status"] = (
                    "outcome_unknown" if receipt["may_have_executed"] else "rejected"
                )
                receipt["error"] = type(exc).__name__
                receipt["finished_at"] = now()
                if receipt["may_have_executed"] and isinstance(exc, Exception):
                    raise ActionOutcomeUnknown(
                        "Action may have happened. Observe application state before retrying.",
                        receipt=copy.deepcopy(receipt),
                        cause=exc.to_dict()
                        if isinstance(exc, BrowserError)
                        else str(exc),
                    ) from exc
                if receipt["may_have_executed"]:
                    exc.add_note(
                        "Browser input may have been dispatched; inspect page.last_action and observe before retrying."
                    )
                raise

    async def evaluate(
        self,
        expression: str,
        *,
        frame_id: str | None = None,
        timeout: float | None = None,
    ) -> Any:
        """Evaluate an expression or a zero-argument function; await promises; return data.

        This is a raw escape hatch: it may mutate the page and bypasses action locking.
        Explicitly invoke functions taking arguments in the supplied expression.
        """
        self._check()
        cdp = self.cdp
        params: JSON = {
            "expression": f"(async () => {{ const v = ({expression}\n); return typeof v === 'function' ? await v() : await v; }})()",
            "returnByValue": True,
            "awaitPromise": True,
        }
        if frame_id is not None:
            cdp = await self.frame_cdp(frame_id)
            # Isolated worlds avoid guessing a frame's default executionContextId.
            world = await cdp.send(
                "Page.createIsolatedWorld",
                {
                    "frameId": frame_id,
                    "worldName": "prime-browser-evaluate",
                },
            )
            params["contextId"] = world["executionContextId"]
        result = await cdp.send("Runtime.evaluate", params, timeout=timeout)
        return self._value(result)

    async def goto(
        self, url: str, *, wait: str = "domcontentloaded", timeout: float = 30
    ) -> JSON:
        if wait not in {"commit", "domcontentloaded", "load"}:
            raise ValueError("wait must be commit, domcontentloaded, or load")
        positive(timeout, "timeout")
        async with self._action("goto") as receipt:
            try:
                async with asyncio.timeout(timeout):
                    # Subscribe first: lifecycle events may precede the navigate reply.
                    async with self.events(
                        "Page.lifecycleEvent", enable=False
                    ) as events:
                        receipt["may_have_executed"] = True
                        result = await self.cdp.send(
                            "Page.navigate", {"url": url}, timeout=timeout
                        )
                        if result.get("errorText") or result.get("isDownload"):
                            raise BrowserError(
                                "Navigation failed or became a download.", result=result
                            )
                        loader = result.get("loaderId")
                        if wait != "commit" and loader:
                            milestone = (
                                "DOMContentLoaded"
                                if wait == "domcontentloaded"
                                else "load"
                            )
                            await events.next(
                                where=lambda e: (
                                    e.get("loaderId") == loader
                                    and e.get("name") == milestone
                                ),
                                timeout=timeout,
                            )
                        receipt["navigation"] = result
                        receipt["milestone"] = wait
            except TimeoutError as exc:
                raise WaitTimeout(
                    "Navigation did not reach its milestone before the deadline."
                ) from exc
        return copy.deepcopy(receipt)

    async def wait_for(
        self,
        *,
        timeout: float = 10,
        interval: float = 0.15,
        state: str = "present",
        **query: Any,
    ) -> Node | None:
        """Poll semantic state; no input is dispatched. 'absent' requires full coverage."""
        positive(timeout, "timeout")
        positive(interval, "interval")
        if state not in {"present", "absent"}:
            raise ValueError("state must be present or absent")
        if not query:
            raise ValueError(
                "Supply a semantic query such as role='button', name='Save'"
            )
        last_coverage = None
        try:
            async with asyncio.timeout(timeout):
                while True:
                    snapshot = await self.observe(frame_id=query.get("frame_id"))
                    matches = find(snapshot, **query)
                    last_coverage = snapshot["coverage"]
                    if state == "absent" and not matches:
                        if last_coverage["complete"]:
                            return None
                    elif state == "present" and matches:
                        if len(matches) != 1:
                            raise SelectionError(
                                "State query is ambiguous.",
                                count=len(matches),
                                query=query,
                            )
                        return matches[0]
                    await asyncio.sleep(interval)
        except TimeoutError as exc:
            raise WaitTimeout(
                "Semantic condition was not met.",
                query=query,
                state=state,
                last_coverage=last_coverage,
            ) from exc

    async def _hit(self, cdp: CDPSession, x: float, y: float, viewport: JSON) -> JSON:
        # DOM hit testing uses document coordinates; Input uses viewport coordinates.
        try:
            return await cdp.send(
                "DOM.getNodeForLocation",
                {
                    "x": int(x + viewport["pageX"]),
                    "y": int(y + viewport["pageY"]),
                    "includeUserAgentShadowDOM": True,
                },
            )
        except ProtocolError as exc:
            raise NotActionable("No hit-testable node at this point.") from exc

    async def _to_main_point(
        self, state: FrameSession, x: float, y: float
    ) -> tuple[float, float]:
        """Cross OOPIF boundaries, checking each frame's embedding for occlusion."""
        visited = set()
        while state.cdp.session_id != self.cdp.session_id:
            if state.root_frame in visited:
                raise UnsupportedOperation("Cycle in the frame ancestry map.")
            visited.add(state.root_frame)
            parent_id = self._frame_info.get(state.root_frame, {}).get("parent_id")
            parent_sid = self._frame_sessions.get(parent_id)
            parent = self._sessions.get(parent_sid)
            if parent is None:
                raise StaleReference("Frame ancestry changed; observe again.")
            owner = await parent.cdp.send(
                "DOM.getFrameOwner", {"frameId": state.root_frame}
            )
            await parent.cdp.send(
                "DOM.scrollIntoViewIfNeeded", {"backendNodeId": owner["backendNodeId"]}
            )
            box = (
                await parent.cdp.send(
                    "DOM.getBoxModel", {"backendNodeId": owner["backendNodeId"]}
                )
            )["model"]["content"]
            # Scale/translate are supported. Do not guess perspective/rotation geometry.
            if (
                abs(box[0] - box[6]) > 0.25
                or abs(box[2] - box[4]) > 0.25
                or abs(box[1] - box[3]) > 0.25
                or abs(box[5] - box[7]) > 0.25
                or box[2] <= box[0]
                or box[7] <= box[1]
            ):
                raise UnsupportedOperation(
                    "Rotated, mirrored or perspective-transformed cross-process frames are not supported."
                )
            size = self._value(
                await state.cdp.send(
                    "Runtime.evaluate",
                    {
                        "expression": "({width:innerWidth,height:innerHeight})",
                        "returnByValue": True,
                    },
                )
            )
            if not size or size["width"] <= 0 or size["height"] <= 0:
                raise NotActionable("Frame has no visible viewport.")
            x = box[0] + x * (box[2] - box[0]) / size["width"]
            y = box[1] + y * (box[7] - box[1]) / size["height"]
            metrics = await parent.cdp.send("Page.getLayoutMetrics")
            viewport = metrics["cssLayoutViewport"]
            if not (
                0 <= x < viewport["clientWidth"] and 0 <= y < viewport["clientHeight"]
            ):
                raise NotActionable(
                    "Frame click point is clipped by its parent viewport."
                )
            hit = await self._hit(parent.cdp, x, y, viewport)
            if hit.get("backendNodeId") != owner["backendNodeId"]:
                raise NotActionable("An ancestor frame is obscured by another element.")
            state = parent
        return x, y

    async def _click_point(
        self, record: Reference, state: FrameSession, object_id: str
    ) -> tuple[float, float]:
        await state.cdp.send(
            "DOM.scrollIntoViewIfNeeded", {"backendNodeId": record.backend_id}
        )
        quads = (
            await state.cdp.send(
                "DOM.getContentQuads", {"backendNodeId": record.backend_id}
            )
        )["quads"]
        metrics = await state.cdp.send("Page.getLayoutMetrics")
        viewport = metrics["cssLayoutViewport"]
        if metrics["cssVisualViewport"].get("scale", 1) != 1:
            raise UnsupportedOperation(
                "Pinch-zoomed viewports are not supported by high-level click."
            )
        width, height = viewport["clientWidth"], viewport["clientHeight"]
        covering: JSON = {}
        for quad in quads:
            xs, ys = quad[0::2], quad[1::2]
            left, right = max(0, min(xs)), min(width, max(xs))
            top, bottom = max(0, min(ys)), min(height, max(ys))
            if right - left < 1 or bottom - top < 1:
                continue
            x, y = (left + right) / 2, (top + bottom) / 2
            try:
                hit = await self._hit(state.cdp, x, y, viewport)
            except NotActionable:
                continue
            matched = hit.get("backendNodeId") == record.backend_id
            if not matched:
                obj = (
                    await state.cdp.send(
                        "DOM.resolveNode", {"backendNodeId": hit["backendNodeId"]}
                    )
                )["object"].get("objectId")
                if obj:
                    try:
                        matched = bool(
                            await self._call(
                                state.cdp,
                                object_id,
                                js.CONTAINS,
                                arguments=[{"objectId": obj}],
                            )
                        )
                    except (JavaScriptError, ProtocolError):
                        pass
                    finally:
                        await state.cdp.send("Runtime.releaseObject", {"objectId": obj})
            if matched:
                return await self._to_main_point(state, x, y)
            covering = await self._ax_summary(state, hit.get("backendNodeId"))
        raise NotActionable(
            "No unobstructed, visible click point was found"
            + ("; something else is over the target." if covering else "."),
            **({"covering": covering} if covering else {}),
        )

    async def _ax_summary(self, state: FrameSession, backend_id: int | None) -> JSON:
        """Name what is over a target, so "something covers it" is actionable."""
        if backend_id is None:
            return {}
        try:
            nodes = (
                await state.cdp.send(
                    "Accessibility.getPartialAXTree",
                    {
                        "backendNodeId": backend_id,
                        "fetchRelatives": False,
                    },
                )
            )["nodes"]
        except ProtocolError:
            return {"backendNodeId": backend_id}
        node = next((n for n in nodes if n.get("backendDOMNodeId") == backend_id), None)
        if node is None:
            return {"backendNodeId": backend_id}
        return {
            "backendNodeId": backend_id,
            "role": str(node.get("role", {}).get("value", "")),
            "name": str(node.get("name", {}).get("value", ""))[:80],
        }

    async def click(self, ref: str | Node, *, button: str = "left") -> JSON:
        if button not in {"left", "right", "middle"}:
            raise ValueError("button must be left, right, or middle")
        async with self._action("click", ref) as receipt:
            async with self._resolved(ref) as (key, record, state, obj, status):
                if status["disabled"]:
                    raise NotActionable("Element is disabled.", ref=key)
                x, y = await self._click_point(record, state, obj)
                self._ref(key)
                receipt["may_have_executed"] = True
                pressed = False
                try:
                    await self.cdp.send(
                        "Input.dispatchMouseEvent",
                        {"type": "mouseMoved", "x": x, "y": y},
                    )
                    pressed = True
                    await self.cdp.send(
                        "Input.dispatchMouseEvent",
                        {
                            "type": "mousePressed",
                            "x": x,
                            "y": y,
                            "button": button,
                            "clickCount": 1,
                        },
                    )
                    await self.cdp.send(
                        "Input.dispatchMouseEvent",
                        {
                            "type": "mouseReleased",
                            "x": x,
                            "y": y,
                            "button": button,
                            "clickCount": 1,
                        },
                    )
                    pressed = False
                finally:
                    if pressed and not self.browser._connection.closed:
                        with contextlib.suppress(Exception):
                            await self.cdp.send(
                                "Input.dispatchMouseEvent",
                                {
                                    "type": "mouseReleased",
                                    "x": x,
                                    "y": y,
                                    "button": button,
                                    "clickCount": 1,
                                },
                                timeout=1,
                            )
        return copy.deepcopy(receipt)

    async def fill(self, ref: str | Node, text: str) -> JSON:
        if not isinstance(text, str):
            raise TypeError("text must be a string")
        async with self._action("fill", ref) as receipt:
            async with self._resolved(ref) as (key, record, state, obj, status):
                if status["disabled"] or status["readonly"] or not status["visible"]:
                    held = [
                        name
                        for name, value in (
                            ("disabled", status["disabled"]),
                            ("read-only", status["readonly"]),
                            ("not visible", not status["visible"]),
                        )
                        if value
                    ]
                    raise NotActionable(
                        f"Field is {' and '.join(held)}.",
                        ref=key,
                        disabled=status["disabled"],
                        readonly=status["readonly"],
                    )
                supported = (
                    status["tag"] == "textarea"
                    or status["editable"]
                    or (
                        status["tag"] == "input"
                        and status["type"]
                        in {"text", "search", "url", "tel", "email", "password"}
                    )
                )
                if not supported:
                    raise UnsupportedOperation(
                        "fill supports text-like inputs, textarea and contenteditable only.",
                        tag=status["tag"],
                        type=status["type"],
                    )
                receipt["may_have_executed"] = True
                if not await self._call(state.cdp, obj, js.SELECT_TEXT):
                    raise NotActionable("Could not focus and select the field.")
                if text:
                    await state.cdp.send("Input.insertText", {"text": text})
                else:
                    await self._key(state.cdp, "Backspace")
                actual = await self._call(state.cdp, obj, js.READ_VALUE)
                if actual != text:
                    raise BrowserError(
                        "Field value did not match after input; no retry was attempted.",
                        expected_length=len(text),
                        actual_length=len(str(actual)),
                    )
        return copy.deepcopy(receipt)

    async def _key(self, cdp: CDPSession, key: str) -> None:
        special = {
            "Enter": ("Enter", 13, "\r"),
            "Tab": ("Tab", 9, ""),
            "Escape": ("Escape", 27, ""),
            "Backspace": ("Backspace", 8, ""),
            "Delete": ("Delete", 46, ""),
            "ArrowLeft": ("ArrowLeft", 37, ""),
            "ArrowUp": ("ArrowUp", 38, ""),
            "ArrowRight": ("ArrowRight", 39, ""),
            "ArrowDown": ("ArrowDown", 40, ""),
            "Home": ("Home", 36, ""),
            "End": ("End", 35, ""),
            "PageUp": ("PageUp", 33, ""),
            "PageDown": ("PageDown", 34, ""),
            "Space": ("Space", 32, " "),
        }
        if key not in special:
            raise ValueError(
                "Unsupported key; use named keys or raw Input.dispatchKeyEvent for chords"
            )
        code, vk, text = special[key]
        params = {
            "key": " " if key == "Space" else key,
            "code": code,
            "windowsVirtualKeyCode": vk,
        }
        try:
            await cdp.send(
                "Input.dispatchKeyEvent", {"type": "keyDown", **params, "text": text}
            )
        finally:
            if not self.browser._connection.closed:
                await cdp.send(
                    "Input.dispatchKeyEvent", {"type": "keyUp", **params}, timeout=2
                )

    async def press(self, key: str, *, ref: str | Node | None = None) -> JSON:
        if key not in {
            "Enter",
            "Tab",
            "Escape",
            "Backspace",
            "Delete",
            "ArrowLeft",
            "ArrowUp",
            "ArrowRight",
            "ArrowDown",
            "Home",
            "End",
            "PageUp",
            "PageDown",
            "Space",
        }:
            raise ValueError(
                "Unsupported key; use named keys or raw Input.dispatchKeyEvent for chords"
            )
        async with self._action("press", ref) as receipt:
            if ref is None:
                receipt["may_have_executed"] = True
                await self._key(self.cdp, key)
            else:
                async with self._resolved(ref) as (refkey, record, state, obj, status):
                    if status["disabled"] or not status["visible"]:
                        raise NotActionable(
                            "Element cannot receive keyboard input.", ref=refkey
                        )
                    receipt["may_have_executed"] = True
                    await state.cdp.send(
                        "DOM.focus", {"backendNodeId": record.backend_id}
                    )
                    await self._key(state.cdp, key)
        return copy.deepcopy(receipt)

    async def scroll(
        self,
        *,
        dy: float = 600,
        dx: float = 0,
        x: float | None = None,
        y: float | None = None,
    ) -> JSON:
        if not all(math.isfinite(v) for v in (dx, dy)):
            raise ValueError("Scroll deltas must be finite")
        async with self._action("scroll") as receipt:
            metrics = await self.cdp.send("Page.getLayoutMetrics")
            viewport = metrics.get("cssLayoutViewport", metrics["layoutViewport"])
            x = viewport["clientWidth"] / 2 if x is None else x
            y = viewport["clientHeight"] / 2 if y is None else y
            if not all(math.isfinite(v) for v in (x, y)):
                raise ValueError("Scroll coordinates must be finite")
            receipt["may_have_executed"] = True
            await self.cdp.send(
                "Input.dispatchMouseEvent",
                {
                    "type": "mouseWheel",
                    "x": x,
                    "y": y,
                    "deltaX": dx,
                    "deltaY": dy,
                },
            )
        return copy.deepcopy(receipt)

    async def screenshot(
        self,
        path: str | Path | None = None,
        *,
        full_page: bool = False,
        max_pixels: int = 16_000_000,
    ) -> Path:
        self._check()
        if max_pixels < 1:
            raise ValueError("max_pixels must be positive")
        metrics = await self.cdp.send("Page.getLayoutMetrics")
        viewport = metrics.get("cssLayoutViewport", metrics["layoutViewport"])
        params: JSON = {
            "format": "png",
            "fromSurface": True,
            "captureBeyondViewport": full_page,
        }
        if full_page:
            size = metrics.get("cssContentSize", metrics["contentSize"])
            width, height = size["width"], size["height"]
            params["clip"] = {
                "x": size["x"],
                "y": size["y"],
                "width": width,
                "height": height,
                "scale": 1,
            }
        else:
            width, height = viewport["clientWidth"], viewport["clientHeight"]
        ratio = await self.evaluate("devicePixelRatio")
        if width * height * ratio * ratio > max_pixels:
            raise UnsupportedOperation(
                "Screenshot exceeds pixel budget; use a viewport capture or raise max_pixels."
            )
        data = (await self.cdp.send("Page.captureScreenshot", params))["data"]
        destination = (
            Path(path)
            if path is not None
            else Path(".browser-artifacts") / (uuid.uuid4().hex + ".png")
        )
        destination = destination.expanduser().resolve()
        if destination.suffix.lower() != ".png":
            raise ValueError("Screenshot path must end in .png")
        raw = base64.b64decode(data, validate=True)
        if hooks.attach_image is not None:
            try:
                hooks.attach_image(raw)
            except Exception:
                pass

        def write() -> None:
            destination.parent.mkdir(parents=True, exist_ok=True)
            destination.write_bytes(raw)

        await asyncio.to_thread(write)
        return destination

    async def close(self, *, allow_existing: bool = False) -> None:
        self._check()
        if not self.owned and not allow_existing:
            raise UnsupportedOperation(
                "This tab was not created by this connection; pass allow_existing=True to explicitly close it."
            )
        await self.browser.cdp.send("Target.closeTarget", {"targetId": self.target_id})
        self._invalidate(destroyed=True)
        await self._stop_tasks()
        self.browser._pages.pop(self.target_id, None)

    def __reduce_ex__(self, protocol: SupportsIndex, /) -> Any:
        raise TypeError(
            "Live pages cannot be saved; save page.descriptor() and ordinary observations."
        )