From e322bbd9ddfd02ca96187321cc8789180190e3dd Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sun, 31 May 2026 12:24:03 -0400 Subject: [PATCH] qa/corpus: add core_keyword_match + doc_body to Corpus protocol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 160 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 160 insertions(+) diff --git a/arborist/qa/corpus.py b/arborist/qa/corpus.py index f60ab13..9b2e700 100644 --- a/arborist/qa/corpus.py +++ b/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/ 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]: