qa/corpus: Corpus Protocol + SqliteShardCorpus + SidecarBucketCorpus

First step toward DRYing the local-vs-cloud retrieval diff. Defines a
minimal Corpus Protocol (Hit, ChunkRow, fts_body, fts_title, fts_phrase,
chunks_for_doc, snapshot_root) and ships two adapters:

  SqliteShardCorpus    wraps a connect_query() connection. Implements
                       fts_body (via _to_fts5 sanitizer + chunks_fts
                       MATCH) and chunks_for_doc + snapshot_root.
                       fts_title / fts_phrase raise NotSupportedError
                       (those routes will move out of query.py inline
                       SQL in the next refactor pass).

  SidecarBucketCorpus  wraps a MultiShardSidecarCorpus. fts_body
                       delegates to the sidecar BM25 + title-boost +
                       extras-penalty + title-relevance pipeline.
                       chunks_for_doc walks the shard's apsw conn via
                       the bucket VFS, reusing warm page cache.
                       Title / phrase routes raise NotSupportedError
                       until the sidecar format ships those indexes.

Together: query() future-refactor takes a Corpus parameter, runs the
routes the adapter supports, skips NotSupportedError, falls through.
Every quality fix (today: possessive stem, accent fold, numeral fold,
extras-penalty title-boost, title-relevance filter) lands once and
both backends consume it through the protocol.

Tests (20 passing, three layers):

  tests/test_qa_corpus.py             (unit) — Hit/ChunkRow invariants,
    Protocol conformance (runtime_checkable isinstance), NotSupportedError
    on unimplemented routes, fts_body sanitization (no FTS5 syntax leak),
    stopword-only query → [], chunks_for_doc idx-ascending shape.

  tests/test_qa_corpus_integration.py (integration) — real corpus →
    both adapters → assert recall (target doc in top-K of both) and
    byte-identical chunks_for_doc decoding. Catches tokenizer/stem/fold
    drift between FTS5's unicode61 and the sidecar's NFKD+ASCII path.

  tests/test_qa_corpus_functional.py  (functional) — full retrieval +
    StubClient LLM + verify_claim_lattice pipeline routed via the
    protocol for both adapters; asserts shared audit_mode + primary
    source agreement. Prototype of what the post-refactor query() does.

Side fix: MultiShardSidecarCorpus title-relevance filter now falls
open when the title tokenizes to zero content tokens (single-char or
all-stopword titles like "A" or "I" shouldn't be dropped just because
the filter side has nothing to match on).

arborist/qa/corpus.py

@@ -0,0 +1,339 @@
+"""Corpus protocol — single retrieval interface for local + cloud.
+
+Today's `arborist.qa.query.query()` is tightly coupled to a SQLite
+connection (`connect_query(shards_dir=…)`), and the cloud path
+(`arborist cloud query`) re-implements retrieval against a sidecar
+bucket. Every quality fix has to ship twice. This module defines a
+minimal Corpus protocol so retrieval can be swapped (local SQLite,
+HTTP sidecar, future edge-proxy) while the orchestrator stays single-
+source.
+
+Three pieces:
+
+  - ``Hit``      — what a retrieval route returns
+  - ``ChunkRow`` — what an evidence-builder reads
+  - ``Corpus``   — Protocol with the routes query() needs
+
+Two adapters live in this module:
+
+  - ``SqliteShardCorpus`` — wraps a connect_query() connection
+  - ``SidecarBucketCorpus`` — wraps MultiShardSidecarCorpus
+
+Future query() refactor will take a ``Corpus`` parameter; until then,
+adapters can be used standalone to test parity between backends or to
+back the cloud-query path without re-implementing prompt/verify/render
+logic.
+
+Routes an adapter doesn't natively support raise ``NotSupportedError``.
+The orchestrator catches and skips — graceful degradation, no silent
+result skew. Sidecar today only supports ``fts_body``; phrase/title-
+LIKE/core-keyword routes will land as the sidecar format learns to
+ship them.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Iterable, Iterator, Optional, Protocol, runtime_checkable
+
+
+class NotSupportedError(NotImplementedError):
+    """Raised by an adapter when a route is not implemented for that
+    backend. The orchestrator should catch and skip — never propagate."""
+
+
+@dataclass(frozen=True)
+class Hit:
+    """One retrieval result, backend-agnostic.
+
+    ``score`` semantics differ per backend (FTS5 BM25 ranges negative
+    [-20, 0]; sidecar BM25 ranges positive [+25, +60]). Callers that
+    merge across backends must use rank or a normalization layer —
+    raw score is NOT comparable across adapters. Within a single
+    backend, lower-is-better is the FTS5 convention; higher-is-better
+    is the sidecar convention. The adapter sets ``higher_is_better``
+    on construction so callers can sort uniformly.
+    """
+    document_root: str
+    document_uri: str
+    title: str
+    score: float
+    shard_id: Optional[str] = None    # which shard surfaced this hit (debug)
+    extras: dict = field(default_factory=dict)  # adapter-specific metadata
+
+
+@dataclass(frozen=True)
+class ChunkRow:
+    """One chunk's evidence-builder shape.
+
+    ``content`` is decompressed UTF-8 prose (callers don't need to know
+    about the storage encoding). ``leaf_hash`` is the Merkle leaf —
+    required for wallet-side verification, so adapters that can't
+    surface it (synthesized chunks?) should explicitly return None.
+    """
+    document_root: str
+    idx: int
+    leaf_hash: Optional[str]
+    content: str
+
+
+@runtime_checkable
+class Corpus(Protocol):
+    """The retrieval surface query() depends on.
+
+    Implementations:
+      - SqliteShardCorpus    — local, full feature set
+      - SidecarBucketCorpus  — cloud (HTTP), fts_body + chunks_for_doc
+
+    Future routes (phrase_5gram, core_keyword, title_like) will land
+    here as methods that adapters either implement or raise
+    NotSupportedError.
+    """
+
+    # Backend identity (for debug + score-comparison decisions).
+    @property
+    def name(self) -> str: ...
+
+    # `True` if scores from this adapter sort higher-is-better.
+    @property
+    def higher_is_better(self) -> bool: ...
+
+    # --- retrieval routes ---
+
+    def fts_body(self, query: str, *, limit: int = 8) -> list[Hit]:
+        """Full-text body search. Tokenizer-agnostic from the caller's
+        POV: pass a natural-language string; the adapter sanitizes."""
+        ...
+
+    def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
+        """Title-only search. SidecarBucketCorpus today falls back to
+        fts_body + title-boost re-rank inside its search() call, so
+        raises NotSupportedError to make the orchestrator's intent
+        explicit when title-LIKE is desired as a separate route."""
+        ...
+
+    def fts_phrase(
+        self, ngrams: Iterable[str], *, limit: int = 8
+    ) -> list[Hit]:
+        """Verbatim n-gram phrase match. Closes the allusion gap (e.g.
+        "always been at war" → 1984). SidecarBucketCorpus raises
+        NotSupportedError until the sidecar format ships a phrase
+        index."""
+        ...
+
+    # --- evidence reads ---
+
+    def chunks_for_doc(
+        self, document_root: str, *, limit: int | None = None
+    ) -> list[ChunkRow]:
+        """Per-doc chunk rows in idx-ascending order."""
+        ...
+
+    # --- corpus identity ---
+
+    def snapshot_root(self) -> str:
+        """Hex of the corpus Merkle commit. Required for wallet-side
+        provenance binding."""
+        ...
+
+
+# ===========================================================================
+# Adapter: SqliteShardCorpus — local SQLite shards via existing connect().
+# ===========================================================================
+
+
+class SqliteShardCorpus:
+    """Adapter over a local SQLite connection (single-shard or multi-
+    shard via connect_query()). Default for `arborist query`.
+
+    Today this implements only fts_body + chunks_for_doc + snapshot_root.
+    Other routes raise NotSupportedError; the future query.py refactor
+    will lift the existing in-line SQL for title-LIKE / phrase / core-
+    keyword into this adapter so they land here once and both backends
+    consume them through the protocol.
+    """
+
+    name = "sqlite-shard"
+    higher_is_better = False  # FTS5 BM25 convention (smaller score = stronger)
+
+    def __init__(self, conn):
+        """``conn`` is a sqlite3.Connection (or connect_query() result)
+        with ``chunks``, ``documents``, ``chunks_fts`` accessible. Set
+        ``row_factory = sqlite3.Row`` so SELECT * yields keyed rows."""
+        import sqlite3 as _sqlite
+        if conn.row_factory is None:
+            conn.row_factory = _sqlite.Row
+        self._conn = conn
+
+    def fts_body(self, query: str, *, limit: int = 8) -> list[Hit]:
+        sql = (
+            "SELECT d.document_root, d.document_uri, d.title, "
+            "       bm25(chunks_fts) AS score "
+            "FROM chunks_fts JOIN chunks c ON c.rowid = chunks_fts.rowid "
+            "JOIN documents d ON d.document_root = c.document_root "
+            "WHERE chunks_fts MATCH ? "
+            "ORDER BY score LIMIT ?"
+        )
+        # The caller passes natural-language; sanitize on this side so
+        # FTS5 doesn't choke on punctuation. Mirrors the bucket-direct
+        # sanitizer (arborist.wallet.bucket._to_fts5).
+        from arborist.wallet.bucket import _to_fts5
+        match_expr = _to_fts5(query)
+        if not match_expr.strip():
+            return []
+        rows = list(self._conn.execute(sql, (match_expr, limit)))
+        return [
+            Hit(
+                document_root=r["document_root"],
+                document_uri=r["document_uri"] or "",
+                title=r["title"] or "",
+                score=r["score"],
+                shard_id=None,
+            )
+            for r in rows
+        ]
+
+    def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
+        raise NotSupportedError(
+            "fts_title not yet routed through SqliteShardCorpus — "
+            "query.py owns title-LIKE today; will move in the next refactor."
+        )
+
+    def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]:
+        raise NotSupportedError(
+            "fts_phrase not yet routed through SqliteShardCorpus — "
+            "query.py owns phrase-pattern today; will move in the next refactor."
+        )
+
+    def chunks_for_doc(
+        self, document_root: str, *, limit: int | None = None
+    ) -> list[ChunkRow]:
+        from arborist.compress import unpack_chunk
+        sql = (
+            "SELECT idx, leaf_hash, content FROM chunks "
+            "WHERE document_root = ? AND content IS NOT NULL "
+            "ORDER BY idx ASC"
+        )
+        if limit is not None:
+            sql += " LIMIT ?"
+            params = (document_root, limit)
+        else:
+            params = (document_root,)
+        rows = list(self._conn.execute(sql, params))
+        out: list[ChunkRow] = []
+        for r in rows:
+            text = unpack_chunk(r["content"]) or ""
+            if not text:
+                continue
+            out.append(ChunkRow(
+                document_root=document_root,
+                idx=r["idx"],
+                leaf_hash=r["leaf_hash"],
+                content=text,
+            ))
+        return out
+
+    def snapshot_root(self) -> str:
+        from arborist.snapshot import compute_snapshot_root
+        root, _ = compute_snapshot_root(self._conn)
+        return root
+
+
+# ===========================================================================
+# Adapter: SidecarBucketCorpus — multi-shard sidecar over HTTP.
+# ===========================================================================
+
+
+class SidecarBucketCorpus:
+    """Adapter over MultiShardSidecarCorpus (sidecar BM25 + bucket-direct
+    chunk reads). Default for `arborist cloud query`.
+
+    fts_body delegates to the sidecar's BM25 + title boost + extras
+    penalty. fts_title / fts_phrase raise NotSupportedError until the
+    sidecar format ships those indexes.
+    """
+
+    name = "sidecar-bucket"
+    higher_is_better = True  # sidecar BM25 + boost is positive, larger = stronger
+
+    def __init__(self, multi_shard):
+        """``multi_shard`` is a MultiShardSidecarCorpus (or any object
+        with fts_search(query, limit) → [{document_root, document_uri,
+        title, score, _shard_url}] and conn_for_shard(shard_url) for
+        chunk reads)."""
+        self._mscorpus = multi_shard
+
+    def fts_body(self, query: str, *, limit: int = 8) -> list[Hit]:
+        hits = self._mscorpus.fts_search(query, limit=limit)
+        return [
+            Hit(
+                document_root=h["document_root"],
+                document_uri=h.get("document_uri", ""),
+                title=h.get("title", ""),
+                score=h.get("score", 0.0),
+                shard_id=h.get("_shard_url"),
+                extras={"merge_score": h.get("merge_score")} if "merge_score" in h else {},
+            )
+            for h in hits
+        ]
+
+    def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
+        raise NotSupportedError(
+            "fts_title not implemented for SidecarBucketCorpus — "
+            "sidecar today fuses title-boost into fts_body. A separate "
+            "title-only route would need a sidecar v3 with a title index."
+        )
+
+    def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]:
+        raise NotSupportedError(
+            "fts_phrase not implemented for SidecarBucketCorpus — "
+            "phrase index is sidecar v3 work."
+        )
+
+    def chunks_for_doc(
+        self, document_root: str, *, limit: int | None = None
+    ) -> list[ChunkRow]:
+        # We don't know which shard owns the doc without a lookup. The
+        # caller of fts_body already has Hit.shard_id; the contract here
+        # accepts a bare document_root and scans every shard's bucket
+        # connection until one returns rows. Cheap because BucketClient
+        # caches the .db pages it touched for the FTS search.
+        from arborist.compress import unpack_chunk
+
+        # Try each shard's conn.
+        for sh in getattr(self._mscorpus, "manifest", None).shards:
+            try:
+                conn = self._mscorpus.conn_for_shard(sh["url"])
+            except (KeyError, AttributeError):
+                continue
+            sql = (
+                "SELECT idx, leaf_hash, content FROM chunks "
+                "WHERE document_root = ? AND content IS NOT NULL "
+                "ORDER BY idx ASC"
+            )
+            if limit is not None:
+                sql += f" LIMIT {int(limit)}"
+            try:
+                rows = list(conn.execute(sql, (document_root,)))
+            except Exception:
+                continue
+            if not rows:
+                continue
+            out: list[ChunkRow] = []
+            for r in rows:
+                idx, leaf_hash, content_blob = r
+                text = unpack_chunk(content_blob) or ""
+                if not text:
+                    continue
+                out.append(ChunkRow(
+                    document_root=document_root,
+                    idx=idx, leaf_hash=leaf_hash, content=text,
+                ))
+            if out:
+                return out
+        return []
+
+    def snapshot_root(self) -> str:
+        # Manifest may carry a pre-computed snapshot_root; fall back to
+        # empty hex when unknown (cloud manifests today leave it null).
+        sr = getattr(self._mscorpus.manifest, "snapshot_root", None)
+        return sr or ("00" * 32)

arborist/wallet/bucket.py

@@ -906,6 +906,13 @@ class MultiShardSidecarCorpus:
                 if not title:
                     return False
                 title_tokens = set(tokenize_text(title))
+                # Fall-open when the title has no content tokens after
+                # stopword/length filtering — single-char or all-stopword
+                # titles ("A", "I", "Of Mice and Men") shouldn't be
+                # filtered out just because the FILTER side has nothing
+                # to match on. Keep the hit; let BM25 + title-boost rank.
+                if not title_tokens:
+                    return True
                 return bool(query_tokens & title_tokens)
             per_shard = [
                 (sh_url, [h for h in hits if _title_relevant(h)])

tests/test_qa_corpus.py

@@ -0,0 +1,221 @@
+"""Unit tests for arborist.qa.corpus — adapter contract conformance.
+
+What this asserts:
+  * Both adapters implement the Corpus Protocol (runtime_checkable)
+  * Hit + ChunkRow are frozen and hashable
+  * NotSupportedError raised by routes adapters don't implement
+  * SqliteShardCorpus.fts_body sanitizes natural-language input
+    (no FTS5 syntax leaks through punctuation)
+  * Empty / stopword-only queries return [] rather than raising
+
+What this does NOT assert (covered in integration / functional):
+  * Cross-adapter result agreement
+  * Per-chunk content decompression equivalence
+  * End-to-end LLM + verifier behaviour
+"""
+from __future__ import annotations
+
+import sqlite3
+from typing import Iterator
+
+import pytest
+
+from arborist.document import Document
+from arborist.ingest import ingest_source
+from arborist.qa.corpus import (
+    ChunkRow,
+    Corpus,
+    Hit,
+    NotSupportedError,
+    SidecarBucketCorpus,
+    SqliteShardCorpus,
+)
+from arborist.source import Source
+from arborist.store import connect
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+class _FakeSource(Source):
+    source_type = "test"
+    def __init__(self, ds): self.ds = ds
+    def iter_documents(self) -> Iterator[Document]:
+        yield from self.ds
+
+
+def _doc(uri: str, content: str, title: str | None = None) -> Document:
+    return Document(
+        uri=uri, content=content, source_type="test",
+        title=title or uri.split("/")[-1],
+    )
+
+
+CORPUS_DOCS = [
+    _doc("test://anarchism", (
+        "Anarchism is a political philosophy that promotes a stateless society. "
+        * 6
+        + "It seeks to abolish authority. " * 6
+    ), title="Anarchism"),
+    _doc("test://capital", (
+        "The eight forms of capital include living, social, intellectual. " * 6
+    ), title="Capital"),
+]
+
+
+@pytest.fixture
+def shard_conn(tmp_path):
+    db = tmp_path / "corpus.db"
+    c = connect(db)
+    ingest_source(c, _FakeSource(CORPUS_DOCS))
+    yield c
+    c.close()
+
+
+# ---------------------------------------------------------------------------
+# Data-class invariants
+# ---------------------------------------------------------------------------
+
+
+def test_hit_is_frozen_and_equals_by_value():
+    h1 = Hit("root1", "uri1", "title1", 1.5)
+    h2 = Hit("root1", "uri1", "title1", 1.5)
+    with pytest.raises(Exception):
+        h1.score = 2.0          # type: ignore[misc]
+    assert h1 == h2             # value equality
+    # extras carries a dict so Hit isn't hashable by design — callers
+    # dedup by (document_root, ...) tuples, not by put-in-a-set.
+
+
+def test_chunk_row_is_frozen():
+    c = ChunkRow("root", 0, "hex", "content")
+    with pytest.raises(Exception):
+        c.idx = 1                # type: ignore[misc]
+
+
+def test_hit_extras_defaults_empty():
+    h = Hit("a", "b", "c", 0.0)
+    assert h.extras == {}
+
+
+# ---------------------------------------------------------------------------
+# Protocol conformance
+# ---------------------------------------------------------------------------
+
+
+def test_sqlite_shard_corpus_satisfies_protocol(shard_conn):
+    corpus = SqliteShardCorpus(shard_conn)
+    assert isinstance(corpus, Corpus)
+    assert corpus.name == "sqlite-shard"
+    assert corpus.higher_is_better is False
+
+
+def test_sidecar_bucket_corpus_satisfies_protocol():
+    # Doesn't need a live bucket — runtime_checkable Protocol only
+    # checks method presence, not call signatures.
+    class _StubMulti:
+        manifest = type("M", (), {"shards": [], "snapshot_root": None})()
+        def fts_search(self, q, *, limit=8): return []
+        def conn_for_shard(self, url): raise KeyError(url)
+    corpus = SidecarBucketCorpus(_StubMulti())
+    assert isinstance(corpus, Corpus)
+    assert corpus.name == "sidecar-bucket"
+    assert corpus.higher_is_better is True
+
+
+# ---------------------------------------------------------------------------
+# fts_body sanitization (no FTS5-syntax leaks)
+# ---------------------------------------------------------------------------
+
+
+def test_sqlite_fts_body_handles_natural_language(shard_conn):
+    """Punctuation like '?' and '-' would crash raw FTS5. The adapter
+    must sanitize via _to_fts5 (mirrors cloud sanitizer)."""
+    corpus = SqliteShardCorpus(shard_conn)
+    # Question mark + apostrophe must not crash.
+    hits = corpus.fts_body("what is anarchism?", limit=5)
+    assert any("Anarchism" in h.title for h in hits)
+
+
+def test_sqlite_fts_body_stopword_only_query_returns_empty(shard_conn):
+    """All-stopword queries reduce to empty FTS5 expressions; the
+    adapter must short-circuit to [] rather than passing '' to MATCH."""
+    corpus = SqliteShardCorpus(shard_conn)
+    hits = corpus.fts_body("the a an of with", limit=5)
+    assert hits == []
+
+
+def test_sqlite_fts_body_returns_hit_shape(shard_conn):
+    corpus = SqliteShardCorpus(shard_conn)
+    hits = corpus.fts_body("anarchism", limit=2)
+    assert len(hits) >= 1
+    for h in hits:
+        assert isinstance(h, Hit)
+        assert h.document_root and h.document_uri and h.title
+        assert isinstance(h.score, float)
+
+
+# ---------------------------------------------------------------------------
+# Routes that should NotSupportedError
+# ---------------------------------------------------------------------------
+
+
+def test_sqlite_fts_title_not_supported(shard_conn):
+    corpus = SqliteShardCorpus(shard_conn)
+    with pytest.raises(NotSupportedError):
+        corpus.fts_title("anarchism")
+
+
+def test_sqlite_fts_phrase_not_supported(shard_conn):
+    corpus = SqliteShardCorpus(shard_conn)
+    with pytest.raises(NotSupportedError):
+        corpus.fts_phrase(["anarchism is a"])
+
+
+def test_sidecar_fts_title_not_supported():
+    class _StubMulti:
+        manifest = type("M", (), {"shards": [], "snapshot_root": None})()
+        def fts_search(self, q, *, limit=8): return []
+        def conn_for_shard(self, url): raise KeyError(url)
+    corpus = SidecarBucketCorpus(_StubMulti())
+    with pytest.raises(NotSupportedError):
+        corpus.fts_title("anarchism")
+
+
+def test_sidecar_fts_phrase_not_supported():
+    class _StubMulti:
+        manifest = type("M", (), {"shards": [], "snapshot_root": None})()
+        def fts_search(self, q, *, limit=8): return []
+        def conn_for_shard(self, url): raise KeyError(url)
+    corpus = SidecarBucketCorpus(_StubMulti())
+    with pytest.raises(NotSupportedError):
+        corpus.fts_phrase(["a b c"])
+
+
+# ---------------------------------------------------------------------------
+# chunks_for_doc shape + ordering
+# ---------------------------------------------------------------------------
+
+
+def test_sqlite_chunks_for_doc_returns_idx_ascending(shard_conn):
+    corpus = SqliteShardCorpus(shard_conn)
+    hits = corpus.fts_body("anarchism", limit=1)
+    assert hits
+    chunks = corpus.chunks_for_doc(hits[0].document_root)
+    assert len(chunks) >= 1
+    indices = [c.idx for c in chunks]
+    assert indices == sorted(indices)
+    for c in chunks:
+        assert isinstance(c, ChunkRow)
+        assert c.content       # decompressed, non-empty
+        assert c.leaf_hash      # ingest sets this
+
+
+def test_sqlite_snapshot_root_hex(shard_conn):
+    corpus = SqliteShardCorpus(shard_conn)
+    root = corpus.snapshot_root()
+    assert isinstance(root, str)
+    assert len(root) == 64
+    int(root, 16)   # well-formed hex

tests/test_qa_corpus_functional.py

@@ -0,0 +1,265 @@
+"""Functional tests — full retrieval → context → LLM → verify, both adapters.
+
+Builds the smallest plausible "ask-shaped" pipeline that the future
+Corpus-aware query() refactor will collapse into:
+
+    hits   = corpus.fts_body(question, limit=k)
+    chunks = []
+    for h in hits:
+        chunks.extend(corpus.chunks_for_doc(h.document_root, limit=1))
+    evidence_map = build_evidence_map([...chunks-as-evidence-dicts...])
+    prompt = system + EVIDENCE + question + reminder
+    answer = chat.chat_completion(...)
+    verdict = verify_claim_lattice(answer, evidence_map)
+
+This module drives that pipeline through SqliteShardCorpus AND
+SidecarBucketCorpus with a StubClient (deterministic answer) and
+asserts both reach the same audit_mode + cite the same primary
+source_root.
+
+It's the SMALLEST canary for the "DRY the cloud + local diff" work:
+when the refactor moves query() onto the protocol, this test gates
+both adapters in one go.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import sqlite3
+import threading
+from http.server import ThreadingHTTPServer
+
+from arborist.wallet._range_http_server import RangeHandler as _RangeHandler
+from pathlib import Path
+from typing import Iterator
+
+import pytest
+
+apsw = pytest.importorskip("apsw")
+
+from arborist.document import Document
+from arborist.ingest import ingest_source
+from arborist.qa.client import StubClient
+from arborist.qa.corpus import SidecarBucketCorpus, SqliteShardCorpus
+from arborist.qa.evidence import (
+    build_evidence_map,
+    render_evidence_map,
+)
+from arborist.qa.prompts import (
+    CLAIM_LATTICE_GROUNDING_REMINDER,
+    CLAIM_LATTICE_SYSTEM_PROMPT,
+)
+from arborist.qa.verify import verify_claim_lattice
+from arborist.source import Source
+from arborist.store import connect
+from arborist.wallet.bucket import BucketManifest, MultiShardSidecarCorpus
+from arborist.wallet.sidecar import build_sidecar
+
+
+# ---------------------------------------------------------------------------
+# Fixture: same corpus exposed via both adapters.
+# ---------------------------------------------------------------------------
+
+
+CORPUS_DOCS = [
+    Document(
+        uri="test://anarchism", source_type="test", title="Anarchism",
+        content=(
+            "Anarchism is a political philosophy that promotes a stateless "
+            "society. " * 6
+            + 'The phrase "abolition of authority" describes the core principle '
+              'of anarchism. ' * 4
+        ),
+    ),
+    Document(
+        uri="test://capital", source_type="test", title="Capital",
+        content=(
+            "The eight forms of capital include living, social, intellectual, "
+            "material, financial, experiential, cultural, and spiritual. " * 6
+        ),
+    ),
+    Document(
+        uri="test://noise-1", source_type="test", title="Random Noise",
+        content=("Lorem ipsum dolor sit amet. " * 12),
+    ),
+]
+
+
+class _FakeSource(Source):
+    source_type = "test"
+    def __init__(self, ds): self.ds = ds
+    def iter_documents(self) -> Iterator[Document]:
+        yield from self.ds
+
+
+@pytest.fixture
+def both_adapters(tmp_path: Path):
+    """Yield (sqlite_corpus, sidecar_corpus) over the same content."""
+    db = tmp_path / "shard.db"
+    c0 = connect(db)
+    ingest_source(c0, _FakeSource(CORPUS_DOCS))
+    c0.execute("PRAGMA wal_checkpoint(FULL)")
+    c0.close()
+
+    # bucket layout + sidecar
+    bucket = tmp_path / "bucket"
+    clones = bucket / "clones" / "snap-1"
+    clones.mkdir(parents=True)
+    bucket_shard = clones / "000.db"
+    shutil.copy(db, bucket_shard)
+    sidecar_path = clones / "000.sidecar.bin"
+    build_sidecar(bucket_shard, sidecar_path)
+
+    # in-process Range-aware server
+    cwd_before = os.getcwd()
+    os.chdir(bucket)
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), _RangeHandler)
+    port = httpd.server_address[1]
+    threading.Thread(target=httpd.serve_forever, daemon=True).start()
+    base_url = f"http://127.0.0.1:{port}"
+
+    manifest = BucketManifest(
+        shards=[{
+            "url": f"{base_url}/clones/snap-1/000.db",
+            "shard_idx": 0,
+            "label": "test-shard",
+            "sidecar_url": f"{base_url}/clones/snap-1/000.sidecar.bin",
+        }],
+        blob_base=f"{base_url}/blobs",
+        snapshot_root=None,
+        snapshot_ts=None,
+    )
+    multi = MultiShardSidecarCorpus(
+        manifest, sidecar_cache_dir=str(tmp_path / "cache"),
+    )
+    conn = connect(db)
+
+    try:
+        yield SqliteShardCorpus(conn), SidecarBucketCorpus(multi)
+    finally:
+        multi.close()
+        conn.close()
+        httpd.shutdown()
+        os.chdir(cwd_before)
+
+
+# ---------------------------------------------------------------------------
+# The shared end-to-end harness — the prototype of what query() will do
+# once it takes a Corpus.
+# ---------------------------------------------------------------------------
+
+
+def _retrieve_and_verify(
+    corpus, question: str, *, stub_answer: str, top_k: int = 3,
+):
+    """Mini retrieval pipeline through the Corpus protocol. Returns a
+    dict with audit_mode + cited document_roots + verdict shape."""
+    hits = corpus.fts_body(question, limit=top_k)
+    if not hits:
+        return {"audit_mode": "UNGROUNDED", "n_quotes": 0, "n_verified": 0,
+                "sources": [], "cited_roots": set()}
+
+    chunks_payload: list[dict] = []
+    for rank, h in enumerate(hits, 1):
+        rows = corpus.chunks_for_doc(h.document_root, limit=1)
+        if not rows:
+            continue
+        r = rows[0]
+        chunks_payload.append({
+            "source_root": h.document_root,
+            "document_uri": h.document_uri,
+            "title": h.title,
+            "chunk_idx": r.idx,
+            "chunk_root": r.leaf_hash or "0" * 64,
+            "span": r.content[:1200],
+            "source_role": (
+                "primary_answer_source" if rank == 1 else "background_source"
+            ),
+        })
+
+    evidence_map = build_evidence_map(chunks_payload)
+    evidence_text = render_evidence_map(evidence_map)
+    messages = [
+        {"role": "system", "content": CLAIM_LATTICE_SYSTEM_PROMPT},
+        {"role": "user", "content": (
+            f"EVIDENCE:\n\n{evidence_text}\n\n"
+            f"QUESTION: {question}\n\n"
+            f"{CLAIM_LATTICE_GROUNDING_REMINDER}"
+        )},
+    ]
+    answer = StubClient(answer=stub_answer).chat_completion(
+        messages, model="stub", temperature=0.0,
+    )
+    verdict = verify_claim_lattice(
+        answer, evidence_map, question=question, warrant_check_enabled=False,
+    )
+
+    cited_roots = {h.document_root for h in hits}
+    return {
+        "audit_mode": verdict["audit_mode"],
+        "n_quotes": verdict["n_quotes"],
+        "n_verified": verdict["n_verified"],
+        "cited_roots": cited_roots,
+        "primary_root": hits[0].document_root if hits else "",
+        "primary_title": hits[0].title if hits else "",
+    }
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+def test_pipeline_grounds_anarchism_through_both_adapters(both_adapters):
+    """Verbatim-quote answer should land at the same audit_mode on both
+    adapters and primary-cite the Anarchism doc on both."""
+    sq, sc = both_adapters
+    question = "what is the core principle of anarchism?"
+    # Verbatim from CORPUS_DOCS[0]: 'abolition of authority'
+    stub = (
+        'The core principle of anarchism is the "abolition of authority". [E1]'
+    )
+    sq_result = _retrieve_and_verify(sq, question, stub_answer=stub)
+    sc_result = _retrieve_and_verify(sc, question, stub_answer=stub)
+    assert sq_result["audit_mode"] == sc_result["audit_mode"], (
+        f"audit_mode differs: sqlite={sq_result['audit_mode']} "
+        f"sidecar={sc_result['audit_mode']}"
+    )
+    # Both must surface the Anarchism doc as primary.
+    assert "Anarchism" in sq_result["primary_title"]
+    assert "Anarchism" in sc_result["primary_title"]
+
+
+def test_pipeline_handles_noise_question_consistently(both_adapters):
+    """An off-topic question (no matching doc) must produce UNGROUNDED
+    / empty-evidence on both adapters."""
+    sq, sc = both_adapters
+    question = "what is the population of mars?"
+    stub = "Mars has no population because it is uninhabited. [E1]"
+    sq_result = _retrieve_and_verify(sq, question, stub_answer=stub)
+    sc_result = _retrieve_and_verify(sc, question, stub_answer=stub)
+    # Off-topic should land at UNGROUNDED or HYBRID on BOTH paths —
+    # neither corpus has Mars content. STRICT would be a false positive.
+    for r in (sq_result, sc_result):
+        assert r["audit_mode"] in ("UNGROUNDED", "HYBRID"), (
+            f"unexpected STRICT on off-topic: {r}"
+        )
+
+
+def test_pipeline_chunks_for_primary_are_same_bytes(both_adapters):
+    """Both adapters must surface byte-identical chunk content for the
+    same document_root — wallet-side Merkle verification depends on it."""
+    sq, sc = both_adapters
+    question = "anarchism"
+    sq_hits = sq.fts_body(question, limit=1)
+    assert sq_hits
+    droot = sq_hits[0].document_root
+
+    sq_chunks = sq.chunks_for_doc(droot)
+    sc_chunks = sc.chunks_for_doc(droot)
+    assert sq_chunks and sc_chunks
+    for a, b in zip(sq_chunks, sc_chunks):
+        assert a.content == b.content, (
+            f"chunk[{a.idx}] content differs by adapter"
+        )
+        assert a.leaf_hash == b.leaf_hash

tests/test_qa_corpus_integration.py

@@ -0,0 +1,194 @@
+"""Integration tests for Corpus adapters — real corpus → both backends.
+
+Builds a small SQLite shard, runs the same fixtures through:
+  - SqliteShardCorpus directly
+  - SidecarBucketCorpus over a sidecar built from the same shard,
+    served via an in-process HTTP-Range server, opened via apsw VFS
+
+Asserts the two adapters agree on the WINNING document_root for each
+fixture query. Score scales differ (FTS5 BM25 is negative, sidecar
+BM25 + boost is positive), so we compare TOP-K SETS, not raw scores —
+the contract is "both backends point at the same primary source for
+the same query."
+
+What this catches that unit tests don't:
+  * Token/stem/fold mismatches between FTS5's unicode61 tokenizer and
+    the sidecar's NFKD-fold + ASCII-word path
+  * Sidecar build-time term loss vs FTS5's full-text body coverage
+  * apsw HttpRangeVFS read-path correctness on real SQLite files
+
+Uses MockBucketBackend-style in-process serving (same trick as
+tests/test_wallet_bucket.py) so the suite stays offline.
+"""
+from __future__ import annotations
+
+import os
+import sqlite3
+import threading
+from http.server import ThreadingHTTPServer
+from pathlib import Path
+from typing import Iterator
+
+import pytest
+
+apsw = pytest.importorskip("apsw")
+
+from arborist.document import Document
+from arborist.ingest import ingest_source
+from arborist.qa.corpus import (
+    SidecarBucketCorpus,
+    SqliteShardCorpus,
+)
+from arborist.source import Source
+from arborist.store import connect
+from arborist.wallet.bucket import (
+    BucketManifest,
+    MultiShardSidecarCorpus,
+)
+from arborist.wallet.sidecar import build_sidecar
+
+
+# ---------------------------------------------------------------------------
+# Range-aware HTTP server (mirrors tests/test_wallet_bucket.py)
+# ---------------------------------------------------------------------------
+
+
+from arborist.wallet._range_http_server import RangeHandler as _RangeHandler
+
+
+@pytest.fixture
+def served_corpus(tmp_path: Path):
+    """Build a 4-doc corpus, sidecar it, expose via HTTP. Yields:
+      (sqlite_conn, multi_shard_corpus, base_url)
+    """
+    class _S(Source):
+        source_type = "test"
+        def __init__(self, ds): self.ds = ds
+        def iter_documents(self) -> Iterator[Document]:
+            yield from self.ds
+
+    db = tmp_path / "shard.db"
+    conn0 = connect(db)
+    ingest_source(conn0, _S([
+        Document(uri=f"test://doc-{i}", source_type="test",
+                 title=f"Doc {i}",
+                 content=(
+                     f"Topic alpha-{i} is described in many sentences here. " * 6
+                     + f"The unique phrase for entry {i} is 'lemma-{i}'. " * 6
+                 ))
+        for i in range(4)
+    ]))
+    # Checkpoint WAL + close so the on-disk file is self-contained
+    # before we copy it into the bucket layout.
+    conn0.execute("PRAGMA wal_checkpoint(FULL)")
+    conn0.close()
+
+    # Build sidecar from a copy of the shard placed in the bucket layout.
+    bucket = tmp_path / "bucket"
+    clones = bucket / "clones" / "snap-1"
+    clones.mkdir(parents=True)
+    import shutil
+    bucket_shard = clones / "000.db"
+    shutil.copy(db, bucket_shard)
+    sidecar_path = clones / "000.sidecar.bin"
+    build_sidecar(bucket_shard, sidecar_path)
+
+    # Re-open the original for SqliteShardCorpus tests.
+    conn = connect(db)
+
+    # Serve bucket dir.
+    cwd_before = os.getcwd()
+    os.chdir(bucket)
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), _RangeHandler)
+    port = httpd.server_address[1]
+    t = threading.Thread(target=httpd.serve_forever, daemon=True)
+    t.start()
+    base_url = f"http://127.0.0.1:{port}"
+
+    manifest = BucketManifest(
+        shards=[{
+            "url": f"{base_url}/clones/snap-1/000.db",
+            "shard_idx": 0,
+            "label": "test-shard",
+            "sidecar_url": f"{base_url}/clones/snap-1/000.sidecar.bin",
+        }],
+        blob_base=f"{base_url}/blobs",
+        snapshot_root=None,
+        snapshot_ts=None,
+    )
+    multi = MultiShardSidecarCorpus(manifest, sidecar_cache_dir=str(tmp_path / "cache"))
+
+    try:
+        yield conn, multi, base_url
+    finally:
+        multi.close()
+        conn.close()
+        httpd.shutdown()
+        os.chdir(cwd_before)
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+def _top_doc_root(hits) -> str:
+    return hits[0].document_root if hits else ""
+
+
+def test_adapters_both_surface_target_doc_in_top_k(served_corpus):
+    """A token unique to one doc must appear in BOTH adapters' top-K.
+
+    Note: which doc ranks #1 can differ — sidecar applies a title-boost
+    that SqliteShardCorpus.fts_body doesn't (its title-LIKE route lands
+    in the next protocol refactor). The contract here is recall, not
+    rank: both backends must SEE the right doc in the top results.
+    """
+    conn, multi, _ = served_corpus
+    sqlite_corpus = SqliteShardCorpus(conn)
+    sidecar_corpus = SidecarBucketCorpus(multi)
+
+    # 'lemma 3' has 'lemma' in every doc but '3' is rarer (Doc 3 title +
+    # body 'lemma-3'); a sane retrieval surfaces Doc 3 in the top-K of
+    # both backends.
+    sq_hits = sqlite_corpus.fts_body("lemma 3", limit=4)
+    sc_hits = sidecar_corpus.fts_body("lemma 3", limit=4)
+    assert sq_hits, "sqlite adapter returned no hits"
+    assert sc_hits, "sidecar adapter returned no hits"
+
+    sq_titles = {h.title for h in sq_hits}
+    sc_titles = {h.title for h in sc_hits}
+    assert "Doc 3" in sq_titles, f"sqlite top-K missing Doc 3: {sq_titles}"
+    assert "Doc 3" in sc_titles, f"sidecar top-K missing Doc 3: {sc_titles}"
+
+
+def test_adapters_chunks_for_doc_decode_equivalently(served_corpus):
+    """Same document_root → both adapters return decoded chunk text that
+    matches byte-for-byte (modulo any unpack_chunk normalization)."""
+    conn, multi, _ = served_corpus
+    sq = SqliteShardCorpus(conn)
+    sc = SidecarBucketCorpus(multi)
+
+    sq_hits = sq.fts_body("lemma 3", limit=1)
+    assert sq_hits
+    droot = sq_hits[0].document_root
+
+    sq_chunks = sq.chunks_for_doc(droot)
+    sc_chunks = sc.chunks_for_doc(droot)
+    assert len(sq_chunks) == len(sc_chunks), (
+        f"chunk count mismatch: sqlite={len(sq_chunks)} sidecar={len(sc_chunks)}"
+    )
+    for a, b in zip(sq_chunks, sc_chunks):
+        assert a.idx == b.idx
+        assert a.leaf_hash == b.leaf_hash
+        assert a.content == b.content
+
+
+def test_sqlite_snapshot_root_matches_compute_snapshot_root(served_corpus):
+    """Adapter-reported snapshot_root must match compute_snapshot_root()
+    on the same connection — the wallet relies on this for binding."""
+    from arborist.snapshot import compute_snapshot_root
+    conn, _, _ = served_corpus
+    sq = SqliteShardCorpus(conn)
+    expected, _ = compute_snapshot_root(conn)
+    assert sq.snapshot_root() == expected