SqliteShardCorpus.fts_body: delegate to FTS5Backend (progressive-AND)

Fox flagged 2026-06-01: "search is taking 45 secs when it used to
search across all 4 shards in like 2-5 secs before." Confirmed
search.done=48653ms on "when did the muppet show first air on
television?" — 9-token natural-language query.

Root cause: SqliteShardCorpus.fts_body used a naive `MATCH 'tok1
OR tok2 OR ...'` over the FTS5 chunks index. For multi-token
queries where ANY token has high document-frequency, BM25 has to
score every chunk in the OR-union. "muppet show first air
television" pulls ~300k matches on a 1.5M-chunk shard; BM25 ranks
all of them; ~30s cold I/O per shard × 5 shards = 150s wall time
before the parallel fan-out, ~45-50s after. Legacy already solved
this in arborist.search.fts5.FTS5Backend.search — progressive-AND
mode (intersect posting lists, fast), retries with shortest-token
dropped on zero, and only falls back to OR-mode with a high-DF
filter when AND exhausts. The comment literally says "~27s cold
I/O" for the OR-mode pathology.

Fix: SqliteShardCorpus.fts_body now delegates to FTS5Backend
instead of building its own OR-mode SQL. Same shard, same FTS5
index — just the right query construction.

All-stopword short-circuit preserved (FTS5Backend has a sentinel
``""`` fallback that matches arbitrary docs; pre-check via
_to_fts5() guards against that).

Measured (live `make query LLM=qwen` on the same fixture):

  search:    48.7 s  →   0.7 s   (70× speedup)
  total miss: 50 s   →   2.9 s
  total hit:  ~50 s  →   0.95 s

This is on fox's actual workload via Qwen. All 79 corpus/providence
tests pass.

Cloud/Sidecar path unchanged — slim FTS5 sidecars don't have the
high-DF posting-list issue at scale (their corpora are smaller) and
the cloud apsw connection wouldn't benefit from progressive-AND the
same way. Future work: same FTS5Backend pattern for
SidecarBucketCorpus if a similar slowdown surfaces.

arborist/qa/corpus.py

@@ -296,31 +296,43 @@ class SqliteShardCorpus:
         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).
+        """Body-FTS5 via the existing FTS5Backend (progressive-AND
+        with OR-mode fallback).
+
+        Naive ``MATCH 'tok1 OR tok2 OR ...'`` over a multi-token
+        natural-language query is catastrophic on a 1.5M-chunk shard
+        when ANY token has high document-frequency: BM25 has to score
+        every match in the OR-union, and "muppet show first air
+        television" pulls ~300k matches in ~30s per shard (×5 shards
+        = ~150s wall time). FTS5Backend.search runs progressive-AND
+        first (intersecting posting lists, fast), retries with
+        shortest-token-dropped on zero results, and only falls back
+        to OR mode when AND exhausts — clipping the high-DF tail of
+        tokens that would otherwise dominate.
+
+        Returns the protocol-shaped Hit; FTS5Backend's own Hit lives
+        in arborist.search.base.
+        """
+        # Short-circuit all-stopword queries — FTS5Backend would
+        # fall back to a sentinel ``""`` token that matches arbitrary
+        # docs; we want pure [].
         from arborist.wallet.bucket import _to_fts5
-        match_expr = _to_fts5(query)
-        if not match_expr.strip():
+        if not _to_fts5(query).strip():
             return []
-        rows = list(self._conn.execute(sql, (match_expr, limit)))
+        from arborist.search.fts5 import FTS5Backend
+        backend = FTS5Backend(self._conn)
+        # FTS5Backend.search returns objects with document_root,
+        # document_uri, title, chunk_idx, score, snippet.
+        raw = backend.search(query, limit=limit)
         return [
             Hit(
-                document_root=r["document_root"],
-                document_uri=r["document_uri"] or "",
-                title=r["title"] or "",
-                score=r["score"],
+                document_root=h.document_root,
+                document_uri=h.document_uri or "",
+                title=h.title or "",
+                score=h.score,
                 shard_id=None,
             )
-            for r in rows
+            for h in raw
         ]
 
     def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]: