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]: