semantic_read_data.py

from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import re
import time
import uuid
from dataclasses import dataclass
from typing import Any, Callable
from urllib.parse import urlparse

from browser_core import (
    BrowserBridgeError,
    canonical_https_origin,
    inventory_location,
    redact_sensitive_query_values,
)


VERSION = "0.2.0-field-safe-blocks-r5"
CURSOR_TTL_SECONDS = 60.0
MAX_CURSOR_RECORDS = 256
MIN_OUTPUT_CHARS = 4_000
MAX_OUTPUT_CHARS = 60_000
DEFAULT_OUTPUT_CHARS = 36_000
MAX_TEXT_CHUNK = 2_000
MAX_CELL_CHARS = 512
MAX_RECORD_CHARS = 7_000
MAX_TABLES = 25
MAX_ROWS_PER_TABLE = 1_000
MAX_COLUMNS = 60
MAX_LISTS = 40
MAX_LIST_ITEMS = 500
MAX_LINKS = 750
MAX_JSON_LD = 40
_CURSOR_RE = re.compile(r"^cur_[a-f0-9]{32}$")
_PAGE_RE = re.compile(r"^pg_[a-f0-9]{24}$")
_FRAME_RE = re.compile(r"^fr_[a-f0-9]{24}$")
_AUTH_LINE_RE = re.compile(
    r"password|mật\s*khẩu|passcode|otp|one.?time|verification.?code|auth.?code|"
    r"access.?token|refresh.?token|secret|credential",
    re.IGNORECASE,
)
_NEXT_RE = re.compile(r"^(next|next page|older|more|tiếp|trang sau|sau)$", re.IGNORECASE)


@dataclass
class CursorRecord:
    cursor_id: str
    owner: str
    session_id: str
    lease_digest: str
    snapshot_id: str
    page_id: str
    frame_id: str
    mode: str
    max_output_chars: int
    records: tuple[dict[str, Any], ...]
    offset: int
    metadata: dict[str, Any]
    source_fingerprint: str
    created: float
    expires: float
    state: str = "ACTIVE"
    error_code: str = ""


class SemanticCursorStore:
    """Server-held one-use cursors over one frozen semantic snapshot."""

    def __init__(
        self,
        *,
        ttl_seconds: float = CURSOR_TTL_SECONDS,
        max_records: int = MAX_CURSOR_RECORDS,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        self.ttl_seconds = float(ttl_seconds)
        self.max_records = int(max_records)
        self.clock = clock
        self.records: dict[str, CursorRecord] = {}
        self.lock = asyncio.Lock()

    def _purge(self, now: float) -> None:
        for key, row in list(self.records.items()):
            if now >= row.expires + 300:
                self.records.pop(key, None)
        if len(self.records) < self.max_records:
            return
        retired = sorted(
            (r for r in self.records.values() if r.state != "ACTIVE" or now >= r.expires),
            key=lambda r: (r.expires, r.created, r.cursor_id),
        )
        for row in retired:
            self.records.pop(row.cursor_id, None)
            if len(self.records) < self.max_records:
                return

    @staticmethod
    def _lease_digest(lease_id: str) -> str:
        return hashlib.sha256(str(lease_id or "").encode("utf-8")).hexdigest()

    @staticmethod
    def _measure(value: Any) -> int:
        return len(json.dumps(value, ensure_ascii=False, separators=(",", ":")))

    @classmethod
    def _slice(cls, records: tuple[dict[str, Any], ...], start: int, max_chars: int) -> tuple[list[dict[str, Any]], int]:
        selected: list[dict[str, Any]] = []
        used = 2
        index = start
        while index < len(records):
            item = records[index]
            size = cls._measure(item) + (1 if selected else 0)
            if selected and used + size > max_chars:
                break
            if not selected and size > max_chars:
                # Records are normalized below to stay under MAX_RECORD_CHARS. This is a final fail-closed guard.
                raise BrowserBridgeError(
                    "READ_DATA_RECORD_TOO_LARGE",
                    "One normalized semantic record exceeds the requested output bound.",
                )
            selected.append(item)
            used += size
            index += 1
        return selected, index

    async def start(
        self,
        *,
        owner: str,
        session_id: str,
        lease_id: str,
        page_id: str,
        frame_id: str,
        mode: str,
        max_output_chars: int,
        records: list[dict[str, Any]],
        metadata: dict[str, Any],
        source_fingerprint: str,
    ) -> dict[str, Any]:
        frozen = tuple(records)
        items, next_offset = self._slice(frozen, 0, max_output_chars)
        snapshot_id = "snap_" + uuid.uuid4().hex
        next_cursor = None
        now = self.clock()
        if next_offset < len(frozen):
            async with self.lock:
                self._purge(now)
                if len(self.records) >= self.max_records:
                    raise BrowserBridgeError("READ_DATA_CURSOR_CAPACITY", "No semantic-read cursor capacity is available.")
                next_cursor = "cur_" + uuid.uuid4().hex
                self.records[next_cursor] = CursorRecord(
                    cursor_id=next_cursor,
                    owner=owner.casefold(),
                    session_id=session_id,
                    lease_digest=self._lease_digest(lease_id),
                    snapshot_id=snapshot_id,
                    page_id=page_id,
                    frame_id=frame_id,
                    mode=mode,
                    max_output_chars=max_output_chars,
                    records=frozen,
                    offset=next_offset,
                    metadata=dict(metadata),
                    source_fingerprint=source_fingerprint,
                    created=now,
                    expires=now + self.ttl_seconds,
                )
        return {
            "snapshot_id": snapshot_id,
            "items": items,
            "returned_count": len(items),
            "total_record_count": len(frozen),
            "page_index": 1,
            "next_cursor": next_cursor,
            "cursor_expires_in_ms": round(self.ttl_seconds * 1000) if next_cursor else None,
            "source_fingerprint": source_fingerprint,
            "metadata": metadata,
        }

    async def continue_page(
        self,
        *,
        cursor_id: str,
        owner: str,
        session_id: str,
        lease_id: str,
        max_output_chars: int,
    ) -> dict[str, Any]:
        if not _CURSOR_RE.fullmatch(str(cursor_id or "")):
            raise BrowserBridgeError("READ_DATA_CURSOR_INVALID", "The semantic-read cursor is invalid.")
        now = self.clock()
        async with self.lock:
            self._purge(now)
            row = self.records.get(cursor_id)
            if row is None:
                raise BrowserBridgeError("READ_DATA_CURSOR_NOT_FOUND", "The semantic-read cursor is unavailable.")
            if row.state != "ACTIVE":
                raise BrowserBridgeError(row.error_code or "READ_DATA_CURSOR_ALREADY_USED", "The semantic-read cursor cannot be replayed.")
            if now >= row.expires:
                row.state = "EXPIRED"
                row.error_code = "READ_DATA_CURSOR_EXPIRED"
                raise BrowserBridgeError(row.error_code, "The semantic-read cursor expired.")
            if not hmac.compare_digest(row.owner, owner.casefold()):
                raise BrowserBridgeError("READ_DATA_CURSOR_OWNER_MISMATCH", "The semantic-read cursor owner does not match.")
            if not hmac.compare_digest(row.session_id, session_id):
                raise BrowserBridgeError("READ_DATA_CURSOR_SESSION_MISMATCH", "The semantic-read cursor session does not match.")
            if not hmac.compare_digest(row.lease_digest, self._lease_digest(lease_id)):
                raise BrowserBridgeError("READ_DATA_CURSOR_LEASE_MISMATCH", "The semantic-read cursor lease does not match.")
            if row.max_output_chars != max_output_chars:
                raise BrowserBridgeError("READ_DATA_CURSOR_BOUND_MISMATCH", "Keep max_output_chars unchanged while continuing a cursor.")
            items, next_offset = self._slice(row.records, row.offset, max_output_chars)
            row.state = "USED"
            row.error_code = "READ_DATA_CURSOR_ALREADY_USED"
            next_cursor = None
            if next_offset < len(row.records):
                if len(self.records) >= self.max_records:
                    self._purge(now)
                if len(self.records) >= self.max_records:
                    raise BrowserBridgeError("READ_DATA_CURSOR_CAPACITY", "No semantic-read cursor capacity is available.")
                next_cursor = "cur_" + uuid.uuid4().hex
                self.records[next_cursor] = CursorRecord(
                    cursor_id=next_cursor,
                    owner=row.owner,
                    session_id=row.session_id,
                    lease_digest=row.lease_digest,
                    snapshot_id=row.snapshot_id,
                    page_id=row.page_id,
                    frame_id=row.frame_id,
                    mode=row.mode,
                    max_output_chars=row.max_output_chars,
                    records=row.records,
                    offset=next_offset,
                    metadata=dict(row.metadata),
                    source_fingerprint=row.source_fingerprint,
                    created=row.created,
                    expires=row.expires,
                )
            return {
                "snapshot_id": row.snapshot_id,
                "items": items,
                "returned_count": len(items),
                "total_record_count": len(row.records),
                "page_index": (row.offset // max(1, len(items))) + 1 if items else 1,
                "next_cursor": next_cursor,
                "cursor_expires_in_ms": max(0, round((row.expires - now) * 1000)) if next_cursor else None,
                "source_fingerprint": row.source_fingerprint,
                "metadata": dict(row.metadata),
            }


class SemanticReadData:
    """Lease-bound, read-only semantic extraction for one explicit page/frame.

    This does not scroll, click pagination controls, change the selected page, take screenshots,
    or return attachments. It reports current-DOM completeness separately from logical-dataset
    completeness so virtualized/paginated data is never silently promoted to complete.
    """

    def __init__(self, manager: Any, *, clock: Callable[[], float] = time.monotonic) -> None:
        self.manager = manager
        self.clock = clock
        self.cursors = SemanticCursorStore(clock=clock)
        self._page_ids: dict[str, dict[int, str]] = {}
        self._frame_ids: dict[str, dict[int, str]] = {}

    @staticmethod
    def _new_id(prefix: str) -> str:
        return prefix + hashlib.sha256(uuid.uuid4().bytes).hexdigest()[:24]

    def _stable_page_id(self, session_id: str, page: Any) -> str:
        mapping = self._page_ids.setdefault(session_id, {})
        key = id(page)
        if key not in mapping:
            mapping[key] = self._new_id("pg_")
        return mapping[key]

    def _stable_frame_id(self, session_id: str, frame: Any) -> str:
        mapping = self._frame_ids.setdefault(session_id, {})
        key = id(frame)
        if key not in mapping:
            mapping[key] = self._new_id("fr_")
        return mapping[key]

    def _page_inventory(self, session: Any) -> tuple[list[dict[str, Any]], dict[str, Any], int]:
        live = []
        by_id = {}
        blocked_count = 0
        allowed = set(getattr(getattr(session, "lease", None), "allowed_origins", ()) or ())
        for page in list(session.context.pages):
            try:
                if page.is_closed():
                    continue
                origin, path = inventory_location(page.url)
                if origin not in allowed:
                    blocked_count += 1
                    continue
                page_id = self._stable_page_id(session.session_id, page)
                row = {
                    "page_id": page_id,
                    "current_origin": origin,
                    "current_path": path,
                    "query_fragment_title_text_and_fields_omitted": True,
                }
                live.append(row)
                by_id[page_id] = page
            except Exception:
                blocked_count += 1
                continue
        live.sort(key=lambda row: row["page_id"])
        return live, by_id, blocked_count

    def _frame_inventory(self, session: Any, page: Any) -> tuple[list[dict[str, Any]], dict[str, Any], int]:
        rows = []
        by_id = {}
        blocked_count = 0
        allowed = set(getattr(getattr(session, "lease", None), "allowed_origins", ()) or ())
        frames = list(page.frames)
        for frame in frames:
            try:
                effective_url = str(frame.url or "")
                if effective_url == "about:blank" and frame.parent_frame is not None:
                    effective_url = str(frame.parent_frame.url or "")
                origin, path = inventory_location(effective_url)
                if origin not in allowed:
                    blocked_count += 1
                    continue
                frame_id = self._stable_frame_id(session.session_id, frame)
                parent = frame.parent_frame
                parent_id = None
                if parent is not None:
                    parent_url = str(parent.url or "")
                    if parent_url == "about:blank" and parent.parent_frame is not None:
                        parent_url = str(parent.parent_frame.url or "")
                    parent_origin, _parent_path = inventory_location(parent_url)
                    if parent_origin in allowed:
                        parent_id = self._stable_frame_id(session.session_id, parent)
                row = {
                    "frame_id": frame_id,
                    "parent_frame_id": parent_id,
                    "main_frame": frame is page.main_frame,
                    "current_origin": origin,
                    "current_path": path,
                    "query_fragment_text_and_fields_omitted": True,
                }
                rows.append(row)
                by_id[frame_id] = frame
            except Exception:
                blocked_count += 1
                continue
        rows.sort(key=lambda row: (not row["main_frame"], row["frame_id"]))
        return rows, by_id, blocked_count

    @staticmethod
    def _require_frame_origin(session: Any, frame: Any) -> None:
        lease = getattr(session, "lease", None)
        if lease is None:
            raise BrowserBridgeError("READ_DATA_LEASE_REQUIRED", "Semantic read_data is available only on an owned leased session.")
        raw = str(frame.url or "")
        if raw == "about:blank" and frame.parent_frame is not None:
            raw = str(frame.parent_frame.url or "")
        try:
            origin = canonical_https_origin(raw, origin_only=False)
        except Exception as exc:
            raise BrowserBridgeError("READ_DATA_FRAME_ORIGIN_UNAVAILABLE", "The selected frame origin cannot be verified.") from exc
        if origin not in lease.allowed_origins:
            raise BrowserBridgeError("READ_DATA_FRAME_ORIGIN_BLOCKED", "The selected frame is outside the task lease origin set.")

    @staticmethod
    def _sanitize_string(value: str) -> tuple[str, int]:
        text = str(value or "")
        omitted = 0
        lines = []
        for line in text.splitlines() or [text]:
            if _AUTH_LINE_RE.search(line):
                lines.append("[AUTH_FIELD_OMITTED]")
                omitted += 1
            else:
                lines.append(line)
        return "\n".join(lines), omitted

    @classmethod
    def _sanitize_value(cls, value: Any) -> tuple[Any, int]:
        if isinstance(value, str):
            return cls._sanitize_string(value)
        if isinstance(value, list):
            out = []
            total = 0
            for item in value:
                cleaned, count = cls._sanitize_value(item)
                out.append(cleaned)
                total += count
            return out, total
        if isinstance(value, dict):
            out: dict[str, Any] = {}
            total = 0
            for key, item in value.items():
                if _AUTH_LINE_RE.search(str(key)):
                    out[str(key)] = "[AUTH_FIELD_OMITTED]"
                    total += 1
                    continue
                cleaned, count = cls._sanitize_value(item)
                out[str(key)] = cleaned
                total += count
            return out, total
        return value, 0

    @staticmethod
    def _bounded_record(record: dict[str, Any]) -> dict[str, Any]:
        candidate = dict(record)
        raw = json.dumps(candidate, ensure_ascii=False, separators=(",", ":"))
        if len(raw) <= MAX_RECORD_CHARS:
            return candidate
        if candidate.get("kind") == "table_row":
            cells = list(candidate.get("cells") or [])
            while cells and len(json.dumps({**candidate, "cells": cells}, ensure_ascii=False, separators=(",", ":"))) > MAX_RECORD_CHARS:
                cells.pop()
            candidate["cells"] = cells
            candidate["record_truncated"] = True
            candidate["incomplete_reason"] = "row_output_bound"
            return candidate
        candidate = {
            "kind": str(candidate.get("kind") or "record"),
            "record_truncated": True,
            "incomplete_reason": "record_output_bound",
            "sha256": hashlib.sha256(raw.encode("utf-8")).hexdigest(),
        }
        return candidate

    @staticmethod
    def _records(extracted: dict[str, Any]) -> list[dict[str, Any]]:
        records: list[dict[str, Any]] = []
        text = str(extracted.get("page_text") or "")
        for index in range(0, len(text), MAX_TEXT_CHUNK):
            records.append({"kind": "page_text", "chunk_index": (index // MAX_TEXT_CHUNK) + 1, "text": text[index:index + MAX_TEXT_CHUNK]})
        for table in list(extracted.get("tables") or []):
            table_index = int(table.get("table_index") or 0)
            headers = [str(x)[:MAX_CELL_CHARS] for x in list(table.get("headers") or [])[:MAX_COLUMNS]]
            records.append({
                "kind": "table_header",
                "table_index": table_index,
                "table_kind": str(table.get("table_kind") or ""),
                "caption": str(table.get("caption") or "")[:1000],
                "headers": headers,
                "observed_row_count": int(table.get("row_count") or 0),
                "declared_row_count": table.get("declared_row_count"),
                "column_count": int(table.get("column_count") or len(headers)),
            })
            for row_index, row in enumerate(list(table.get("rows") or []), start=1):
                cells = [str(x)[:MAX_CELL_CHARS] for x in list(row)[:MAX_COLUMNS]]
                if any(_AUTH_LINE_RE.search(cell) for cell in cells):
                    records.append({
                        "kind": "auth_omitted",
                        "source_kind": "table_row",
                        "table_index": table_index,
                        "row_index": row_index,
                        "value": "[AUTH_FIELD_OMITTED]",
                    })
                else:
                    records.append({"kind": "table_row", "table_index": table_index, "row_index": row_index, "cells": cells})
        for row in list(extracted.get("lists") or []):
            list_index = int(row.get("list_index") or 0)
            for item_index, item in enumerate(list(row.get("items") or []), start=1):
                records.append({"kind": "list_item", "list_index": list_index, "item_index": item_index, "text": str(item)[:MAX_TEXT_CHUNK]})
        for index, link in enumerate(list(extracted.get("links") or []), start=1):
            records.append({
                "kind": "link",
                "index": index,
                "text": str(link.get("text") or "")[:1000],
                "url": redact_sensitive_query_values(str(link.get("url") or ""))[:3000],
            })
        for index, value in enumerate(list(extracted.get("json_ld") or []), start=1):
            records.append({"kind": "json_ld", "index": index, "value": value})
        return [SemanticReadData._bounded_record(item) for item in records]

    async def _extract_frame(self, frame: Any, mode: str) -> dict[str, Any]:
        script = r"""
        ({ mode, maxTables, maxRows, maxColumns, maxLists, maxListItems, maxLinks, maxJsonLd }) => {
          const clean = (v, n = 4000) => { const text = String(v || '').replace(/\s+/g, ' ').trim(); if (text.length > n) result.coverage.text_limit_hit = true; return text.slice(0, n); };
          const result = {
            page_text: '', tables: [], lists: [], links: [], json_ld: [],
            coverage: {
              text_limit_hit: false, table_limit_hit: false, row_limit_hit: false, column_limit_hit: false,
              list_limit_hit: false, link_limit_hit: false, json_ld_limit_hit: false,
              virtualized_incomplete: false, virtualized_evidence: [],
              pagination_incomplete: false, pagination_evidence: [],
              scan_limit_hit: false, excluded_field_count: 0, excluded_hidden_count: 0,
              open_shadow_roots_visited: 0, scanned_node_count: 0
            }
          };
          // Read the composed DOM without cloning/mutating it or touching field values.
          const omittedFields = new WeakSet();
          const omittedHidden = new WeakSet();
          const shadowRoots = new WeakSet();
          let visitedNodes = 0;
          const startedAt = performance.now();
          const excludedTags = new Set(['INPUT','TEXTAREA','SELECT','OPTION','OUTPUT','SCRIPT','STYLE','TEMPLATE','NOSCRIPT','IFRAME','OBJECT','EMBED']);
          const fieldTags = new Set(['INPUT','TEXTAREA','SELECT','OPTION','OUTPUT']);
          const fieldRoles = new Set(['textbox','searchbox','combobox','spinbutton','slider']);
          const parentOf = el => el?.assignedSlot || el?.parentElement || el?.getRootNode?.()?.host || null;
          const fieldElement = el => fieldTags.has(el.tagName) || el.isContentEditable || fieldRoles.has((el.getAttribute('role') || '').toLowerCase());
          const permitted = el => {
            let cursor = el;
            let depth = 0;
            while (cursor && depth++ < 96) {
              if (fieldElement(cursor)) {
                if (!omittedFields.has(cursor)) { omittedFields.add(cursor); result.coverage.excluded_field_count++; }
                return false;
              }
              if (excludedTags.has(cursor.tagName)) return false;
              // Range rectangles may exist for descendants of closed <details>.
              // Only its first summary participates in the rendered tree.
              if (cursor.tagName === 'DETAILS' && !cursor.open && cursor !== el) {
                const summary = Array.from(cursor.children).find(child => child.tagName === 'SUMMARY');
                if (!summary || !(summary === el || summary.contains(el))) {
                  if (!omittedHidden.has(cursor)) { omittedHidden.add(cursor); result.coverage.excluded_hidden_count++; }
                  return false;
                }
              }
              const s = getComputedStyle(cursor);
              if (cursor.hidden || cursor.getAttribute('aria-hidden') === 'true' || cursor.hasAttribute('inert') ||
                  s.display === 'none' || s.visibility === 'hidden' || s.visibility === 'collapse' || Number(s.opacity) === 0) {
                if (!omittedHidden.has(cursor)) { omittedHidden.add(cursor); result.coverage.excluded_hidden_count++; }
                return false;
              }
              cursor = parentOf(cursor);
            }
            if (cursor) { result.coverage.scan_limit_hit = true; return false; }
            return true;
          };
          const safeText = (root, limit = 4000, keepLines = false) => {
            if (!root || root.nodeType !== Node.ELEMENT_NODE || !permitted(root)) return '';
            const pieces = [];
            const seen = new WeakSet();
            let chars = 0;
            const push = text => {
              if (!text) return;
              if (chars + text.length > limit) result.coverage.text_limit_hit = true;
              const part = text.slice(0, Math.max(0, limit - chars));
              pieces.push(part); chars += part.length;
            };
            const walk = (node, depth) => {
              if (seen.has(node)) return;
              seen.add(node);
              visitedNodes++;
              if (visitedNodes > 30000 || depth > 96 || performance.now() - startedAt > 1800) {
                result.coverage.scan_limit_hit = true;
                return;
              }
              if (chars >= limit) { result.coverage.text_limit_hit = true; return; }
              if (node.nodeType === Node.TEXT_NODE) {
                const parent = node.parentElement || node.getRootNode()?.host;
                if (!parent || !permitted(parent)) return;
                const range = document.createRange();
                range.selectNodeContents(node);
                const rendered = Array.from(range.getClientRects()).some(r => r.width > 0 && r.height > 0);
                if (rendered) push(String(node.nodeValue || '').replace(/\s+/g, ' '));
                return;
              }
              if (node.nodeType !== Node.ELEMENT_NODE) return;
              if (fieldElement(node)) {
                if (!omittedFields.has(node)) { omittedFields.add(node); result.coverage.excluded_field_count++; }
                // Internal boundary only. Removed after the unchanged auth filter.
                // It prevents an empty password field consuming the next paragraph.
                if (keepLines) push(' [FORM_VALUE_OMITTED] ');
                return;
              }
              if (!permitted(node)) return;
              const display = getComputedStyle(node).display;
              const block = node.tagName === 'BR' || node.tagName === 'LABEL' ||
                /^(block|flex|grid|list-item|flow-root|table|table-row|table-cell|table-caption)$/.test(display);
              if (block) push('\n');
              let children;
              if (node.tagName === 'DETAILS' && !node.open) {
                const summary = Array.from(node.children).find(child => child.tagName === 'SUMMARY');
                children = summary ? [summary] : [];
                if (node.childNodes.length > children.length && !omittedHidden.has(node)) {
                  omittedHidden.add(node); result.coverage.excluded_hidden_count++;
                }
              } else if (node.shadowRoot) {
                if (!shadowRoots.has(node.shadowRoot)) { shadowRoots.add(node.shadowRoot); result.coverage.open_shadow_roots_visited++; }
                children = node.shadowRoot.childNodes;
              } else if (node.tagName === 'SLOT' && node.assignedNodes().length) {
                children = node.assignedNodes({flatten: true});
              } else {
                children = node.childNodes;
              }
              for (const child of children) {
                if (chars >= limit || result.coverage.scan_limit_hit) break;
                walk(child, depth + 1);
              }
              if (block) push('\n');
            };
            walk(root, 0);
            result.coverage.scanned_node_count = visitedNodes;
            const lines = pieces.join('').split(/\n/).map(s => s.replace(/\s+/g, ' ').trim()).filter(Boolean);
            return keepLines ? lines.join('\n') : lines.join(' ');
          };
          if (mode === 'all' || mode === 'text') {
            result.page_text = safeText(document.body, 200000, true);
          }
          if (mode === 'all' || mode === 'tables') {
            const tableNodes = Array.from(document.querySelectorAll('table')).filter(permitted);
            if (tableNodes.length > maxTables) result.coverage.table_limit_hit = true;
            for (const [ti, table] of tableNodes.slice(0, maxTables).entries()) {
              const allRows = Array.from(table.rows || []).filter(permitted);
              if (allRows.length > maxRows) result.coverage.row_limit_hit = true;
              const domRows = allRows.slice(0, maxRows);
              if (!domRows.length) continue;
              const firstCellsAll = Array.from(domRows[0].cells || []);
              if (firstCellsAll.length > maxColumns) result.coverage.column_limit_hit = true;
              const firstCells = firstCellsAll.slice(0, maxColumns);
              const header = firstCells.some(c => c.tagName.toLowerCase() === 'th');
              const headers = firstCells.map((c, i) => safeText(c, 512) || `column_${i+1}`);
              const rows = (header ? domRows.slice(1) : domRows).map(row => {
                const allCells = Array.from(row.cells || []);
                if (allCells.length > maxColumns) result.coverage.column_limit_hit = true;
                return allCells.slice(0, maxColumns).map(c => safeText(c, 512));
              }).filter(row => row.some(Boolean));
              const declared = (table.hasAttribute('aria-rowcount') ? Number(table.getAttribute('aria-rowcount')) : NaN);
              if (Number.isFinite(declared) && declared > rows.length + (header ? 1 : 0)) {
                result.coverage.virtualized_incomplete = true;
                result.coverage.virtualized_evidence.push({table_index: ti + 1, declared_row_count: declared, observed_dom_rows: rows.length});
              }
              result.tables.push({table_index: ti + 1, table_kind: 'html_table', caption: safeText(table.caption, 500), headers, rows, row_count: rows.length, declared_row_count: Number.isFinite(declared) ? declared : null, column_count: Math.max(headers.length, ...rows.map(r => r.length), 0)});
            }
            const grids = Array.from(document.querySelectorAll('[role="grid"],[role="treegrid"]')).filter(g => g.tagName.toLowerCase() !== 'table' && permitted(g));
            if (result.tables.length + grids.length > maxTables) result.coverage.table_limit_hit = true;
            for (const grid of grids.slice(0, Math.max(0, maxTables - result.tables.length))) {
              const headerNodes = Array.from(grid.querySelectorAll('[role="columnheader"]')).slice(0, maxColumns);
              const headers = headerNodes.map((c, i) => safeText(c, 512) || `column_${i+1}`);
              const allRows = Array.from(grid.querySelectorAll('[role="row"]')).filter(permitted);
              if (allRows.length > maxRows) result.coverage.row_limit_hit = true;
              const rows = [];
              for (const row of allRows.slice(0, maxRows)) {
                const cellsAll = Array.from(row.querySelectorAll('[role="gridcell"],[role="cell"]'));
                if (!cellsAll.length) continue;
                if (cellsAll.length > maxColumns) result.coverage.column_limit_hit = true;
                rows.push(cellsAll.slice(0, maxColumns).map(c => safeText(c, 512)));
              }
              const declared = (grid.hasAttribute('aria-rowcount') ? Number(grid.getAttribute('aria-rowcount')) : NaN);
              if (Number.isFinite(declared) && declared > rows.length) {
                result.coverage.virtualized_incomplete = true;
                result.coverage.virtualized_evidence.push({table_index: result.tables.length + 1, declared_row_count: declared, observed_dom_rows: rows.length});
              }
              result.tables.push({table_index: result.tables.length + 1, table_kind: 'aria_grid', caption: clean(grid.getAttribute('aria-label') || '', 500), headers, rows, row_count: rows.length, declared_row_count: Number.isFinite(declared) ? declared : null, column_count: Math.max(headers.length, ...rows.map(r => r.length), 0)});
            }
          }
          if (mode === 'all' || mode === 'lists') {
            const roots = Array.from(document.querySelectorAll('ul,ol')).filter(x => permitted(x) && (!x.parentElement || !x.parentElement.closest('ul,ol')));
            if (roots.length > maxLists) result.coverage.list_limit_hit = true;
            for (const [li, list] of roots.slice(0, maxLists).entries()) {
              const allItems = Array.from(list.children || []).filter(x => x.tagName?.toLowerCase() === 'li' && permitted(x));
              if (allItems.length > maxListItems) result.coverage.list_limit_hit = true;
              const items = allItems.slice(0, maxListItems).map(x => safeText(x, 4000)).filter(Boolean);
              if (items.length) result.lists.push({list_index: li + 1, ordered: list.tagName.toLowerCase() === 'ol', items});
            }
          }
          if (mode === 'all' || mode === 'links') {
            const allLinks = Array.from(document.querySelectorAll('a[href]')).filter(a => {
              const s = getComputedStyle(a); const r = a.getBoundingClientRect();
              return permitted(a) && s.visibility !== 'hidden' && s.display !== 'none' && r.width > 0 && r.height > 0;
            });
            if (allLinks.length > maxLinks) result.coverage.link_limit_hit = true;
            result.links = allLinks.slice(0, maxLinks).map(a => ({text: safeText(a, 1000) || clean(a.getAttribute('aria-label'), 1000), url: clean(a.href, 3000)}));
          }
          if (mode === 'all' || mode === 'json_ld') {
            const scripts = Array.from(document.querySelectorAll('script[type="application/ld+json"]'));
            if (scripts.length > maxJsonLd) result.coverage.json_ld_limit_hit = true;
            for (const node of scripts.slice(0, maxJsonLd)) {
              const fullRaw = String(node.textContent || '').trim(); if (fullRaw.length > 200000) result.coverage.text_limit_hit = true; const raw = fullRaw.slice(0, 200000);
              if (!raw) continue;
              try { result.json_ld.push(JSON.parse(raw)); } catch { result.json_ld.push({_parse_error: true, raw}); }
            }
          }
          const nextSignals = Array.from(document.querySelectorAll('a[rel="next"],button,a')).filter(el => {
            const label = clean(el.getAttribute('aria-label') || el.getAttribute('title') || safeText(el, 120), 120);
            return el.matches('a[rel="next"]') || /^(next|next page|older|more|tiếp|trang sau|sau)$/i.test(label);
          }).slice(0, 8).map(el => clean(el.getAttribute('aria-label') || el.getAttribute('title') || safeText(el, 120), 120));
          if (nextSignals.length) {
            result.coverage.pagination_incomplete = true;
            result.coverage.pagination_evidence = nextSignals;
          }
          return result;
        }
        """
        try:
            return await frame.evaluate(
                script,
                {
                    "mode": mode,
                    "maxTables": MAX_TABLES,
                    "maxRows": MAX_ROWS_PER_TABLE,
                    "maxColumns": MAX_COLUMNS,
                    "maxLists": MAX_LISTS,
                    "maxListItems": MAX_LIST_ITEMS,
                    "maxLinks": MAX_LINKS,
                    "maxJsonLd": MAX_JSON_LD,
                },
            )
        except Exception as exc:
            raise BrowserBridgeError(
                "READ_DATA_EXTRACTION_FAILED",
                "The selected page/frame could not be read semantically.",
                details={"error_type": type(exc).__name__},
            ) from exc

    @staticmethod
    def _coverage(extracted: dict[str, Any], records: list[dict[str, Any]]) -> dict[str, Any]:
        raw = dict(extracted.get("coverage") or {})
        reasons = []
        for key in ("text_limit_hit", "table_limit_hit", "row_limit_hit", "column_limit_hit", "list_limit_hit", "link_limit_hit", "json_ld_limit_hit", "scan_limit_hit"):
            if raw.get(key):
                reasons.append(key)
        if raw.get("virtualized_incomplete"):
            reasons.append("virtualized_rows_not_fully_present_in_current_dom")
        if raw.get("pagination_incomplete"):
            reasons.append("next_page_signal_present_but_read_data_does_not_navigate")
        if any(item.get("record_truncated") for item in records):
            reasons.append("normalized_record_output_bound")
        return {
            "scope": "selected_page_selected_frame_current_dom_only",
            "reader_revision": VERSION,
            "field_values_included": False,
            "excluded_field_count": int(raw.get("excluded_field_count", 0)),
            "excluded_hidden_count": int(raw.get("excluded_hidden_count", 0)),
            "scanned_node_count": int(raw.get("scanned_node_count", 0)),
            "open_shadow_roots_visited": int(raw.get("open_shadow_roots_visited", 0)),
            "shadow_scope": "text_open_composed_roots_structured_modes_light_dom_closed_unverified",
            "current_dom_complete_within_bounds": not any(
                raw.get(key) for key in ("text_limit_hit", "table_limit_hit", "row_limit_hit", "column_limit_hit", "list_limit_hit", "link_limit_hit", "json_ld_limit_hit", "scan_limit_hit")
            ) and not any(item.get("record_truncated") for item in records),
            "logical_dataset_complete": False if raw.get("virtualized_incomplete") or raw.get("pagination_incomplete") else None,
            "virtualized_data_complete": False if raw.get("virtualized_incomplete") else None,
            "pagination_complete": False if raw.get("pagination_incomplete") else None,
            "incomplete_reasons": list(dict.fromkeys(reasons)),
            "virtualized_evidence": list(raw.get("virtualized_evidence") or [])[:20],
            "pagination_evidence": list(raw.get("pagination_evidence") or [])[:8],
            "absence_of_incomplete_signal_is_not_proof_of_full_logical_dataset": True,
        }

    async def read_data(
        self,
        owner: str,
        *,
        session_id: str,
        lease_id: str,
        page_id: str = "",
        frame_id: str = "",
        mode: str = "tables",
        cursor: str = "",
        max_output_chars: int = DEFAULT_OUTPUT_CHARS,
    ) -> dict[str, Any]:
        selected_mode = str(mode or "tables").casefold().strip()
        if selected_mode not in {"all", "tables", "lists", "links", "json_ld", "text"}:
            raise BrowserBridgeError("READ_DATA_MODE_INVALID", "mode must be all, tables, lists, links, json_ld, or text.")
        if type(max_output_chars) is not int or not MIN_OUTPUT_CHARS <= max_output_chars <= MAX_OUTPUT_CHARS:
            raise BrowserBridgeError("READ_DATA_OUTPUT_BOUND_INVALID", f"max_output_chars must be {MIN_OUTPUT_CHARS} to {MAX_OUTPUT_CHARS}.")
        if not str(lease_id or "").startswith("lease_"):
            raise BrowserBridgeError("READ_DATA_LEASE_REQUIRED", "An exact owned-session lease_id is required.")

        # manager.get() revalidates owner/session/lease/TTL/current origin because app.py wraps this
        # operation in manager.lease_scope(lease_id). Continuations therefore cannot bypass lease expiry.
        session = await self.manager.get(owner, session_id, touch=False)
        if getattr(session, "lease", None) is None:
            raise BrowserBridgeError("READ_DATA_LEASE_REQUIRED", "Semantic read_data requires an owned leased session.")

        if cursor:
            async with session.lock:
                await self.manager.revalidate_locked_session(owner, session_id, session)
                page = await self.cursors.continue_page(
                    cursor_id=cursor,
                    owner=owner,
                    session_id=session_id,
                    lease_id=lease_id,
                    max_output_chars=max_output_chars,
                )
            metadata = dict(page.pop("metadata"))
            return {
                "ok": True,
                "status": "READ_DATA_COMPLETE" if page.get("next_cursor") is None else "READ_DATA_PARTIAL",
                "session_id": session_id,
                "lease_required": True,
                "page_id": metadata["page_id"],
                "frame_id": metadata["frame_id"],
                "mode": metadata["mode"],
                "items": page["items"],
                "returned_count": page["returned_count"],
                "total_record_count": page["total_record_count"],
                "snapshot_id": page["snapshot_id"],
                "source_fingerprint": page["source_fingerprint"],
                "next_cursor": page["next_cursor"],
                "cursor_expires_in_ms": page["cursor_expires_in_ms"],
                "reader_revision": VERSION,
                "coverage": metadata["coverage"],
                "auth_field_omissions": metadata["auth_field_omissions"],
                "page_inventory": metadata["page_inventory"],
                "frame_inventory": metadata["frame_inventory"],
                "blocked_page_count": metadata["blocked_page_count"],
                "blocked_frame_count": metadata["blocked_frame_count"],
                "attachment_returned": False,
                "page_or_frame_changed": False,
                "content_origin": "untrusted_web_page",
                "secret_values_exposed": False,
            }

        async with session.lock:
            await self.manager.revalidate_locked_session(owner, session_id, session)
            pages, page_map, blocked_page_count = self._page_inventory(session)
            if not pages:
                raise BrowserBridgeError(
                    "READ_DATA_NO_LEASE_SCOPED_PAGES",
                    "The owned session has no readable page inside the task lease origin set.",
                    details={"blocked_page_count": blocked_page_count},
                )
            if page_id:
                if not _PAGE_RE.fullmatch(page_id) or page_id not in page_map:
                    raise BrowserBridgeError("READ_DATA_PAGE_NOT_FOUND", "The requested page_id is unavailable. Read the returned page inventory again.")
                page = page_map[page_id]
                selected_page_id = page_id
            elif len(pages) == 1:
                selected_page_id = pages[0]["page_id"]
                page = page_map[selected_page_id]
            else:
                return {
                    "ok": True,
                    "status": "READ_DATA_PAGE_SELECTION_REQUIRED",
                    "session_id": session_id,
                    "lease_required": True,
                    "page_inventory": pages,
                    "page_count": len(pages),
                    "blocked_page_count": blocked_page_count,
                    "complete": False,
                    "attachment_returned": False,
                    "secret_values_exposed": False,
                }
            frames, frame_map, blocked_frame_count = self._frame_inventory(session, page)
            if frame_id:
                if not _FRAME_RE.fullmatch(frame_id) or frame_id not in frame_map:
                    raise BrowserBridgeError("READ_DATA_FRAME_NOT_FOUND", "The requested frame_id is unavailable. Read the returned frame inventory again.")
                frame = frame_map[frame_id]
                selected_frame_id = frame_id
            else:
                frame = page.main_frame
                selected_frame_id = self._stable_frame_id(session_id, frame)
            self._require_frame_origin(session, frame)
            extracted = await self._extract_frame(frame, selected_mode)
            # Keep a standalone auth label with the following block before filtering.
            # This avoids exposing a value laid out on the next line after "Password:".
            lines = str(extracted.get("page_text", "")).splitlines()
            merged = []
            index = 0
            while index < len(lines):
                line = lines[index]
                if _AUTH_LINE_RE.search(line) and "[FORM_VALUE_OMITTED]" not in line and (line.endswith(":") or len(line.split()) <= 3) and index + 1 < len(lines):
                    line += " " + lines[index + 1]
                    index += 1
                merged.append(line)
                index += 1
            extracted["page_text"] = "\n".join(merged)
            sanitized, omitted = self._sanitize_value(extracted)
            if isinstance(sanitized, dict):
                sanitized["page_text"] = str(sanitized.get("page_text", "")).replace("[FORM_VALUE_OMITTED]", "")
            sanitized = sanitized if isinstance(sanitized, dict) else {}
            records = self._records(sanitized)
            coverage = self._coverage(sanitized, records)
            if omitted or coverage["excluded_field_count"]:
                coverage["returned_content_complete"] = False
                reasons = coverage["incomplete_reasons"]
                if omitted:
                    reasons.append("auth_content_redacted")
                if coverage["excluded_field_count"]:
                    reasons.append("form_value_subtrees_excluded")
            page_origin, page_path = inventory_location(page.url)
            frame_origin, frame_path = inventory_location(frame.url)
            fingerprint_payload = {
                "session_id": session_id,
                "page_id": selected_page_id,
                "frame_id": selected_frame_id,
                "mode": selected_mode,
                "records": records,
                "coverage": coverage,
            }
            fingerprint = hashlib.sha256(json.dumps(fingerprint_payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
            metadata = {
                "page_id": selected_page_id,
                "frame_id": selected_frame_id,
                "mode": selected_mode,
                "coverage": coverage,
                "auth_field_omissions": omitted,
                "page_inventory": pages,
                "frame_inventory": frames,
                "blocked_page_count": blocked_page_count,
                "blocked_frame_count": blocked_frame_count,
                "page_origin": page_origin,
                "page_path": page_path,
                "frame_origin": frame_origin,
                "frame_path": frame_path,
            }
            frozen = await self.cursors.start(
                owner=owner,
                session_id=session_id,
                lease_id=lease_id,
                page_id=selected_page_id,
                frame_id=selected_frame_id,
                mode=selected_mode,
                max_output_chars=max_output_chars,
                records=records,
                metadata=metadata,
                source_fingerprint=fingerprint,
            )

        metadata = dict(frozen.pop("metadata"))
        return {
            "ok": True,
            "status": "READ_DATA_COMPLETE" if frozen.get("next_cursor") is None else "READ_DATA_PARTIAL",
            "session_id": session_id,
            "lease_required": True,
            "page_id": selected_page_id,
            "frame_id": selected_frame_id,
            "page_origin": metadata["page_origin"],
            "page_path": metadata["page_path"],
            "frame_origin": metadata["frame_origin"],
            "frame_path": metadata["frame_path"],
            "mode": selected_mode,
            "items": frozen["items"],
            "returned_count": frozen["returned_count"],
            "total_record_count": frozen["total_record_count"],
            "snapshot_id": frozen["snapshot_id"],
            "source_fingerprint": frozen["source_fingerprint"],
            "next_cursor": frozen["next_cursor"],
            "cursor_expires_in_ms": frozen["cursor_expires_in_ms"],
            "reader_revision": VERSION,
            "coverage": coverage,
            "auth_field_omissions": omitted,
            "page_inventory": pages,
            "frame_inventory": frames,
            "blocked_page_count": blocked_page_count,
            "blocked_frame_count": blocked_frame_count,
            "attachment_returned": False,
            "page_or_frame_changed": False,
            "screenshot_taken": False,
            "os_input_used": False,
            "content_origin": "untrusted_web_page",
            "secret_values_exposed": False,
        }