qa/corpus: add core_keyword_match + doc_body to Corpus protocol

Phase 1 step 4 of #53. Two new protocol methods that the unified
run_query needs but the protocol didn't expose:

- core_keyword_match(qtokens, *, limit) → list[Hit]
  TF-IDF core-keyword route. Finds source docs whose distilled
  tfidf-core content contains any of qtokens. Closes the neologism
  gap (e.g. "permacomputer" matching a Grok conversation about it
  via its TF-IDF core, even though permacomputer never appears in
  a title). Returns Hits whose .score is the integer match_count
  (also in .extras["match_count"]) — UNIQUE among routes in being
  higher-is-better, not bm25 lower-is-better.

- doc_body(document_root) → str | None
  Concatenated chunk text for one document. Used by the body-
  coverage rerank stage that needs the full body (not just top-K
  chunks) to decide whether the doc actually discusses qtokens.

Implementations:
- SqliteShardCorpus: full SQL, lifted verbatim from query.py's
  _docs_with_core_keyword_match (same word-boundary LIKE +
  match_count tallying). Gracefully returns [] if derivations
  table missing (older shard).
- MultiShardSqliteCorpus: per-shard fan-out, dedupe by doc_root
  keeping MAX match_count across shards (HIGHER wins for this
  route).
- SidecarBucketCorpus: core_keyword_match raises NotSupportedError
  (the derivations table that maps tfidf-core → source isn't in
  the slim FTS5 sidecar). doc_body concatenates chunks_for_doc
  output (works on bucket via blob fallback).

Smoke: on the genesis wikipedia shards core_keyword_match returns
[] cleanly — those shards have 92 claim_pack derivations but 0
tfidf-core derivations, so the route correctly produces nothing.

Validation: 257 tests in the query/corpus/sidecar/wallet/bucket/
claim_lattice gate pass.

arborist/qa/corpus.py

@@ -213,6 +213,21 @@ class Corpus(Protocol):
         index."""
         ...
 
+    def core_keyword_match(
+        self, qtokens: list[str], *, limit: int = 32
+    ) -> list[Hit]:
+        """TF-IDF core-keyword route: find SOURCE docs whose distilled
+        TF-IDF cores contain any of ``qtokens``. Closes the neologism
+        gap for terms that never make it into a title but ARE the
+        distinctive low-frequency vocabulary of a doc (e.g.
+        ``permacomputer`` matching a Grok conversation about it).
+        Hits carry the per-doc ``match_count`` in ``extras``.
+
+        SidecarBucketCorpus raises NotSupportedError — the
+        derivations table that holds the core↔source link isn't
+        published to the cloud manifest today."""
+        ...
+
     # --- evidence reads ---
 
     def chunks_for_doc(
@@ -221,6 +236,16 @@ class Corpus(Protocol):
         """Per-doc chunk rows in idx-ascending order."""
         ...
 
+    def doc_body(self, document_root: str) -> str | None:
+        """Concatenated chunk text for one document.
+
+        Used by the body-coverage rerank stage that needs the full
+        body (not just the top-K chunks) to decide whether the doc
+        actually discusses the query tokens. Returns None if the
+        document isn't present, empty string if all chunks unpack
+        empty."""
+        ...
+
     # --- corpus identity ---
 
     def snapshot_root(self) -> str:
@@ -347,6 +372,86 @@ class SqliteShardCorpus:
             for r in rows
         ]
 
+    def core_keyword_match(
+        self, qtokens: list[str], *, limit: int = 32
+    ) -> list[Hit]:
+        """TF-IDF core-keyword route via the derivations table.
+
+        SQL is the same as the legacy ``_docs_with_core_keyword_match``
+        in query.py. Word-boundary LIKE against the comma-separated
+        TF-IDF keyword string (prepends/appends ", " so a single LIKE
+        pattern ``%, token, %`` checks position-agnostic without
+        false-positives from substring noise like ``intelligence``
+        matching ``intel``).
+
+        Returns Hits whose ``score`` is the integer match_count
+        (number of qtokens that hit the doc's core); also surfaced in
+        ``extras["match_count"]`` for explicit callers. Higher
+        match_count = stronger signal — caller decides how to fold it
+        into rank.
+        """
+        if not qtokens:
+            return []
+        case_clauses = " + ".join(
+            ["(CASE WHEN LOWER(', ' || c.content || ', ') LIKE ? THEN 1 ELSE 0 END)"]
+            * len(qtokens)
+        )
+        where_clauses = " OR ".join(
+            ["LOWER(', ' || c.content || ', ') LIKE ?"] * len(qtokens)
+        )
+        patterns = [f"%, {t.lower()}, %" for t in qtokens]
+        params = patterns + patterns + [limit]
+        sql = (
+            "SELECT "
+            "    src.document_root, src.document_uri, src.title, "
+            f"   MAX({case_clauses}) AS match_count "
+            "FROM chunks c "
+            "JOIN documents core ON core.document_root = c.document_root "
+            "JOIN derivations der ON der.core_root = core.document_root "
+            "JOIN documents src ON src.document_root = der.src_root "
+            "WHERE core.source_type LIKE 'core:tfidf-%' "
+            f"  AND ({where_clauses}) "
+            "GROUP BY src.document_root, src.document_uri, src.title "
+            "ORDER BY match_count DESC LIMIT ?"
+        )
+        try:
+            rows = list(self._conn.execute(sql, params))
+        except Exception:
+            # Shard lacks derivations table (older schema) — no hits.
+            return []
+        return [
+            Hit(
+                document_root=r["document_root"],
+                document_uri=r["document_uri"] or "",
+                title=r["title"] or "",
+                score=float(r["match_count"]),
+                shard_id=None,
+                extras={"match_count": int(r["match_count"])},
+            )
+            for r in rows
+        ]
+
+    def doc_body(self, document_root: str) -> str | None:
+        """Concatenated chunk text for one document. Returns None if
+        the document isn't present in this shard."""
+        from arborist.compress import unpack_chunk
+        rows = list(self._conn.execute(
+            "SELECT content FROM chunks "
+            "WHERE document_root = ? AND content IS NOT NULL "
+            "  AND length(content) > 0 "
+            "  AND substr(content, 1, 1) != X'00' "
+            "ORDER BY idx ASC",
+            (document_root,),
+        ))
+        if not rows:
+            return None
+        parts: list[str] = []
+        for r in rows:
+            text = unpack_chunk(r["content"]) or ""
+            if text:
+                parts.append(text)
+        return " ".join(parts)
+
     def chunks_for_doc(
         self, document_root: str, *, limit: int | None = None
     ) -> list[ChunkRow]:
@@ -474,6 +579,41 @@ class MultiShardSqliteCorpus:
         ngs = list(ngrams)
         return self._fanout("fts_phrase", limit, ngrams=ngs)
 
+    def core_keyword_match(
+        self, qtokens: list[str], *, limit: int = 32
+    ) -> list[Hit]:
+        """Per-shard core_keyword_match merged by raw match_count
+        (DESC — higher is better, unlike bm25 routes)."""
+        merged: list[Hit] = []
+        for sh_path, sc in self._shards:
+            for h in sc.core_keyword_match(qtokens, limit=limit):
+                merged.append(Hit(
+                    document_root=h.document_root,
+                    document_uri=h.document_uri,
+                    title=h.title,
+                    score=h.score,
+                    shard_id=sh_path,
+                    extras=dict(h.extras),
+                ))
+        # Higher match_count wins (the only route in the protocol where
+        # higher-is-better — bm25 routes go MIN). Dedupe by doc_root,
+        # keep max match_count across shards.
+        best: dict[str, Hit] = {}
+        for h in merged:
+            cur = best.get(h.document_root)
+            if cur is None or h.score > cur.score:
+                best[h.document_root] = h
+        ranked = sorted(best.values(), key=lambda h: -h.score)
+        return ranked[:limit]
+
+    def doc_body(self, document_root: str) -> str | None:
+        """Walk each shard until one returns body text."""
+        for _, sc in self._shards:
+            body = sc.doc_body(document_root)
+            if body is not None:
+                return body
+        return None
+
     def _fanout(self, method: str, limit: int, **kwargs) -> list[Hit]:
         """Common per-shard fan-out + merge for title/phrase routes.
 
@@ -598,6 +738,26 @@ class SidecarBucketCorpus:
             for h in hits
         ]
 
+    def core_keyword_match(
+        self, qtokens: list[str], *, limit: int = 32
+    ) -> list[Hit]:
+        raise NotSupportedError(
+            "core_keyword_match needs the derivations table that maps "
+            "tfidf-core docs back to source docs. The cloud bucket "
+            "manifest doesn't publish derivations today — only the slim "
+            "FTS5 sidecar (documents + chunks + FTS5 shadows). The route "
+            "stays local-only until a slim-derivations sidecar ships."
+        )
+
+    def doc_body(self, document_root: str) -> str | None:
+        """Concat chunks fetched via chunks_for_doc — content comes from
+        bucket blobs/<hash> or the big-shard fallback per shard
+        client's FtsSidecarShardClient.fetch_chunk_body policy."""
+        chunks = self.chunks_for_doc(document_root)
+        if not chunks:
+            return None
+        return " ".join(c.content for c in chunks if c.content)
+
     def chunks_for_doc(
         self, document_root: str, *, limit: int | None = None
     ) -> list[ChunkRow]: