arborist/arborist/read.py

"""arborist.read — read-only seam for dashboards and verifiers.

Counterpart to :mod:`arborist.embed`: that module is the seam embedders
use to *write* into arborist (Document → ingest → root). This module is
the seam read-only consumers (Merkle Command Center / arborist-viz,
third-party verifiers, archival mirrors) use to *read* from arborist.

Why a seam? Same reason as ``arborist.embed`` — so internal refactors
(schema bumps, compression scheme changes, sharding plans, wikitext
projection version, qa-vs-document shard separation) don't break the
downstream consumers. Consumers import from here; they never touch
arborist's tables, columns, or helper modules directly.

Surface summary
---------------

::

    open_shards(paths)                  → Shards
    shards.roots(limit=50)              → list[Root]
    shards.root(document_root)          → Root | None
    shards.leaves(document_root, *, reveal_private=False, project=True)
                                        → list[Leaf]
    shards.proof(document_root, leaf_index)
                                        → Proof
    shards.tree_layers(document_root)   → list[list[str]]
    shards.audit_event(event_hash)      → AuditEvent | None
    shards.audit_chain(head, limit)     → list[AuditEvent]
    shards.audit_recent(limit)          → list[AuditEvent]
    shards.audit_by_root(root, limit)   → list[AuditEvent]
    shards.audit_since(cursor)          → (list[AuditEvent], dict)
    shards.qa(cache_key)                → QaRecord | None
    shards.qa_recent(limit)             → list[QaRecord]
    shards.qa_by_root(root, limit)      → list[QaRecord]
    shards.qa_search(text, limit)       → list[QaRecord]
    shards.resolve(hex_hash)            → ResolveResult
    shards.counts()                     → ShardCounts

Privacy
-------

By default, ``leaves()`` does not decompress chunk content (the bytes
the leaf hash commits to). Pass ``reveal_private=True`` to opt in. The
chunk content stays compressed-on-disk regardless; reveal only affects
whether the read seam decompresses and returns it.

The ``project=True`` (default) flag adds a deterministic prose
projection (:mod:`arborist.wikitext.to_base`, versioned by
``BASE_VERSION``) alongside the raw bytes. Pass ``project=False`` to
skip the projection when only raw bytes are needed.

Privacy does **not** apply to: hashes, leaf indices, audit-event
metadata, providence_cache rows excluding ``answer_text`` /
``question_text`` (which are part of the Q&A record by design).
"""

from __future__ import annotations

import json
import logging
import sqlite3
from collections.abc import Iterable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

from arborist.compress import unpack_chunk
from arborist.merkle import MerkleTree, proof_to_dict, verify_proof
from arborist.store import connect


log = logging.getLogger(__name__)


__all__ = [
    "open_shards",
    "Shards",
    "Root",
    "Leaf",
    "Proof",
    "AuditEvent",
    "QaRecord",
    "ContextRoot",
    "ResolveResult",
    "ShardCounts",
]


# ---------- dataclasses ------------------------------------------------


@dataclass(frozen=True)
class Root:
    """One document root + its surface metadata. ``shard_path`` lets the
    consumer cite where the answer came from."""

    document_root: str
    document_uri: str
    source_type: str
    kind: str
    title: Optional[str]
    chunking_version: str
    canonicalization_version: str
    schema_version: str
    ingest_ts: int
    hit_count: int
    leaf_count: int
    shard_path: str


@dataclass(frozen=True)
class Leaf:
    """One chunk of a document. ``content`` is the raw bytes-under-the-hash
    (decompressed only when ``reveal_private=True`` was passed). ``prose``
    is the deterministic projection — derivable from ``content``, versioned
    by ``base_version``."""

    document_root: str
    idx: int
    leaf_hash: str
    tier: str
    content: Optional[str] = None
    prose: Optional[str] = None
    base_version: Optional[str] = None


@dataclass(frozen=True)
class Proof:
    """A Merkle inclusion proof. Hash check is performed at construction;
    ``passed`` is final."""

    document_root: str
    leaf_index: int
    leaf_hash: str
    sibling_hashes: list[str]
    left_right_flags: list[bool]
    computed_root: str
    expected_root: str
    passed: bool
    shard_path: str
    raw: dict  # the arborist.merkle.proof_to_dict form, for verifiers


@dataclass(frozen=True)
class AuditEvent:
    seq: int
    event_hash: str
    prev_event_hash: Optional[str]
    event_type: str
    subject_root: Optional[str]
    body: dict
    ts: int
    shard_path: str


@dataclass(frozen=True)
class QaRecord:
    """One providence_cache row — the v9.8 8-dim invariant + the cached
    answer. Only consumers that respect the §14 privacy contract should
    surface ``question_text`` / ``answer_text``."""

    cache_key: str
    source_root: str
    document_uri: str
    question_hash: str
    question_text: str
    answer_text: str
    model_profile_hash: str
    conversation_hash: str
    governance_policy_hash: str
    schema_version: str
    canonicalization_version: str
    chunking_version: str
    falsification_state: str
    audit_mode: Optional[str]
    audit_event_hash: Optional[str]
    run_dag_root: Optional[str]
    created_at: int
    last_hit_at: Optional[int]
    hit_count: int
    n_quotes: Optional[int]
    n_verified: Optional[int]
    verifier_method: Optional[str]
    merkle_proof: dict
    run_dag_blob: dict
    shard_path: str


@dataclass(frozen=True)
class ContextRoot:
    """A synthetic root for a multi-source QA context.

    Arborist's providence_cache rows reference a ``source_root`` that
    isn't always a single document — when a Q&A is answered against
    chunks assembled from several documents, the ``source_root`` is a
    merkle root *over those sources*. That root has no row in the
    ``documents`` table; its evidence manifest lives in the QA record's
    ``merkle_proof`` blob.

    A ``ContextRoot`` exposes that synthesized identity so consumers can
    look up any hash uniformly via the read seam, without knowing whether
    the row landed in ``documents`` or only in ``providence_cache``.
    QA may eventually merge back into the document shards; that's the
    seam's problem, not the consumer's.
    """

    context_root: str
    cache_key: str
    question_text: str
    document_uri: str
    sources: list[dict]
    chunking_version: str
    canonicalization_version: str
    schema_version: str
    created_at: int
    hit_count: int
    shard_path: str
    # Real provenance, derived from `sources`. ``document_uri`` above is
    # headlined with ``primary_source_uri`` (instead of the opaque
    # ``corpus://multi-source`` sentinel) so consumers show where the
    # knowledge actually came from. ``source_domains`` lists the distinct
    # contributing hostnames, primary answer source first.
    primary_source_uri: Optional[str] = None
    source_domains: list[str] = field(default_factory=list)


@dataclass(frozen=True)
class ResolveResult:
    """Outcome of ``shards.resolve(hex_hash)``. ``kind`` is one of:
    document_root | leaf_hash | audit_event | merkle_interior |
    qa_cache_key | context_root | qa_run_dag_root | unknown.

    ``context_root`` fires when the hash matches a ``providence_cache``
    row's ``source_root`` but isn't itself a ``documents`` row — i.e. a
    synthesized multi-source context (see :class:`ContextRoot`)."""

    kind: str
    hash: str
    shard_path: Optional[str] = None
    extra: dict = field(default_factory=dict)


@dataclass(frozen=True)
class ShardCounts:
    documents: int
    audit_events: int
    qa_records: int
    shard_count: int


# ---------- the Shards handle ------------------------------------------


class Shards:
    """Read-only handle over one or more arborist shards.

    Open with :func:`open_shards`. The handle holds a per-shard SQLite
    connection (opened ``mode=ro``) and fans out every query across the
    set. ``mode=ro`` plus ``check_same_thread=False`` makes a single
    connection safe for read-only sharing across a thread pool (waitress,
    gunicorn) — see SQLite docs on shared-read connections.

    The schema versions arborist stores are *not* exposed by Shards as
    a strict version check; that is the caller's job if they want it.
    Schema migrations are arborist's contract, not the seam's.
    """

    def __init__(self, paths: list[str]):
        self._paths: list[str] = list(paths)
        self._conns: dict[str, sqlite3.Connection] = {}
        for p in self._paths:
            try:
                # Run arborist's migration probe via the canonical connect()
                # to bring the schema current, then drop it.
                connect(p).close()
                conn = sqlite3.connect(
                    "file:{0}?mode=ro".format(p), uri=True, check_same_thread=False
                )
                conn.row_factory = sqlite3.Row
                self._conns[p] = conn
                log.debug("arborist.read: opened %s (read-only)", p)
            except Exception as exc:
                log.warning("arborist.read: failed to open shard %s: %s", p, exc)

    # ---------- meta ---------------------------------------------------

    @property
    def paths(self) -> list[str]:
        return list(self._conns.keys())

    def close(self) -> None:
        for c in self._conns.values():
            try:
                c.close()
            except Exception:
                pass
        self._conns.clear()

    def counts(self) -> ShardCounts:
        docs = events = qa = 0
        for c in self._conns.values():
            try:
                docs += int(c.execute("SELECT COUNT(*) FROM documents").fetchone()[0])
            except sqlite3.Error:
                pass
            try:
                events += int(
                    c.execute("SELECT COALESCE(MAX(seq), 0) FROM audit_events").fetchone()[0]
                )
            except sqlite3.Error:
                pass
            try:
                qa += int(c.execute("SELECT COUNT(*) FROM providence_cache").fetchone()[0])
            except sqlite3.Error:
                pass
        return ShardCounts(documents=docs, audit_events=events, qa_records=qa, shard_count=len(self._conns))

    # ---------- documents / roots --------------------------------------

    def root(self, document_root: str) -> Optional[Root]:
        for path, c in self._conns.items():
            r = c.execute(
                "SELECT document_root, document_uri, source_type, kind, title, "
                "       chunking_version, canonicalization_version, schema_version, "
                "       ingest_ts, hit_count "
                "FROM documents WHERE document_root = ?",
                (document_root,),
            ).fetchone()
            if r is None:
                continue
            leaf_count = c.execute(
                "SELECT COUNT(*) FROM chunks WHERE document_root = ?",
                (document_root,),
            ).fetchone()[0]
            return Root(leaf_count=int(leaf_count), shard_path=path, **dict(r))
        return None

    def roots(self, limit: int = 50) -> list[Root]:
        merged: list[Root] = []
        for path, c in self._conns.items():
            rows = c.execute(
                "SELECT document_root, document_uri, source_type, kind, title, "
                "       chunking_version, canonicalization_version, schema_version, "
                "       ingest_ts, hit_count "
                "FROM documents ORDER BY ingest_ts DESC LIMIT ?",
                (limit,),
            ).fetchall()
            for r in rows:
                d = dict(r)
                lc = c.execute(
                    "SELECT COUNT(*) FROM chunks WHERE document_root = ?",
                    (d["document_root"],),
                ).fetchone()[0]
                merged.append(Root(leaf_count=int(lc), shard_path=path, **d))
        merged.sort(key=lambda r: r.ingest_ts, reverse=True)
        return merged[:limit]

    # ---------- leaves / chunks ----------------------------------------

    def leaves(
        self,
        document_root: str,
        *,
        reveal_private: bool = False,
        project: bool = True,
    ) -> list[Leaf]:
        """Read every chunk under ``document_root`` from its shard.

        Decompresses content only when ``reveal_private=True`` (§14
        default-deny). When ``project=True`` (default) and the
        ``arborist[wikitext]`` extra is installed, also computes the
        deterministic prose projection alongside the raw bytes.

        Returns an empty list when the root is unknown.
        """
        conn = self._owning_conn_for_root(document_root)
        if conn is None:
            return []

        rows = conn.execute(
            "SELECT document_root, idx, leaf_hash, content, tier "
            "FROM chunks WHERE document_root = ? ORDER BY idx",
            (document_root,),
        ).fetchall()

        to_base, base_version = _get_wikitext_projector() if project else (None, None)

        out: list[Leaf] = []
        for r in rows:
            content: Optional[str] = None
            prose: Optional[str] = None
            stamp: Optional[str] = None
            if reveal_private and r["content"] is not None:
                try:
                    content = unpack_chunk(r["content"])
                except Exception as exc:
                    log.warning("unpack_chunk failed for leaf %s: %s", r["leaf_hash"], exc)
                if content is not None and to_base is not None:
                    try:
                        prose = to_base(content)
                        stamp = base_version
                    except Exception as exc:
                        log.warning("wikitext.to_base failed for leaf %s: %s", r["leaf_hash"], exc)
            out.append(
                Leaf(
                    document_root=r["document_root"],
                    idx=int(r["idx"]),
                    leaf_hash=r["leaf_hash"],
                    tier=r["tier"],
                    content=content,
                    prose=prose,
                    base_version=stamp,
                )
            )
        return out

    # ---------- proofs / tree ------------------------------------------

    def proof(self, document_root: str, leaf_index: int) -> Optional[Proof]:
        """Compute a Merkle inclusion proof against the shard that owns
        the root. Returns ``None`` if the root is unknown; raises
        ``IndexError`` if ``leaf_index`` is out of range."""
        conn = self._owning_conn_for_root(document_root)
        path = self._path_for_root(document_root)
        if conn is None or path is None:
            return None
        rows = conn.execute(
            "SELECT idx, leaf_hash FROM chunks WHERE document_root = ? ORDER BY idx",
            (document_root,),
        ).fetchall()
        if not rows:
            return None
        if leaf_index < 0 or leaf_index >= len(rows):
            raise IndexError(
                "leaf_index {0} out of range (root has {1} leaves)".format(
                    leaf_index, len(rows)
                )
            )
        leaves = [bytes.fromhex(r["leaf_hash"]) for r in rows]
        tree = MerkleTree.build(leaves)
        p = tree.proof(leaf_index)
        passed = verify_proof(p) and tree.root.hex() == document_root
        raw = proof_to_dict(p)
        return Proof(
            document_root=document_root,
            leaf_index=leaf_index,
            leaf_hash=p.leaf.hex(),
            sibling_hashes=[s["hash"] for s in raw["siblings"]],
            left_right_flags=[bool(s["is_left"]) for s in raw["siblings"]],
            computed_root=tree.root.hex(),
            expected_root=document_root,
            passed=passed,
            shard_path=path,
            raw=raw,
        )

    def tree_layers(self, document_root: str) -> Optional[list[list[str]]]:
        """All Merkle tree layers as hex (layer 0 = leaves, last = root).
        Drives the 3D lattice visualization."""
        conn = self._owning_conn_for_root(document_root)
        if conn is None:
            return None
        rows = conn.execute(
            "SELECT leaf_hash FROM chunks WHERE document_root = ? ORDER BY idx",
            (document_root,),
        ).fetchall()
        if not rows:
            return None
        leaves = [bytes.fromhex(r["leaf_hash"]) for r in rows]
        tree = MerkleTree.build(leaves)
        return [[node.hex() for node in layer] for layer in tree.layers]

    # ---------- audit --------------------------------------------------

    def audit_event(self, event_hash: str) -> Optional[AuditEvent]:
        for path, c in self._conns.items():
            r = c.execute(
                "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts "
                "FROM audit_events WHERE event_hash = ?",
                (event_hash,),
            ).fetchone()
            if r is not None:
                return _row_to_audit(r, path)
        return None

    def audit_chain(self, head_event_hash: str, *, limit: int = 100) -> list[AuditEvent]:
        out: list[AuditEvent] = []
        cursor = head_event_hash
        for _ in range(limit):
            if cursor is None:
                break
            ev = self.audit_event(cursor)
            if ev is None:
                break
            out.append(ev)
            cursor = ev.prev_event_hash
        return out

    def audit_recent(self, *, limit: int = 100) -> list[AuditEvent]:
        merged: list[AuditEvent] = []
        for path, c in self._conns.items():
            for r in c.execute(
                "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts "
                "FROM audit_events ORDER BY seq DESC LIMIT ?",
                (limit,),
            ).fetchall():
                merged.append(_row_to_audit(r, path))
        merged.sort(key=lambda e: e.ts, reverse=True)
        return merged[:limit]

    def audit_by_root(self, root: str, *, limit: int = 50) -> list[AuditEvent]:
        merged: list[AuditEvent] = []
        for path, c in self._conns.items():
            for r in c.execute(
                "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts "
                "FROM audit_events WHERE subject_root = ? ORDER BY seq DESC LIMIT ?",
                (root, limit),
            ).fetchall():
                merged.append(_row_to_audit(r, path))
        merged.sort(key=lambda e: e.seq, reverse=True)
        return merged[:limit]

    def audit_since(
        self, cursor: Optional[dict] = None, *, limit: int = 200
    ) -> tuple[list[AuditEvent], dict]:
        """Stream-friendly tail. ``cursor`` is opaque to the caller — pass
        ``None`` first time, then pass the dict returned from the previous
        call. Returns ``(new_events, next_cursor)``.

        Each shard tracks its own ``last_seq`` so the seam can drive an
        SSE stream over all shards without the consumer knowing the
        sharding layout.
        """
        cursor = dict(cursor or {})
        new_events: list[AuditEvent] = []
        for path, c in self._conns.items():
            last_seq = int(cursor.get(path, 0))
            for r in c.execute(
                "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts "
                "FROM audit_events WHERE seq > ? ORDER BY seq ASC LIMIT ?",
                (last_seq, limit),
            ).fetchall():
                ev = _row_to_audit(r, path)
                new_events.append(ev)
                cursor[path] = max(int(cursor.get(path, 0)), ev.seq)
        return new_events, cursor

    def audit_cursor(self) -> dict:
        """Cursor at the current head of every shard (use as starting
        point for a tail that only wants future events)."""
        return {
            path: int(c.execute("SELECT COALESCE(MAX(seq), 0) FROM audit_events").fetchone()[0])
            for path, c in self._conns.items()
        }

    # ---------- providence_cache (Q&A) ---------------------------------

    def context(self, context_root: str) -> Optional[ContextRoot]:
        """Synthesize a :class:`ContextRoot` for a multi-source QA root.

        Looks up the first ``providence_cache`` row whose ``source_root``
        equals ``context_root`` and surfaces the evidence manifest parsed
        out of its ``merkle_proof`` JSON. Returns ``None`` when no QA
        record cites this root.

        Callers should fall through from :meth:`root` to this method when
        a hash is "a root but not a document" — that case happens when
        the QA pipeline answered against assembled context rather than a
        single ingested document. Consumers don't need to know which
        case applies; they just chain the calls.
        """
        for path, c in self._conns.items():
            try:
                r = c.execute(
                    _QA_SELECT + " WHERE source_root = ? LIMIT 1",
                    (context_root,),
                ).fetchone()
            except sqlite3.Error:
                continue
            if r is None:
                continue
            qa = _row_to_qa(r, path)
            sources = qa.merkle_proof.get("sources", []) if isinstance(qa.merkle_proof, dict) else []
            primary_uri, domains = _summarize_sources(sources)
            # Headline the real source. The stored document_uri for a
            # multi-source answer is the opaque ``corpus://multi-source``
            # sentinel — replace it with the primary answer source so the
            # node says where the knowledge came from (e.g.
            # russell.ballestrini.net), not a placeholder.
            headline_uri = qa.document_uri or ""
            if (not headline_uri or headline_uri == _MULTI_SOURCE_URI) and primary_uri:
                headline_uri = primary_uri
            return ContextRoot(
                context_root=context_root,
                cache_key=qa.cache_key,
                question_text=qa.question_text or "",
                document_uri=headline_uri,
                sources=sources,
                chunking_version=qa.chunking_version,
                canonicalization_version=qa.canonicalization_version,
                schema_version=qa.schema_version,
                created_at=qa.created_at,
                hit_count=qa.hit_count,
                shard_path=path,
                primary_source_uri=primary_uri,
                source_domains=domains,
            )
        return None

    def context_layers(self, context_root: str) -> Optional[list[list[str]]]:
        """Rebuild the context tree's layers from the sorted source roots.

        Mirrors :meth:`tree_layers` for synthetic context roots. Layer 0
        is leaves (the sorted source document_roots); the last layer is
        ``[context_root]``. Uses the same canonicalization as
        :func:`arborist.qa.query._context_root` — sort source roots, build
        a :class:`MerkleTree`, return its layers as hex.

        Returns ``None`` when the context root isn't in providence_cache.
        """
        ctx = self.context(context_root)
        if ctx is None:
            return None
        leaves_hex = _context_leaf_hashes(ctx.sources)
        if not leaves_hex:
            return [[context_root]]
        leaves = [bytes.fromhex(h) for h in leaves_hex]
        tree = MerkleTree.build(leaves)
        return [[node.hex() for node in layer] for layer in tree.layers]

    def context_proof(self, context_root: str, leaf_index: int) -> Optional[Proof]:
        """Inclusion proof for one source under a context root.

        Mirrors :meth:`proof`. The leaf at index ``leaf_index`` is the
        sorted-source document_root. Raises ``IndexError`` if out of
        range. Returns ``None`` when the context root isn't known.
        """
        ctx = self.context(context_root)
        if ctx is None:
            return None
        leaves_hex = _context_leaf_hashes(ctx.sources)
        if not leaves_hex:
            return None
        if leaf_index < 0 or leaf_index >= len(leaves_hex):
            raise IndexError(
                "leaf_index {0} out of range (context has {1} sources)".format(
                    leaf_index, len(leaves_hex)
                )
            )
        leaves = [bytes.fromhex(h) for h in leaves_hex]
        tree = MerkleTree.build(leaves)
        p = tree.proof(leaf_index)
        passed = verify_proof(p) and tree.root.hex() == context_root
        raw = proof_to_dict(p)
        return Proof(
            document_root=context_root,
            leaf_index=leaf_index,
            leaf_hash=p.leaf.hex(),
            sibling_hashes=[s["hash"] for s in raw["siblings"]],
            left_right_flags=[bool(s["is_left"]) for s in raw["siblings"]],
            computed_root=tree.root.hex(),
            expected_root=context_root,
            passed=passed,
            shard_path=ctx.shard_path,
            raw=raw,
        )

    def qa(self, cache_key: str) -> Optional[QaRecord]:
        for path, c in self._conns.items():
            r = c.execute(
                _QA_SELECT + " WHERE cache_key = ?", (cache_key,)
            ).fetchone()
            if r is not None:
                return _row_to_qa(r, path)
        return None

    def qa_recent(self, *, limit: int = 50) -> list[QaRecord]:
        merged: list[QaRecord] = []
        for path, c in self._conns.items():
            for r in c.execute(
                _QA_SELECT + " ORDER BY created_at DESC LIMIT ?", (limit,)
            ).fetchall():
                merged.append(_row_to_qa(r, path))
        merged.sort(key=lambda q: q.created_at, reverse=True)
        return merged[:limit]

    def qa_by_root(self, source_root: str, *, limit: int = 50) -> list[QaRecord]:
        merged: list[QaRecord] = []
        for path, c in self._conns.items():
            for r in c.execute(
                _QA_SELECT + " WHERE source_root = ? ORDER BY created_at DESC LIMIT ?",
                (source_root, limit),
            ).fetchall():
                merged.append(_row_to_qa(r, path))
        merged.sort(key=lambda q: q.created_at, reverse=True)
        return merged[:limit]

    def qa_search(self, text: str, *, limit: int = 50) -> list[QaRecord]:
        """Substring search over ``question_text`` (case-insensitive).

        FTS5 isn't indexed on providence_cache rows in the current
        arborist schema; a LIKE scan over 4-5k QA records is cheap
        enough for v1. Swap to FTS5 if the corpus grows past ~100k
        records.
        """
        pat = "%{0}%".format(text.replace("%", r"\%").replace("_", r"\_"))
        merged: list[QaRecord] = []
        for path, c in self._conns.items():
            for r in c.execute(
                _QA_SELECT + " WHERE question_text LIKE ? ESCAPE '\\' "
                "ORDER BY created_at DESC LIMIT ?",
                (pat, limit),
            ).fetchall():
                merged.append(_row_to_qa(r, path))
        merged.sort(key=lambda q: q.created_at, reverse=True)
        return merged[:limit]

    # ---------- hash resolver ------------------------------------------

    def resolve(self, hex_hash: str) -> ResolveResult:
        """Classify any hex hash. Order of search:

        document_root → leaf_hash → audit event_hash → providence_cache
        cache_key → providence_cache source_root → providence_cache
        run_dag_root → merkle_nodes interior hash → unknown.
        """
        h = hex_hash.lower().strip()

        for path, c in self._conns.items():
            if c.execute("SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", (h,)).fetchone():
                return ResolveResult(kind="document_root", hash=h, shard_path=path)

        for path, c in self._conns.items():
            r = c.execute("SELECT document_root, idx FROM chunks WHERE leaf_hash = ? LIMIT 1", (h,)).fetchone()
            if r:
                return ResolveResult(
                    kind="leaf_hash", hash=h, shard_path=path,
                    extra={"document_root": r["document_root"], "leaf_index": int(r["idx"])},
                )

        for path, c in self._conns.items():
            r = c.execute("SELECT subject_root FROM audit_events WHERE event_hash = ? LIMIT 1", (h,)).fetchone()
            if r:
                return ResolveResult(
                    kind="audit_event", hash=h, shard_path=path,
                    extra={"subject_root": r["subject_root"]},
                )

        for path, c in self._conns.items():
            try:
                r = c.execute(
                    "SELECT source_root FROM providence_cache WHERE cache_key = ? LIMIT 1", (h,)
                ).fetchone()
                if r:
                    return ResolveResult(
                        kind="qa_cache_key", hash=h, shard_path=path,
                        extra={"source_root": r["source_root"]},
                    )
            except sqlite3.Error:
                continue

        # providence_cache.source_root: the QA's evidence root. Reaches
        # this branch only when the hash isn't in ``documents`` (the
        # documents probe runs first and returns "document_root"), so
        # any match here is a synthesized multi-source context.
        for path, c in self._conns.items():
            try:
                r = c.execute(
                    "SELECT cache_key FROM providence_cache WHERE source_root = ? LIMIT 1", (h,)
                ).fetchone()
                if r:
                    return ResolveResult(
                        kind="context_root", hash=h, shard_path=path,
                        extra={"cache_key": r["cache_key"]},
                    )
            except sqlite3.Error:
                continue

        for path, c in self._conns.items():
            try:
                r = c.execute(
                    "SELECT cache_key FROM providence_cache WHERE run_dag_root = ? LIMIT 1", (h,)
                ).fetchone()
                if r:
                    return ResolveResult(
                        kind="qa_run_dag_root", hash=h, shard_path=path,
                        extra={"cache_key": r["cache_key"]},
                    )
            except sqlite3.Error:
                continue

        for path, c in self._conns.items():
            r = c.execute(
                "SELECT document_root, layer, idx FROM merkle_nodes WHERE hash = ? LIMIT 1", (h,)
            ).fetchone()
            if r:
                return ResolveResult(
                    kind="merkle_interior", hash=h, shard_path=path,
                    extra={
                        "document_root": r["document_root"],
                        "layer": int(r["layer"]),
                        "idx": int(r["idx"]),
                    },
                )

        return ResolveResult(kind="unknown", hash=h)

    # ---------- internals ----------------------------------------------

    def _owning_conn_for_root(self, document_root: str) -> Optional[sqlite3.Connection]:
        for _path, c in self._conns.items():
            if c.execute(
                "SELECT 1 FROM documents WHERE document_root = ? LIMIT 1",
                (document_root,),
            ).fetchone():
                return c
        return None

    def _path_for_root(self, document_root: str) -> Optional[str]:
        for path, c in self._conns.items():
            if c.execute(
                "SELECT 1 FROM documents WHERE document_root = ? LIMIT 1",
                (document_root,),
            ).fetchone():
                return path
        return None


def open_shards(paths: Iterable[str | Path]) -> Shards:
    """Open every shard at ``paths`` (read-only) and return a ``Shards``
    handle. Shards that fail to open are skipped with a warning."""
    return Shards([str(p) for p in paths])


# ---------- row decoders -----------------------------------------------


_QA_SELECT = (
    "SELECT cache_key, source_root, document_uri, question_hash, question_text, "
    "       answer_text, model_profile_hash, conversation_hash, "
    "       governance_policy_hash, schema_version, canonicalization_version, "
    "       chunking_version, falsification_state, audit_mode, audit_event_hash, "
    "       run_dag_root, created_at, last_hit_at, hit_count, n_quotes, "
    "       n_verified, verifier_method, merkle_proof, run_dag_blob "
    "FROM providence_cache"
)


# Sentinel ``document_uri`` stored on multi-source providence_cache rows
# (see ``arborist.qa.query``). Opaque on purpose at write time; the read
# seam projects it back to real provenance via ``_summarize_sources``.
_MULTI_SOURCE_URI = "corpus://multi-source"


def _domain_of(uri: str) -> str:
    """Hostname of a source URI, or '' for opaque/relative URIs."""
    try:
        from urllib.parse import urlparse

        return (urlparse(uri).netloc or "").lower()
    except Exception:
        return ""


def _summarize_sources(sources: list[dict]) -> tuple[Optional[str], list[str]]:
    """Derive (primary_source_uri, source_domains) from a context root's
    sources. The primary answer source ranks first; remaining sources keep
    their order. Domains are de-duplicated, primary first."""
    if not sources:
        return None, []
    primaries = [
        s for s in sources
        if isinstance(s, dict) and s.get("source_role") == "primary_answer_source"
    ]
    others = [s for s in sources if isinstance(s, dict) and s not in primaries]
    ordered = primaries + others
    primary_uri = None
    for s in ordered:
        if s.get("document_uri"):
            primary_uri = s["document_uri"]
            break
    domains: list[str] = []
    for s in ordered:
        dom = _domain_of(s.get("document_uri") or "")
        if dom and dom not in domains:
            domains.append(dom)
    return primary_uri, domains


def _context_leaf_hashes(sources: list[dict]) -> list[str]:
    """Canonical leaf hashes for a context tree.

    Mirrors :func:`arborist.qa.query._context_root` — the leaves are the
    sources' ``document_root`` values, sorted lexicographically. Keeping
    this helper next to :meth:`Shards.context_layers` /
    :meth:`Shards.context_proof` lets the read seam reconstruct the tree
    and prove inclusion without re-importing the QA pipeline.
    """
    roots = [s.get("document_root") for s in sources if isinstance(s, dict) and s.get("document_root")]
    return sorted(r.lower() for r in roots)


def _row_to_audit(r: sqlite3.Row, shard_path: str) -> AuditEvent:
    try:
        body = json.loads(r["body"]) if r["body"] else {}
    except Exception:
        body = {"_raw": r["body"]}
    return AuditEvent(
        seq=int(r["seq"]),
        event_hash=r["event_hash"],
        prev_event_hash=r["prev_event_hash"],
        event_type=r["event_type"],
        subject_root=r["subject_root"],
        body=body,
        ts=int(r["ts"]),
        shard_path=shard_path,
    )


def _row_to_qa(r: sqlite3.Row, shard_path: str) -> QaRecord:
    def _json_or_empty(s):
        if not s:
            return {}
        try:
            return json.loads(s)
        except Exception:
            return {"_raw": s}

    return QaRecord(
        cache_key=r["cache_key"],
        source_root=r["source_root"],
        document_uri=r["document_uri"],
        question_hash=r["question_hash"],
        question_text=r["question_text"],
        answer_text=r["answer_text"],
        model_profile_hash=r["model_profile_hash"],
        conversation_hash=r["conversation_hash"],
        governance_policy_hash=r["governance_policy_hash"],
        schema_version=r["schema_version"],
        canonicalization_version=r["canonicalization_version"],
        chunking_version=r["chunking_version"],
        falsification_state=r["falsification_state"],
        audit_mode=r["audit_mode"] if "audit_mode" in r.keys() else None,
        audit_event_hash=r["audit_event_hash"] if "audit_event_hash" in r.keys() else None,
        run_dag_root=r["run_dag_root"] if "run_dag_root" in r.keys() else None,
        created_at=int(r["created_at"]),
        last_hit_at=int(r["last_hit_at"]) if r["last_hit_at"] is not None else None,
        hit_count=int(r["hit_count"]),
        n_quotes=int(r["n_quotes"]) if "n_quotes" in r.keys() and r["n_quotes"] is not None else None,
        n_verified=int(r["n_verified"]) if "n_verified" in r.keys() and r["n_verified"] is not None else None,
        verifier_method=r["verifier_method"] if "verifier_method" in r.keys() else None,
        merkle_proof=_json_or_empty(r["merkle_proof"]),
        run_dag_blob=_json_or_empty(r["run_dag_blob"]) if "run_dag_blob" in r.keys() else {},
        shard_path=shard_path,
    )


def _get_wikitext_projector():
    """Resolve arborist.wikitext.to_base lazily (the wikitext extra is
    optional). Returns ``(callable, base_version)`` or ``(None, None)``."""
    try:
        from arborist.wikitext import to_base, BASE_VERSION
        return to_base, BASE_VERSION
    except Exception:
        return None, None