From d9fb6a9b693fc76aa772c8927a1fdeea688c0e51 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sun, 31 May 2026 10:43:03 -0400 Subject: [PATCH] wallet/fts-sidecar: real FTS5 in the cloud sidecar; delete custom BM25 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the hand-rolled BM25 sidecar (arborist/wallet/sidecar.py, ~690 LOC) with a slim SQLite file that just COPIES the source shard's FTS5 shadow tables verbatim + minimal doc/chunk metadata. Cloud retrieval then runs SQLite FTS5 bm25() on the same bytes the local shard uses — bit-for-bit parity by construction. 5/5 source + audit_mode agreement on the smoke fixture between local corpus-query and cloud-query against the new manifest-fts.json. Why --- Custom binary sidecar was per-document BM25; main encyclopedia articles got length-normalized so hard that on "why did the dinosaurs go extinct?" "Edwina, the Dinosaur Who Didn't Know She Was Extinct" beat "Dinosaur" (measured cloud-vs-local divergence). Local FTS5 indexes per-chunk so each chunk is a moderate-length doc and the main article wins multiply. Different granularity, not a tuning knob — fix is to use the same indexer cloud-side. What ships ---------- - arborist/wallet/fts_sidecar_build.py — builder. ATTACH source shard, copy documents (root/uri/title only), copy chunks (id/root/idx/leaf only, NO content), CREATE VIRTUAL TABLE chunks_fts/documents_fts with same DDL as source, bulk-copy the four shadow tables verbatim, VACUUM. 8.78 GB shard → 2.15 GB sidecar (24.5%) in ~45 s; full 4-shard wiki corpus 37.4 GB → 8.1 GB (21.7%) in ~3 min. - arborist/wallet/bucket.py: FtsSidecarShardClient — downloads slim sidecar once into ~/.arborist/sidecar-fts-cache/.idx.db, opens read-only sqlite3 (check_same_thread=False for parallel shard fan- out), runs FTS5 MATCH locally. Chunk content fetches via blobs/ with HTTP-range big-shard fallback when blobs aren't published. MultiShardSidecarCorpus simplified to fts_sidecar_url ∨ bucket-direct (both are FTS5 backends; merge by raw bm25 MIN ascending). - arborist/qa/corpus.py: SidecarBucketCorpus.higher_is_better=False (FTS5 bm25 is negative, lower=better). chunks_for_doc dispatches on fetch_chunk_body attr for the slim-FTS5 client. apply_title_boost imports tokenizer helpers from new arborist/qa/_text_norm.py. - arborist/qa/_text_norm.py — fold_accents, numeral_expand, tokenize_text, STOPWORDS — extracted from the deleted sidecar.py so apply_title_boost keeps its lexical shape. - arborist/cli.py: `arborist sidecar build-fts` subcommand; old `sidecar build`/`sidecar search` removed. cloud_query recognizes fts_sidecar_url + sidecar_url alike. - Makefile: `sidecar-build-fts` + `sidecar-build-fts-all` targets; `sidecar-build` + `sidecar-search` removed. - scripts/upload_fts_sidecars.py — boto3 producer: uploads slim sidecars to clones/sidecars-fts/.idx.db, publishes clones/manifest-fts.json (4 wikipedia shards inherit existing shard_url for content fallback; ACL public-read; idempotent on size match). Existing manifest-sidecar.json untouched. - tests/test_qa_corpus_functional.py + test_qa_corpus_integration.py converted from build_sidecar → build_fts_sidecar; 6 fixtures pass. - bench/slim_fts_parity_bench.py — local 3-way bench (legacy/corpus/slim_fts) over the smoke fixture. Validation ---------- - Per-shard FTS5 parity: slim sidecar returns IDENTICAL rowids + bm25 scores to the source shard for top-10 of "dinosaurs extinct". - 3-way bench (legacy local / corpus local / slim-FTS5 over real bucket fallback): 5/5 source agreement AND 5/5 audit_mode agreement between corpus and slim_fts. Q5 legacy disagreement (Edwina vs Dinosaur) is the pre-existing 2000-line query() retrieval quirk, unrelated. - End-to-end cloud query against published manifest-fts.json (cold- start, ~149 s sidecar download once): STRICT · Dinosaur, every quote verified (2/2). - Full pytest suite: 2737 passed, 28 skipped, 1 xfailed. One pre- existing failure (tests/test_doc_counts.py — claim_pack docs row- count drift) and one pre-existing cold_object failure, both reproduce on main HEAD. Bucket state ------------ - s3://arborist/clones/sidecars-fts/00[0-3].idx.db (8.1 GB) — new - s3://arborist/clones/manifest-fts.json — new - s3://arborist/clones/manifest-sidecar.json — kept live (deprecated but still readable; downstream callers should switch to manifest-fts.json) --- Makefile | 10 - arborist/cli.py | 110 ++-- arborist/qa/_text_norm.py | 84 +++ arborist/qa/corpus.py | 85 ++- arborist/wallet/_sidecar_build_runner.py | 31 - arborist/wallet/bucket.py | 375 +++++++------ arborist/wallet/fts_sidecar_build.py | 209 +++++++ arborist/wallet/sidecar.py | 685 ----------------------- bench/slim_fts_parity_bench.py | 263 +++++++++ scripts/upload_fts_sidecars.py | 194 +++++++ tests/test_qa_corpus.py | 3 +- tests/test_qa_corpus_functional.py | 12 +- tests/test_qa_corpus_integration.py | 10 +- 13 files changed, 1087 insertions(+), 984 deletions(-) create mode 100644 arborist/qa/_text_norm.py delete mode 100644 arborist/wallet/_sidecar_build_runner.py create mode 100644 arborist/wallet/fts_sidecar_build.py delete mode 100644 arborist/wallet/sidecar.py create mode 100644 bench/slim_fts_parity_bench.py create mode 100644 scripts/upload_fts_sidecars.py diff --git a/Makefile b/Makefile index b11ed66..7590dac 100644 --- a/Makefile +++ b/Makefile @@ -1604,11 +1604,6 @@ cloud-snapshot-root: bootstrap ## compute snapshot_root of the bucket-resident s @echo "# shard: $(SHARD_URL)" >&2 @$(ARBORIST) cloud snapshot-root --shard-url "$(SHARD_URL)" --cache-mb $(or $(CACHE_MB),32) -sidecar-build: bootstrap ## build HTTP-optimized inverted-index sidecar [SHARD=path OUT=path] - @test -n "$(SHARD)" || { echo 'usage: make sidecar-build SHARD=/path/to/shard.db OUT=/path/to/sidecar.bin'; exit 2; } - @test -n "$(OUT)" || { echo 'OUT required'; exit 2; } - @$(ARBORIST) sidecar build --shard "$(SHARD)" --out "$(OUT)" - sidecar-build-fts: bootstrap ## build slim FTS5 sidecar (.idx.db) from a shard [SHARD=path OUT=path] @test -n "$(SHARD)" || { echo 'usage: make sidecar-build-fts SHARD=/path/to/shard.db OUT=/path/to/shard.idx.db'; exit 2; } @test -n "$(OUT)" || { echo 'OUT required'; exit 2; } @@ -1626,11 +1621,6 @@ sidecar-build-fts-all: bootstrap ## build slim FTS5 sidecar for every shard in S done @ls -lh "$$OUT_DIR"/*.idx.db 2>/dev/null || true -sidecar-search: bootstrap ## query a local sidecar.bin [Q="..." SIDECAR=/path/sidecar.bin LIMIT=N MODE=or|and] - @test -n "$(Q)" || { echo 'usage: make sidecar-search Q="your question" SIDECAR=/path/sidecar.bin [LIMIT=N] [MODE=or|and]'; exit 2; } - @test -n "$(SIDECAR)" || { echo 'SIDECAR=/path/sidecar.bin required'; exit 2; } - @$(ARBORIST) sidecar search '$(Q)' --sidecar "$(SIDECAR)" --limit $(or $(LIMIT),8) --mode $(or $(MODE),or) - cloud-fetch-chunk: bootstrap ## fetch + hash-verify one chunk from a bucket [LEAF_HASH=hex BLOB_BASE=https://...] @test -n "$(LEAF_HASH)" || { echo 'usage: make cloud-fetch-chunk LEAF_HASH=hex BLOB_BASE=https://.../blobs'; exit 2; } @test -n "$(BLOB_BASE)" || { echo 'BLOB_BASE required (per-chunk blobs/ base URL)'; exit 2; } diff --git a/arborist/cli.py b/arborist/cli.py index dd7d5bb..563cbc2 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -7337,34 +7337,30 @@ def build_parser() -> argparse.ArgumentParser: sidecar_cmd = sub.add_parser( "sidecar", help=( - "build/query an HTTP-optimized inverted-index sidecar — one " - "small file per shard that the client downloads once and " - "queries locally (sub-ms after warmup). Replaces bucket-" - "direct FTS5 over WAN, which is prohibitive on multi-GB shards." + "build a slim FTS5 sidecar — one small SQLite file per shard " + "with the FTS5 index + minimal metadata (no chunk content). " + "The client downloads it once and queries via real FTS5 bm25, " + "bit-for-bit identical to a local shard." ), ) sidecar_sub = sidecar_cmd.add_subparsers(dest="sidecar_cmd", required=True) - sidecar_build = sidecar_sub.add_parser( - "build", - help="build sidecar.bin from a shard .db file", + sidecar_build_fts = sidecar_sub.add_parser( + "build-fts", + help=( + "build a slim FTS5 sidecar from a shard .db — copies the " + "shard's FTS5 shadow tables verbatim + minimal doc/chunk " + "metadata (no content)." + ), ) - sidecar_build.add_argument("--shard", required=True, help="path to shard .db") - sidecar_build.add_argument("--out", required=True, help="output sidecar.bin path") - sidecar_build.set_defaults(func=_cmd_sidecar_build) - - sidecar_search = sidecar_sub.add_parser( - "search", - help="query a local sidecar.bin file", + sidecar_build_fts.add_argument( + "--shard", required=True, help="path to shard .db", ) - sidecar_search.add_argument("query", type=str) - sidecar_search.add_argument("--sidecar", required=True, help="path to sidecar.bin") - sidecar_search.add_argument("--limit", type=int, default=8) - sidecar_search.add_argument( - "--mode", choices=["or", "and"], default="or", - help="OR (default) ranks by BM25; AND requires every query term", + sidecar_build_fts.add_argument( + "--out", required=True, + help="output slim FTS5 sidecar path (recommended: .idx.db)", ) - sidecar_search.set_defaults(func=_cmd_sidecar_search) + sidecar_build_fts.set_defaults(func=_cmd_sidecar_build_fts) return p @@ -7639,16 +7635,40 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: ts = _time.time() manifest = load_bucket_manifest(args.bucket_url) extra_timings["manifest"] = _time.time() - ts - any_sidecar = any(sh.get("sidecar_url") for sh in manifest.shards) + # Detect what kind of sidecars (if any) the manifest publishes. + # MultiShardSidecarCorpus internally dispatches per shard: + # fts_sidecar_url → FtsSidecarShardClient (slim FTS5, real bm25) + # sidecar_url → SidecarShardClient (legacy custom BM25) + # neither → BucketClient (HttpRangeVFS on big shard) + # So we can always go through MultiShardSidecarCorpus; the + # MultiShardBucketCorpus path is only kept for legacy bucket-only + # manifests with no sidecar of either kind (rare). + any_sidecar = any( + sh.get("sidecar_url") or sh.get("fts_sidecar_url") + for sh in manifest.shards + ) + n_fts_sidecars = sum( + 1 for sh in manifest.shards if sh.get("fts_sidecar_url") + ) + n_custom_sidecars = sum( + 1 for sh in manifest.shards + if sh.get("sidecar_url") and not sh.get("fts_sidecar_url") + ) progress.emit( "manifest.done", shards=len(manifest.shards), - sidecars=sum(1 for sh in manifest.shards if sh.get("sidecar_url")), + sidecars=n_fts_sidecars + n_custom_sidecars, + fts_sidecars=n_fts_sidecars, ms=int(extra_timings["manifest"] * 1000), ) - progress.emit( - "corpus.open.start", - mode="sidecar" if any_sidecar else "bucket-direct", - ) + if n_fts_sidecars and not n_custom_sidecars: + mode = "fts-slim" + elif n_custom_sidecars and not n_fts_sidecars: + mode = "custom-sidecar" + elif any_sidecar: + mode = "sidecar-mixed" + else: + mode = "bucket-direct" + progress.emit("corpus.open.start", mode=mode) ts = _time.time() if any_sidecar: multi = MultiShardSidecarCorpus( @@ -7676,7 +7696,7 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: class _Single: manifest = type("M", (), {"shards": [ {"url": client.endpoint.shard_url, "shard_idx": 0, - "label": "single", "sidecar_url": None} + "label": "single"} ], "blob_base": client.endpoint.blob_base, "snapshot_root": None})() def fts_search(self, q, *, limit=8, raw=False): return client.fts_search(q, limit=limit, raw=raw) @@ -7842,41 +7862,13 @@ def _render_cloud_query_human(result: dict, question: str) -> str: return "\n".join(lines) -def _cmd_sidecar_build(args: argparse.Namespace) -> int: - from arborist.wallet.sidecar import build_sidecar - stats = build_sidecar(args.shard, args.out) +def _cmd_sidecar_build_fts(args: argparse.Namespace) -> int: + from arborist.wallet.fts_sidecar_build import build + stats = build(args.shard, args.out) print(json.dumps(stats, indent=2)) return 0 -def _cmd_sidecar_search(args: argparse.Namespace) -> int: - from arborist.wallet.sidecar import SidecarReader - import time as _t - t0 = _t.time() - sc = SidecarReader(args.sidecar) - load_s = _t.time() - t0 - t0 = _t.time() - hits = sc.search(args.query, limit=args.limit, mode=args.mode) - query_s = _t.time() - t0 - print(json.dumps({ - "query": args.query, - "mode": args.mode, - "load_secs": round(load_s, 3), - "query_secs": round(query_s, 4), - "stats": sc.stats(), - "hits": [ - { - "document_root": h.document_root, - "document_uri": h.document_uri, - "title": h.title, - "score": h.score, - } - for h in hits - ], - }, indent=2, ensure_ascii=False)) - return 0 - - def _cmd_cloud_fetch_chunk(args: argparse.Namespace) -> int: from arborist.merkle import hash_leaf client = _make_bucket_client(args) diff --git a/arborist/qa/_text_norm.py b/arborist/qa/_text_norm.py new file mode 100644 index 0000000..9ec56a2 --- /dev/null +++ b/arborist/qa/_text_norm.py @@ -0,0 +1,84 @@ +"""Tokenization helpers shared across query paths. + +Lifted from the (deleted) custom-sidecar binary format module so the +title-boost reranker in arborist.qa.corpus keeps the same lexical +shape: NFKD accent fold, lowercase, drop stopwords + 1-char tokens +(numeric singletons preserved), Roman-Arabic numeral equivalence. + +Stopword set is the single source of truth in +arborist.wallet.bucket._FTS5_STOPWORDS (imported here so any change +to the FTS5 sanitizer carries through to title-boost matching too). +""" +from __future__ import annotations + +import re +import unicodedata + +from arborist.wallet.bucket import _FTS5_STOPWORDS as STOPWORDS + +_WORD_RE = re.compile(r"[A-Za-z0-9_]+") + + +def fold_accents(text: str) -> str: + """NFKD-normalize + strip combining marks → ASCII-equivalent. + + 'pokémon' → 'pokemon' so the ASCII-only word regex below doesn't + split it into ['pok', 'mon']. + """ + return "".join( + c for c in unicodedata.normalize("NFKD", text) + if not unicodedata.combining(c) + ) + + +# Roman ↔ Arabic numeral equivalence (1..20). A user typing 'final +# fantasy 7' should overlap a title 'Final Fantasy VII'. Beyond 20 +# titles use Arabic digits in practice. +_NUMERAL_PAIRS: dict[str, str] = { + "1": "i", "i": "1", + "2": "ii", "ii": "2", + "3": "iii", "iii": "3", + "4": "iv", "iv": "4", + "5": "v", "v": "5", + "6": "vi", "vi": "6", + "7": "vii", "vii": "7", + "8": "viii", "viii": "8", + "9": "ix", "ix": "9", + "10": "x", "x": "10", + "11": "xi", "xi": "11", + "12": "xii", "xii": "12", + "13": "xiii", "xiii": "13", + "14": "xiv", "xiv": "14", + "15": "xv", "xv": "15", + "16": "xvi", "xvi": "16", + "17": "xvii", "xvii": "17", + "18": "xviii", "xviii": "18", + "19": "xix", "xix": "19", + "20": "xx", "xx": "20", +} + + +def numeral_expand(tokens) -> set: + """Return a set with each token's numeral-equivalent form added.""" + out = set(tokens) + for t in tokens: + alt = _NUMERAL_PAIRS.get(t) + if alt is not None: + out.add(alt) + return out + + +def tokenize_text(text: str) -> list[str]: + """Lowercase word tokens; drop stopwords + 1-char tokens (except + numeric singletons like '7' — needed so numeral_expand can later + pair '7' with 'VII'). Accent-fold first.""" + folded = fold_accents(text) + out: list[str] = [] + for raw in _WORD_RE.findall(folded): + t = raw.lower() + if t in STOPWORDS: + continue + if len(t) <= 1 and not t.isdigit(): + continue + out.append(t) + return out diff --git a/arborist/qa/corpus.py b/arborist/qa/corpus.py index ccacb1c..16563ea 100644 --- a/arborist/qa/corpus.py +++ b/arborist/qa/corpus.py @@ -68,18 +68,26 @@ def apply_title_boost( - higher_is_better=False (FTS5 BM25): boost SUBTRACTED from score Effective boost magnitude: - max(0, overlap - extras/2) * boost + max(0, overlap - extras) * boost where overlap = |query_stems ∩ title_stems| and extras = |title_stems - query_stems| + + The full-strength penalty (extras counted 1:1, not halved) keeps + the boost concentrated on titles whose tokens are a SUBSET of the + query. The dinosaur-extinction bench (2026-05-31) showed that + extras/2 was too lenient: "Edwina the Dinosaur Who Didn't Know + She Was Extinct" (overlap 2 with {dinosaur, extinct}, extras 3 + for {edwina, didn, know}) was getting +4 boost — enough to beat + "Dinosaur" main article's +8 boost when the underlying BM25 + favored Edwina's short article by >4. Full penalty zeroes + Edwina's boost, lets Dinosaur win. Stems strip possessive apostrophes + trailing-s plurals; both sides are numeral-expanded (7↔VII) and accent-folded (é→e). """ if not hits or not query.strip() or boost <= 0: return hits - # Lazy-import: sidecar carries the tokenizer + numeral_expand; - # importing at module top would loop on the from-sidecar imports. - from arborist.wallet.sidecar import ( + from arborist.qa._text_norm import ( _WORD_RE, STOPWORDS, fold_accents, numeral_expand, tokenize_text, ) @@ -110,7 +118,7 @@ def apply_title_boost( rescored.append(h) continue extras = len(title_tokens - query_stems) - effective = max(0.0, overlap - extras / 2.0) + effective = max(0.0, overlap - extras) if effective <= 0: rescored.append(h) continue @@ -460,22 +468,22 @@ class MultiShardSqliteCorpus: class SidecarBucketCorpus: - """Adapter over MultiShardSidecarCorpus (sidecar BM25 + bucket-direct - chunk reads). Default for `arborist cloud query`. + """Adapter over MultiShardSidecarCorpus (slim FTS5 sidecar + + bucket-served chunk content). 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. + fts_body delegates to FTS5 bm25(). fts_title / fts_phrase raise + NotSupportedError — the slim sidecar carries a documents_fts + table that could power fts_title, but the wiring isn't done yet. """ name = "sidecar-bucket" - higher_is_better = True # sidecar BM25 + boost is positive, larger = stronger + # FTS5 bm25 is always negative; lower (more negative) = better. + higher_is_better = False 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).""" + title, score, _shard_url}]).""" self._mscorpus = multi_shard def fts_body(self, query: str, *, limit: int = 8) -> list[Hit]: @@ -494,29 +502,60 @@ class SidecarBucketCorpus: 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." + "fts_title not yet wired for SidecarBucketCorpus — the slim " + "FTS5 sidecar already carries documents_fts (title index); " + "exposing it just needs a per-shard MATCH route here." ) def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]: raise NotSupportedError( "fts_phrase not implemented for SidecarBucketCorpus — " - "phrase index is sidecar v3 work." + "FTS5 supports MATCH '\"phrase here\"' directly; the route " + "just isn't surfaced here yet." ) 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. + # Bare document_root lookup → scan shards until one returns rows. + # Two backend shapes, both routed through hasattr checks: + # 1) FtsSidecarShardClient: chunk metadata local, content fetched + # over HTTP from bucket blobs/ (or shard fallback). + # 2) BucketClient (no sidecar): content inline in chunks.content + # of the big shard .db via apsw HttpRangeVFS. from arborist.compress import unpack_chunk - # Try each shard's conn. for sh in getattr(self._mscorpus, "manifest", None).shards: + try: + client = self._mscorpus._by_url[sh["url"]] + except (KeyError, AttributeError): + continue + # Slim FTS5 client path. + if hasattr(client, "fetch_chunk_body") and hasattr( + client, "chunks_for_doc" + ): + meta = client.chunks_for_doc(document_root) + if not meta: + continue + if limit is not None: + meta = meta[: int(limit)] + out: list[ChunkRow] = [] + for m in meta: + try: + body_bytes = client.fetch_chunk_body(m["leaf_hash"]) + except Exception: + continue + text = unpack_chunk(body_bytes) or "" + if not text: + continue + out.append(ChunkRow( + document_root=document_root, + idx=m["idx"], leaf_hash=m["leaf_hash"], content=text, + )) + if out: + return out + continue + # Bucket-direct path: apsw connection on the big shard. try: conn = self._mscorpus.conn_for_shard(sh["url"]) except (KeyError, AttributeError): diff --git a/arborist/wallet/_sidecar_build_runner.py b/arborist/wallet/_sidecar_build_runner.py deleted file mode 100644 index 7226cd3..0000000 --- a/arborist/wallet/_sidecar_build_runner.py +++ /dev/null @@ -1,31 +0,0 @@ -"""CLI-callable sidecar build runner. - -Tiny shim so shell scripts can do - python3 -m arborist.wallet._sidecar_build_runner SRC DST -without inlining multiline Python that gets eaten by shell quoting. -""" -from __future__ import annotations - -import sys -import time - -from arborist.wallet.sidecar import build_sidecar - - -def main(src: str, dst: str) -> int: - t0 = time.time() - stats = build_sidecar(src, dst) - elapsed = time.time() - t0 - print( - f"build_secs={elapsed:.1f} " - f"file_size_mb={stats['file_bytes']/1024**2:.1f} " - f"terms={stats['terms']} docs={stats['docs']}" - ) - return 0 - - -if __name__ == "__main__": - if len(sys.argv) != 3: - print("usage: python -m arborist.wallet._sidecar_build_runner SRC DST", file=sys.stderr) - sys.exit(2) - sys.exit(main(sys.argv[1], sys.argv[2])) diff --git a/arborist/wallet/bucket.py b/arborist/wallet/bucket.py index f0f147d..8b67097 100644 --- a/arborist/wallet/bucket.py +++ b/arborist/wallet/bucket.py @@ -690,22 +690,6 @@ class MultiShardBucketCorpus: # --------------------------------------------------------------------------- -def _sidecar_cache_path(url: str, *, base_dir: str | None = None) -> str: - """Stable local cache path for a sidecar URL. - - Hashes the URL → places under ~/.arborist/sidecar-cache/.bin so - re-runs reuse the download. base_dir overrides for tests. - """ - import hashlib - if base_dir is None: - base_dir = os.path.join( - os.path.expanduser("~"), ".arborist", "sidecar-cache" - ) - os.makedirs(base_dir, exist_ok=True) - h = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16] - return os.path.join(base_dir, f"{h}.bin") - - def _download_sidecar(url: str, dest: str, *, timeout_s: float = 300.0) -> int: """Stream a sidecar URL into `dest`. Returns bytes downloaded. @@ -729,78 +713,209 @@ def _download_sidecar(url: str, dest: str, *, timeout_s: float = 300.0) -> int: return n -class SidecarShardClient: - """One shard's view: a local sidecar (download-then-query-locally) - plus a BucketClient against the shard .db for chunk content reads. +def _fts_sidecar_cache_path(url: str, *, base_dir: str | None = None) -> str: + """Stable local cache path for a slim FTS5 sidecar URL. - fts_search uses the sidecar (sub-second BM25); chunks_for_doc and - SQL go through the bucket VFS so chunk content fetches benefit - from the same on-disk page cache.""" + Separate cache directory from the legacy custom binary sidecar so + a manifest swap doesn't pull from the wrong cache. Hash of URL keeps + the filename stable; ``.idx.db`` extension makes it recognizable as + SQLite. + """ + import hashlib + if base_dir is None: + base_dir = os.path.join( + os.path.expanduser("~"), ".arborist", "sidecar-fts-cache" + ) + os.makedirs(base_dir, exist_ok=True) + h = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16] + return os.path.join(base_dir, f"{h}.idx.db") + + +class FtsSidecarShardClient: + """One shard's view using a slim FTS5 sidecar. + + The slim sidecar (typically ``.idx.db`` in the bucket) is a + standalone SQLite file containing the source shard's contentless + FTS5 index plus minimal doc/chunk metadata — no chunk content. We + download it once into the local cache and query via plain sqlite3 + (no VFS, no HTTP-range): FTS5 MATCH + bm25() answer locally at + full-speed parity with a local shard. + + Chunk CONTENT is fetched on demand from ``blob_base`` (per-chunk + ``blobs//`` storage), same shape as + BucketClient.fetch_chunk_body. The big shard .db is NOT downloaded. + + This is the "real FTS5 in the sidecar" path that replaces the + hand-rolled BM25 in SidecarReader. Identical scoring to a local + shard by construction (same FTS5 shadow tables, same porter + unicode61 tokenizer). + """ def __init__( self, - shard_url: str, - sidecar_url: str, + fts_sidecar_url: str, blob_base: str, *, - cache_bytes: int = DEFAULT_CACHE_BYTES, + shard_url: str | None = None, sidecar_cache_dir: str | None = None, + cache_bytes: int = DEFAULT_CACHE_BYTES, + timeout_s: float = 30.0, ): - from arborist.wallet.sidecar import SidecarReader - + self.fts_sidecar_url = fts_sidecar_url + self.blob_base = blob_base self.shard_url = shard_url - self.sidecar_url = sidecar_url - # 1. Download or reuse cached sidecar. - cache_path = _sidecar_cache_path(sidecar_url, base_dir=sidecar_cache_dir) - if not os.path.exists(cache_path): - _download_sidecar(sidecar_url, cache_path) - self._sidecar_path = cache_path - self.sidecar = SidecarReader(cache_path) - # 2. Bucket client for chunk content reads. - self.bucket = BucketClient( - BucketEndpoint(shard_url=shard_url, blob_base=blob_base), - cache_bytes=cache_bytes, + cache_path = _fts_sidecar_cache_path( + fts_sidecar_url, base_dir=sidecar_cache_dir ) + if not os.path.exists(cache_path): + _download_sidecar(fts_sidecar_url, cache_path) + self._sidecar_path = cache_path + import sqlite3 as _sql + # check_same_thread=False so MultiShardSidecarCorpus can call + # fts_search from a ThreadPoolExecutor worker. The sqlite3 lib + # itself is built thread-safe (threadsafety=1 or 3 on every + # mainstream distro) — the per-connection check is what blocks + # cross-thread use by default. Read-only PRAGMA + URI ro flag + # keep this safe: no writes, no transactions, no statement + # interleaving worth worrying about. + self._conn = _sql.connect( + f"file:{cache_path}?mode=ro", uri=True, + check_same_thread=False, + ) + self._conn.execute("PRAGMA query_only = 1") + self._timeout_s = timeout_s + # Per-client HTTP stats so MultiShard can aggregate. + self._http_requests = 0 + self._bytes_fetched = 0 + # Optional fallback for buckets without published per-chunk blobs: + # when blob fetch 404s, fall back to HTTP-range on the big shard + # .db (BucketClient). Only constructs the BucketClient lazily on + # first miss so the common case (blobs published) stays free. + self._fallback_bucket: BucketClient | None = None + self._fallback_cache_bytes = cache_bytes def close(self): - self.bucket.close() + try: + self._conn.close() + except Exception: + pass - def fts_search(self, query: str, *, limit: int = 8, raw: bool = False) -> list[dict]: - """Sidecar-driven FTS. ``raw`` is accepted for API parity with - BucketClient.fts_search but ignored — sidecar tokenizes via - tokenize_text (matches the OR-sanitizer behavior).""" - hits = self.sidecar.search(query, limit=limit, mode="or") + def fts_search( + self, query: str, *, limit: int = 8, raw: bool = False, + ) -> list[dict]: + """Local FTS5 MATCH against the slim sidecar. + + Mirrors BucketClient.fts_search shape exactly — same SQL, same + bm25 ordering — so the merge layer in MultiShardSidecarCorpus + treats slim-sidecar shards as homogeneous with bucket-direct + FTS5 shards (both return negative bm25 scores, lower = better). + """ + fts_query = query if raw else _to_fts5(query) + if not fts_query.strip(): + return [] + 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 ?" + ) + rows = list(self._conn.execute(sql, (fts_query, limit))) return [ - { - "document_root": h.document_root, - "document_uri": h.document_uri, - "title": h.title, - "score": h.score, - } - for h in hits + {"document_root": r[0], "document_uri": r[1], + "title": r[2], "score": r[3]} + for r in rows ] - @property - def conn(self): - """apsw connection for SQL pulls (chunk content). Goes through - the bucket VFS — chunk reads are cached.""" - return self.bucket.conn + def chunks_for_doc(self, document_root: str) -> list[dict]: + """Return chunk metadata (idx, leaf_hash) for a document. + + The slim sidecar carries this without needing the big shard. + """ + sql = ( + "SELECT idx, leaf_hash FROM chunks " + "WHERE document_root = ? ORDER BY idx ASC" + ) + rows = list(self._conn.execute(sql, (document_root,))) + return [{"idx": r[0], "leaf_hash": r[1]} for r in rows] + + def fetch_chunk_body(self, leaf_hash: str) -> bytes: + """GET blobs// from the bucket. + + Per-chunk content lives in ``blob_base`` (the JUST_ENOUGH=1 + bucket layout). On 403/404 (blob missing) AND when ``shard_url`` + is set we fall back to an HTTP-range read of chunks.content from + the big shard .db — keeps the slim FTS5 client useful in + transitional buckets that haven't published per-chunk blobs yet. + Hash-verification belongs to the caller. + """ + url = ( + f"{self.blob_base.rstrip('/')}/" + f"{leaf_hash[:2]}/{leaf_hash[2:]}" + ) + req = urllib.request.Request(url) + try: + with urllib.request.urlopen(req, timeout=self._timeout_s) as resp: + data = resp.read() + self._http_requests += 1 + self._bytes_fetched += len(data) + return data + except urllib.error.HTTPError as e: + if e.code not in (403, 404) or not self.shard_url: + raise + # Fallback path — open BucketClient lazily, read chunks.content + # by leaf_hash. This is slower (page-level HTTP range reads on + # the big shard) but works without per-chunk blob storage. + if self._fallback_bucket is None: + self._fallback_bucket = BucketClient( + BucketEndpoint( + shard_url=self.shard_url, blob_base=self.blob_base, + ), + cache_bytes=self._fallback_cache_bytes, + ) + rows = list(self._fallback_bucket.conn.execute( + "SELECT content FROM chunks WHERE leaf_hash = ? " + "AND content IS NOT NULL LIMIT 1", + (leaf_hash,), + )) + if not rows: + raise FileNotFoundError( + f"leaf_hash {leaf_hash[:12]}… not found via blobs/* OR " + f"big-shard chunks.content fallback" + ) + self._http_requests += 1 + data = rows[0][0] + self._bytes_fetched += len(data) if data else 0 + return data + + def stats(self) -> dict: + return { + "http_requests": self._http_requests, + "cache": {"bytes_fetched": self._bytes_fetched}, + "sidecar_path": self._sidecar_path, + "sidecar_bytes": ( + os.path.getsize(self._sidecar_path) + if os.path.exists(self._sidecar_path) else 0 + ), + } + + class MultiShardSidecarCorpus: - """Manifest-driven corpus where each shard MAY have a sidecar. + """Manifest-driven corpus. Per shard, two shapes: + - ``fts_sidecar_url`` → FtsSidecarShardClient (download the slim + FTS5 sidecar locally + bucket-served chunk content). Real FTS5 + bm25() with porter+unicode61 tokenizer; bit-for-bit parity with + a local shard by construction. + - no sidecar → BucketClient (HttpRangeVFS on the big shard .db). + Slow on multi-GB shards but works without prepublished sidecars. - Shards with a `sidecar_url` use SidecarShardClient (download-then- - local FTS). Shards without one fall back to BucketClient (in-place - bucket-direct FTS5 — slow on multi-GB shards, but works for small - shards or when no sidecar is published yet). - - fts_search runs per-shard in parallel, merges by score, picks top-K. - Mixed-mode results are score-comparable in practice: sidecar BM25 - is on a similar magnitude (~10-30) as FTS5's BM25 (~-15 to -1) but - sign-flipped (sidecar is positive, FTS5 stored as negative). We - normalize sidecar to negative before merging so a single ORDER BY - DESC works across both. + fts_search runs per-shard in parallel and merges raw bm25 scores + when every contributing shard uses the same backend. Mixed + (slim-FTS5 + bucket-direct) is score-comparable in practice — both + return negative FTS5 bm25 — so the merge uses MIN across either. """ def __init__( @@ -808,29 +923,28 @@ class MultiShardSidecarCorpus: manifest: "BucketManifest", *, cache_bytes_per_shard: int = DEFAULT_CACHE_BYTES, - sidecar_cache_dir: str | None = None, + fts_sidecar_cache_dir: str | None = None, ): from concurrent.futures import ThreadPoolExecutor self.manifest = manifest self._has_sidecar: list[bool] = [ - bool(sh.get("sidecar_url")) for sh in manifest.shards + bool(sh.get("fts_sidecar_url")) for sh in manifest.shards ] # Pre-download every sidecar in parallel so a fresh consumer - # pays max(per-sidecar download) instead of sum. SidecarShardClient - # is idempotent — if the cache file already exists, it just reads - # from disk (no re-download). Bucket-only shards (no sidecar) - # construct fast; they don't pre-pull the .db. + # pays max(per-sidecar download) instead of sum. Idempotent: if + # the cache file already exists, the client just reads from disk. + # Bucket-only shards (no sidecar) construct fast. def _build(item): i, sh = item - if sh.get("sidecar_url"): - return i, SidecarShardClient( - shard_url=sh["url"], - sidecar_url=sh["sidecar_url"], + if sh.get("fts_sidecar_url"): + return i, FtsSidecarShardClient( + fts_sidecar_url=sh["fts_sidecar_url"], blob_base=manifest.blob_base, + shard_url=sh.get("url"), # content fallback + sidecar_cache_dir=fts_sidecar_cache_dir, cache_bytes=cache_bytes_per_shard, - sidecar_cache_dir=sidecar_cache_dir, ) return i, BucketClient( BucketEndpoint(shard_url=sh["url"], blob_base=manifest.blob_base), @@ -851,115 +965,48 @@ class MultiShardSidecarCorpus: def fts_search( self, query: str, *, limit: int = 8, raw: bool = False, - rrf_k: int = 60, ) -> list[dict]: - """Run each shard's FTS in parallel, merge via reciprocal-rank-fusion. + """Run each shard's FTS in parallel, merge by raw bm25 score. - BM25 scores aren't comparable across heterogeneous backends - (sidecar BM25 is positive in the [5, 50] range; FTS5 BM25 is - negative in [-20, 0]). RRF sidesteps the scale mismatch - entirely: each shard ranks its own hits, the global rank is - the sum of reciprocal ranks across shards. Standard k=60 is - TREC-recommended. - - Post-merge title-relevance filter: drop hits whose titles - share zero query tokens with the query. Mirrors local - query.py's ``search.title_filter`` step — without it, - small shards (e.g. a 200-doc personal blog) hand back rank-1 - hits for any matched token, ties with rank-1 from a 1M-doc - shard under RRF, and ranks above topically correct docs. With - the filter, off-topic docs are dropped before the merge. + Every backend now returns FTS5's negative bm25 (lower = better), + so merge takes the MIN per document_root and sorts ascending — + homogeneous scoring, no RRF needed. """ from concurrent.futures import ThreadPoolExecutor - from arborist.wallet.sidecar import fold_accents, tokenize_text def _one(item): - sh_url, sh_client, _ = item + sh_url, sh_client = item try: - # Pull per-shard top-K (oversample so RRF has more to - # work with — a doc that's barely in top-K on one - # shard but #1 on another should still bubble up). + # Oversample 4× per shard so post-merge ranking has more + # candidates — a doc barely in top-K on one shard but #1 + # on another should still bubble up. return sh_url, sh_client.fts_search(query, limit=limit * 4, raw=raw) except Exception: return sh_url, [] - triples = [ - (sh["url"], self._by_url[sh["url"]], self._has_sidecar[i]) - for i, sh in enumerate(self.manifest.shards) + pairs = [ + (sh["url"], self._by_url[sh["url"]]) + for sh in self.manifest.shards ] per_shard: list[tuple[str, list[dict]]] = [] - with ThreadPoolExecutor(max_workers=max(1, len(triples))) as ex: - for sh_url, hits in ex.map(_one, triples): + with ThreadPoolExecutor(max_workers=max(1, len(pairs))) as ex: + for sh_url, hits in ex.map(_one, pairs): per_shard.append((sh_url, hits)) - # Title-relevance filter: drop hits whose title shares zero - # query tokens. tokenize_text already folds accents + drops - # stopwords, so "Pokémon Red and Blue" → {pokemon, red, blue} - # overlaps query {starter, pokemon, red}; Russell Ballestrini - # blog root → {russell, ballestrini} overlaps NEITHER → dropped. - # Falls open (no filter) if the query has no content tokens - # after sanitization. - query_tokens = set(tokenize_text(query)) - if query_tokens: - def _title_relevant(h: dict) -> bool: - title = h.get("title") or "" - 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)]) - for sh_url, hits in per_shard - ] - - # Merge strategy: when every shard CONTRIBUTING HITS is a - # sidecar, BM25 scores are directly comparable (same constants, - # same title-boost formula), so merge by raw score — preserves - # per-shard rank discrimination instead of squashing the top-N - # into a tie-at-1/(k+1) under RRF. When the surviving hits mix - # sidecar + bucket-direct FTS5 (incomparable scales: BM25 - # +25..+60 vs FTS5 −20..0), fall back to RRF. - url_to_sidecar = { - sh["url"]: self._has_sidecar[i] - for i, sh in enumerate(self.manifest.shards) - } - contributing_shards = { - sh_url for sh_url, hits in per_shard if hits - } - use_raw_score = bool(contributing_shards) and all( - url_to_sidecar[u] for u in contributing_shards - ) - hit_by_root: dict[str, dict] = {} score_by_root: dict[str, float] = {} for sh_url, hits in per_shard: - for rank, h in enumerate(hits, start=1): + for h in hits: droot = h["document_root"] - if use_raw_score: - # Take the MAX raw score across shards. - raw = h.get("score") or 0.0 - if droot not in score_by_root or raw > score_by_root[droot]: - score_by_root[droot] = raw - else: - score_by_root[droot] = ( - score_by_root.get(droot, 0.0) + 1.0 / (rrf_k + rank) - ) + raw_score = h.get("score") or 0.0 + if droot not in score_by_root or raw_score < score_by_root[droot]: + score_by_root[droot] = raw_score if droot not in hit_by_root: h["_shard_url"] = sh_url hit_by_root[droot] = h for droot, h in hit_by_root.items(): h["merge_score"] = score_by_root[droot] - ranked = sorted( - hit_by_root.values(), - key=lambda h: -h["merge_score"], - ) + ranked = sorted(hit_by_root.values(), key=lambda h: h["merge_score"]) return ranked[:limit] def conn_for_shard(self, shard_url: str): diff --git a/arborist/wallet/fts_sidecar_build.py b/arborist/wallet/fts_sidecar_build.py new file mode 100644 index 0000000..8adf14d --- /dev/null +++ b/arborist/wallet/fts_sidecar_build.py @@ -0,0 +1,209 @@ +"""Build a slim FTS5 sidecar from an arborist shard. + +A slim sidecar contains: + - documents (document_root, document_uri, title) — for title-boost + display + - chunks (chunk_id, document_root, idx, leaf_hash) — for evidence map + - chunks_fts + documents_fts (contentless FTS5; copied verbatim from source + shadow tables so the index identity is preserved bit-for-bit) + +It does NOT contain chunk content — that lives in the big shard and is +fetched over HTTP-range via the existing chunk-fetch path when evidence +is needed. + +Idea behind this layout: cloud retrieval scoring must equal local +retrieval scoring. The source shard already uses contentless FTS5 with +'porter unicode61' tokenizer; copying its shadow tables means cloud +FTS5.bm25() == local FTS5.bm25() by construction. No custom BM25 +implementation, no hand-rolled tokenizer, no granularity skew. +""" +from __future__ import annotations + +import os +import sqlite3 +import sys +import time + + +SHADOW_TABLES = { + "chunks_fts": [ + "chunks_fts_data", + "chunks_fts_idx", + "chunks_fts_docsize", + "chunks_fts_config", + ], + "documents_fts": [ + "documents_fts_data", + "documents_fts_idx", + "documents_fts_docsize", + "documents_fts_config", + ], +} + +# DDL must EXACTLY match what the source shard uses or the shadow-table +# bulk-copy will land in differently-organized internal storage and +# MATCH lookups will fail. +DDL_CHUNKS_FTS = ( + "CREATE VIRTUAL TABLE chunks_fts USING fts5(" + "content, content='', contentless_delete=1, " + "tokenize='porter unicode61')" +) +DDL_DOCUMENTS_FTS = ( + "CREATE VIRTUAL TABLE documents_fts USING fts5(" + "title, content='', contentless_delete=1, " + "tokenize='porter unicode61')" +) + + +def _copy_shadow(src: sqlite3.Connection, dst: sqlite3.Connection, + table: str) -> int: + """Bulk-copy one FTS5 shadow table src→dst. + + The destination table was just created (empty) by CREATE VIRTUAL + TABLE; we delete it and re-insert the source bytes verbatim so the + bm25() scorer sees the same posting-list layout. + """ + dst.execute(f"DELETE FROM {table}") + cols_info = dst.execute(f"PRAGMA table_info({table})").fetchall() + cols = [r[1] for r in cols_info] + placeholders = ", ".join("?" * len(cols)) + col_list = ", ".join(cols) + n = 0 + cur = src.execute(f"SELECT {col_list} FROM {table}") + batch: list = [] + for row in cur: + batch.append(row) + if len(batch) >= 1000: + dst.executemany( + f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})", + batch, + ) + n += len(batch) + batch.clear() + if batch: + dst.executemany( + f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})", + batch, + ) + n += len(batch) + return n + + +def build(source_path: str, dest_path: str, *, verbose: bool = True) -> dict: + """Build a slim FTS5 sidecar at ``dest_path`` from ``source_path``. + + Returns a stats dict (sizes, rowcounts, elapsed). + """ + if os.path.exists(dest_path): + raise FileExistsError( + f"refusing to overwrite {dest_path} — delete first if you mean it" + ) + t0 = time.time() + src_size = os.path.getsize(source_path) + + src = sqlite3.connect(f"file:{source_path}?mode=ro", uri=True) + dst = sqlite3.connect(dest_path) + try: + dst.execute("PRAGMA journal_mode=OFF") + dst.execute("PRAGMA synchronous=OFF") + dst.execute("PRAGMA cache_size=-2000000") # 2 GB cache during build + + # 1. Slim documents. + dst.execute( + "CREATE TABLE documents (" + "document_root TEXT PRIMARY KEY, " + "document_uri TEXT, " + "title TEXT)" + ) + t = time.time() + dst.executemany( + "INSERT INTO documents VALUES (?, ?, ?)", + src.execute( + "SELECT document_root, document_uri, title FROM documents" + ), + ) + dst.commit() + n_docs = dst.execute("SELECT COUNT(*) FROM documents").fetchone()[0] + if verbose: + print(f" documents: {n_docs:,} rows in {time.time()-t:.1f}s", + file=sys.stderr) + + # 2. Slim chunks (no content). + dst.execute( + "CREATE TABLE chunks (" + "chunk_id INTEGER PRIMARY KEY, " + "document_root TEXT NOT NULL, " + "idx INTEGER NOT NULL, " + "leaf_hash TEXT NOT NULL)" + ) + t = time.time() + dst.executemany( + "INSERT INTO chunks VALUES (?, ?, ?, ?)", + src.execute( + "SELECT chunk_id, document_root, idx, leaf_hash FROM chunks" + ), + ) + dst.execute( + "CREATE INDEX idx_chunks_document_root ON chunks(document_root)" + ) + dst.commit() + n_chunks = dst.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] + if verbose: + print(f" chunks: {n_chunks:,} rows in {time.time()-t:.1f}s", + file=sys.stderr) + + # 3. FTS5 virtual tables — copy shadow tables verbatim. + for vt_name, ddl in (("chunks_fts", DDL_CHUNKS_FTS), + ("documents_fts", DDL_DOCUMENTS_FTS)): + t = time.time() + dst.execute(ddl) + for shadow in SHADOW_TABLES[vt_name]: + n = _copy_shadow(src, dst, shadow) + if verbose: + print(f" {shadow}: {n:,} rows", + file=sys.stderr) + dst.commit() + if verbose: + print(f" {vt_name}: built in {time.time()-t:.1f}s", + file=sys.stderr) + + # 4. Compact. + if verbose: + print(" VACUUM ...", file=sys.stderr) + dst.execute("VACUUM") + dst.commit() + finally: + src.close() + dst.close() + + dest_size = os.path.getsize(dest_path) + elapsed = time.time() - t0 + return { + "source_path": source_path, + "dest_path": dest_path, + "source_bytes": src_size, + "dest_bytes": dest_size, + "ratio": dest_size / src_size if src_size else 0.0, + "n_documents": n_docs, + "n_chunks": n_chunks, + "elapsed_s": round(elapsed, 2), + } + + +def main(): + import argparse + p = argparse.ArgumentParser() + p.add_argument("source", help="path to source shard .db") + p.add_argument("dest", help="path to write slim FTS5 sidecar") + args = p.parse_args() + stats = build(args.source, args.dest) + print() + print(f"source: {stats['source_bytes']/1e9:.2f} GB → " + f"sidecar: {stats['dest_bytes']/1e9:.2f} GB " + f"({stats['ratio']*100:.1f}%)") + print(f"docs: {stats['n_documents']:,}") + print(f"chunks: {stats['n_chunks']:,}") + print(f"time: {stats['elapsed_s']:.0f}s") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/arborist/wallet/sidecar.py b/arborist/wallet/sidecar.py deleted file mode 100644 index 15f7ae1..0000000 --- a/arborist/wallet/sidecar.py +++ /dev/null @@ -1,685 +0,0 @@ -"""HTTP-optimized inverted-index sidecar for bucket-direct queries. - -The bucket-direct cloud-query path runs FTS5 against a SQLite .db file -served over HTTP RANGE. That's correct but slow on multi-GB shards: -FTS5's b-tree walk touches hundreds of random pages per query, each -costing a WAN round-trip. We measured 4+ minutes per query on a 9 GB -genesis shard. - -This sidecar trades query latency for a one-time-per-corpus download: - - Producer (server) Consumer (wallet client) - ───────────────── ──────────────────────── - read shard.db GET sidecar.bin (one HTTP request) - tokenize chunk text load bytes into memory - build inverted index binary-search dictionary - write sidecar.bin decode posting list (sub-ms) - fetch chunks from blobs/ - -Format (binary, little-endian): - - +───────────────────────────────────────────────────+ - | MAGIC "ARBORIST-SIDECAR-V1\0" | 20 B - | dict_count uint64 | 8 B - | docs_count uint64 | 8 B - | dict_offset uint64 | 8 B - | postings_offset uint64 | 8 B - | docs_offset uint64 | 8 B - +───────────────────────────────────────────────────+ <- 60 B header - | Dictionary (sorted by term) | - | per entry: | - | u16 term_len | - | ... term bytes (utf-8) | - | u32 doc_freq | - | u64 posting_offset (absolute) | - | u32 posting_len | - +───────────────────────────────────────────────────+ - | Posting lists (concatenated) | - | per list: | - | varint num_docs | - | varint deltas (sorted-ascending doc_ids) | - +───────────────────────────────────────────────────+ - | Doc table (doc_id-indexed) | - | per entry: | - | 32 B document_root (hex-decoded) | - | u16 uri_len | - | ... uri bytes (utf-8) | - | u16 title_len | - | ... title bytes (utf-8) | - +───────────────────────────────────────────────────+ - | Doc offset table (doc_id → byte offset in docs) | - | docs_count u64 entries | - +───────────────────────────────────────────────────+ - -Conventions: -- term comparison: case-folded UTF-8 byte order (matches tokenize_text) -- doc_ids are dense [0, docs_count) per sidecar -- varint = LEB128 unsigned (per-byte 0..127; high bit = more bytes) -- multi-term AND via posting-list intersection (default cloud-query - semantics is OR; sidecar supports both via SidecarReader API) - -This is NOT a Merkle-bound artifact. It's a soft index — the doc_root -hashes inside ARE Merkle-bound (chunk content → document_root chain -verifiable via the existing wallet path). A malicious sidecar can DOS -(refuse to surface a hit) but cannot forge content because the wallet -still verifies the chunk body's leaf_hash on retrieval. -""" -from __future__ import annotations - -import bisect -import io -import os -import re -import sqlite3 -import struct -from dataclasses import dataclass -from pathlib import Path -from typing import Iterable, Sequence - -import unicodedata - -from arborist.compress import unpack_chunk - - -# --------------------------------------------------------------------------- -# Tokenization. Matches the bucket-direct sanitizer (arborist.wallet.bucket -# ._to_fts5) so that sidecar-build and sidecar-query agree on what's a term. -# --------------------------------------------------------------------------- - -_WORD_RE = re.compile(r"[A-Za-z0-9_]+") - -# Stopword set mirrors arborist.wallet.bucket._FTS5_STOPWORDS exactly; -# importing avoids drift across the two query paths. -from arborist.wallet.bucket import _FTS5_STOPWORDS as STOPWORDS - - -def fold_accents(text: str) -> str: - """NFKD-normalize + strip combining marks → ASCII-equivalent. - - Critical for retrieval: a user typing 'pokemon' should match - Wikipedia titles like 'Pokémon Red and Blue', and our ASCII-only - word regex was splitting 'Pokémon' into ['Pok', 'mon'] (because é - is not in [A-Za-z]), so the term 'pokemon' never landed in the - dictionary. NFKD folds 'é'→'e' before tokenization fires. - """ - return "".join( - c for c in unicodedata.normalize("NFKD", text) - if not unicodedata.combining(c) - ) - - -# Roman numeral <-> Arabic digit equivalence table for 1..20. -# A user typing 'final fantasy 7' should overlap a title 'Final -# Fantasy VII'. This is the same gap local query.py closes with its -# numeral-fold step. We only need 1..20 — beyond that, titles use -# Arabic digits in practice. -_NUMERAL_PAIRS: dict[str, str] = { - "1": "i", "i": "1", - "2": "ii", "ii": "2", - "3": "iii", "iii": "3", - "4": "iv", "iv": "4", - "5": "v", "v": "5", - "6": "vi", "vi": "6", - "7": "vii", "vii": "7", - "8": "viii","viii": "8", - "9": "ix", "ix": "9", - "10": "x", "x": "10", - "11": "xi", "xi": "11", - "12": "xii","xii": "12", - "13": "xiii","xiii": "13", - "14": "xiv","xiv": "14", - "15": "xv", "xv": "15", - "16": "xvi","xvi": "16", - "17": "xvii","xvii": "17", - "18": "xviii","xviii": "18", - "19": "xix","xix": "19", - "20": "xx", "xx": "20", -} - - -def numeral_expand(tokens) -> set: - """Return a set with each token's numeral-equivalent form added. - - 'final fantasy 7' tokens → {final, fantasy, 7, vii} - 'Final Fantasy VII' tokens → {final, fantasy, vii, 7} - Overlap of the two sets is 4 (or 3 if the duplicate folds), vs 2 - without the fold — enough to lift the right title above siblings - that share only 'final' + 'fantasy'. - """ - out = set(tokens) - for t in tokens: - alt = _NUMERAL_PAIRS.get(t) - if alt is not None: - out.add(alt) - return out - - -def tokenize_text(text: str) -> list[str]: - """Lowercase word tokens, drop stopwords + 1-char tokens (except - numeric singletons like '7' which carry meaning — 'Final Fantasy 7' - needs '7' to survive sanitization so the numeral-fold can later - pair it with 'VII'). Accent-fold first so 'pokémon' indexes/queries - as 'pokemon'.""" - folded = fold_accents(text) - out: list[str] = [] - for raw in _WORD_RE.findall(folded): - t = raw.lower() - if t in STOPWORDS: - continue - if len(t) <= 1 and not t.isdigit(): - continue - out.append(t) - return out - - -# --------------------------------------------------------------------------- -# Binary primitives. -# --------------------------------------------------------------------------- - -MAGIC = b"ARBORIST-SIDECAR-V2\0" -HEADER_FMT = "<20sQQQQQ" -HEADER_LEN = struct.calcsize(HEADER_FMT) - - -def _write_varint(buf: io.BytesIO, n: int) -> None: - """LEB128 unsigned varint.""" - if n < 0: - raise ValueError(f"varint expects unsigned, got {n}") - while n >= 0x80: - buf.write(bytes([(n & 0x7F) | 0x80])) - n >>= 7 - buf.write(bytes([n])) - - -def _read_varint(data: bytes, off: int) -> tuple[int, int]: - """Read LEB128 from ``data`` at ``off``. Returns (value, new_offset).""" - n = 0 - shift = 0 - while True: - b = data[off] - off += 1 - n |= (b & 0x7F) << shift - if not (b & 0x80): - return n, off - shift += 7 - - -# --------------------------------------------------------------------------- -# Producer: build_sidecar(shard_db, out_path). -# --------------------------------------------------------------------------- - - -def build_sidecar( - shard_db: str | Path, - out_path: str | Path, - *, - progress_every: int = 50_000, -) -> dict: - """Read a shard's documents + chunks, tokenize, write sidecar.bin. - - Returns a stats dict: {docs, terms, postings_bytes, file_bytes}. - """ - shard_db = str(shard_db) - out_path = Path(out_path) - - # Phase 1: scan shard, assign dense doc_ids, build term → set(doc_id) map. - print(f"sidecar: scanning {shard_db}") - src = sqlite3.connect(f"file:{shard_db}?mode=ro", uri=True) - src.row_factory = sqlite3.Row - try: - # Per-shard doc table. - doc_rows = list(src.execute( - "SELECT document_root, document_uri, title FROM documents " - "ORDER BY document_root ASC" - )) - docs_count = len(doc_rows) - print(f" docs: {docs_count:,}") - - root_to_id: dict[str, int] = { - r["document_root"]: i for i, r in enumerate(doc_rows) - } - - # Build inverted index with term-frequencies. - # index[term] = list of (doc_id, tf) — appended in doc order - # doc_lens[doc_id] = total term count for the doc (BM25 norm) - # tf is capped at 255 (1-byte varint will use 2 bytes for >127 - # but most chunks-per-doc have tf < 100; ceiling protects against - # adversarial repeating-token docs without changing the math). - index: dict[str, list[tuple[int, int]]] = {} - doc_lens: list[int] = [0] * docs_count - seen = 0 - cursor = src.execute( - "SELECT document_root, content FROM chunks " - "WHERE content IS NOT NULL" - ) - for row in cursor: - doc_id = root_to_id.get(row["document_root"]) - if doc_id is None: - continue - text = unpack_chunk(row["content"]) - if not text: - continue - tf_local: dict[str, int] = {} - for tok in tokenize_text(text): - tf_local[tok] = tf_local.get(tok, 0) + 1 - doc_lens[doc_id] += 1 - for term, tf in tf_local.items(): - lst = index.get(term) - # Cap at 65535 (u16) — easily fits in varint; well above - # natural language tf distribution. - tf_capped = min(tf, 65535) - if lst is None: - index[term] = [(doc_id, tf_capped)] - elif lst[-1][0] != doc_id: - lst.append((doc_id, tf_capped)) - else: - # Same doc, second chunk — accumulate tf. - prev_did, prev_tf = lst[-1] - lst[-1] = (prev_did, min(prev_tf + tf, 65535)) - seen += 1 - if seen % progress_every == 0: - print( - f" scanned {seen:,} chunks " - f"distinct terms: {len(index):,}" - ) - finally: - src.close() - print(f" total terms: {len(index):,} (one entry per word)") - - # Phase 2: serialize. Two passes — first compute offsets, second write. - # Posting lists carry monotonically-increasing doc_ids → delta-encode. - # Sort posting-list doc_ids so deltas stay positive. - sorted_terms = sorted(index.keys()) - - # Build postings section in memory; track per-term (offset, len). - # v2 format: per-doc varint pair (delta, tf). - postings_buf = io.BytesIO() - posting_meta: list[tuple[str, int, int, int]] = [] # term, off, length, doc_freq - for term in sorted_terms: - entries = sorted(index[term]) # by doc_id ascending - start_off = postings_buf.tell() - _write_varint(postings_buf, len(entries)) - prev = -1 - for did, tf in entries: - _write_varint(postings_buf, did - prev - 1) - _write_varint(postings_buf, tf) - prev = did - end_off = postings_buf.tell() - posting_meta.append((term, start_off, end_off - start_off, len(entries))) - postings_bytes = postings_buf.getvalue() - - # Build doc-table bytes + doc-offset-table. v2: doc_len after title. - docs_buf = io.BytesIO() - doc_offsets: list[int] = [] - for i, r in enumerate(doc_rows): - doc_offsets.append(docs_buf.tell()) - droot_hex = r["document_root"] - docs_buf.write(bytes.fromhex(droot_hex)) - uri = (r["document_uri"] or "").encode("utf-8") - title = (r["title"] or "").encode("utf-8") - docs_buf.write(struct.pack(" int | None: - i = bisect.bisect_left(self.terms, term) - if i < len(self.terms) and self.terms[i] == term: - return i - return None - - def _posting_list(self, term_index: int) -> list[tuple[int, int]]: - """Decode a posting list. Returns [(doc_id, tf), ...].""" - po = self.posting_offs[term_index] - pl = self.posting_lens[term_index] - data = self._data - n, off = _read_varint(data, po) - out: list[tuple[int, int]] = [] - prev = -1 - end = po + pl - for _ in range(n): - d, off = _read_varint(data, off) - tf, off = _read_varint(data, off) - cur = prev + 1 + d - out.append((cur, tf)) - prev = cur - if off != end: - raise ValueError( - f"posting parse mismatch for term[{term_index}]: ended at {off}, expected {end}" - ) - return out - - # --- query: AND / OR with score = matched-term count --- - - def search( - self, - query: str, - *, - limit: int = 8, - mode: str = "or", # or | and - k1: float = 1.5, - b: float = 0.75, - title_boost: float = 8.0, - ) -> list[SidecarHit]: - """Tokenize ``query`` (matching tokenize_text), look each term - up in the dictionary, score with BM25, return top ``limit`` hits. - - BM25 per (doc, term): - idf(t) = log((N - df(t) + 0.5) / (df(t) + 0.5) + 1) - tf_norm(d, t) = tf(d, t) * (k1 + 1) - / (tf(d, t) + k1 * (1 - b + b * len(d) / avg_len)) - score(d) = sum over query terms of idf(t) * tf_norm(d, t) - - For ``mode='and'``, posting lists are intersected first; only - docs containing every query term get scored. - """ - import math - - terms = tokenize_text(query) - if not terms: - return [] - # Numeral expand so 'final fantasy 7' query also looks up 'vii' - # in the dict (built from chunk text where 'Final Fantasy VII' - # tokenizes to ['final', 'fantasy', 'vii']). The expansion is - # additive — original tokens stay, equivalents joined as OR - # so docs matching either form contribute to BM25. - terms = sorted(numeral_expand(terms)) - N = self.docs_count - if not hasattr(self, "_avg_doc_len"): - # Compute once + cache; needs the full doc_lens. Lazy because - # reading every doc's varint up-front costs ~0.5s on a 1M-doc - # sidecar. - self._doc_len_cache: dict[int, int] = {} - self._avg_doc_len = self._compute_avg_doc_len() - - # Gather posting lists for terms that exist. - per_term: list[tuple[list[tuple[int, int]], float]] = [] - for t in terms: - ti = self._term_index(t) - if ti is None: - if mode == "and": - return [] - continue - df = self.doc_freqs[ti] - idf = math.log((N - df + 0.5) / (df + 0.5) + 1.0) - per_term.append((self._posting_list(ti), idf)) - if not per_term: - return [] - - if mode == "and": - # Intersect doc_id sets. - doc_sets = [set(d for d, _ in pl) for pl, _ in per_term] - common = doc_sets[0] - for s in doc_sets[1:]: - common &= s - else: - common = None # union path below - - scores: dict[int, float] = {} - for pl, idf in per_term: - for did, tf in pl: - if common is not None and did not in common: - continue - dl = self._doc_length(did) - # Avoid div-by-zero on empty docs (shouldn't happen post- - # ingest, but be defensive). - denom = tf + k1 * (1.0 - b + b * dl / max(self._avg_doc_len, 1.0)) - tf_norm = tf * (k1 + 1.0) / denom if denom > 0 else 0.0 - scores[did] = scores.get(did, 0.0) + idf * tf_norm - - # Title-token boost: docs whose TITLE contains a query term get - # a fixed bonus per overlap. BM25-only ranking penalizes long - # main articles ("Homer Simpson" Wikipedia article) vs short - # episode articles even when the main article is the right - # primary source. The local query.py compensates via a - # separate title-LIKE retrieval route; sidecar emulates with - # a render-time boost using the title we already store in the - # doc table. Stems both sides (possessive + plural collapse — - # matches the title-mismatch verifier's _stem helper) so - # "simpsons" (query) overlaps "simpson" (title). title_boost=0 - # disables; default 8 is enough to lift a #3-by-BM25 long-doc - # match above #1-by-BM25 short-doc matches on the homer query. - if title_boost > 0 and scores: - def _t_stem(t: str) -> str: - t = t.replace("'", "").replace("’", "") - if len(t) > 4 and t.endswith("s") and not t.endswith("ss"): - return t[:-1] - return t - query_stems = numeral_expand({_t_stem(t) for t in terms}) - for did in scores: - # Same accent-fold + tokenize + numeral-expand on the - # title side so 'pokemon'/'pokémon' AND '7'/'VII' both - # overlap correctly. Stopword + length filter (mirrors - # tokenize_text) drops 's' from "Mona Lisa's", 'of'/'the' - # from titles, etc. — otherwise they'd inflate the - # extras count below. - title = fold_accents(self._doc_title(did).lower()) - raw_title_tokens: set[str] = set() - for tok in _WORD_RE.findall(title): - if tok in STOPWORDS: - continue - if len(tok) <= 1 and not tok.isdigit(): - continue - raw_title_tokens.add(_t_stem(tok)) - title_tokens = numeral_expand(raw_title_tokens) - overlap = len(query_stems & title_tokens) - if not overlap: - continue - # Penalty for extra title tokens not in the query — - # without this, "Mona Lisa" (extras 0) and "Mona Lisa's - # Revenge" (extras 1: revenge) tied at the same boost - # and BM25 picked the shorter movie article over the - # painting. extras/2 means each extra halves the per- - # overlap-token bonus. - extras = len(title_tokens - query_stems) - effective = max(0.0, overlap - extras / 2.0) - if effective: - scores[did] += title_boost * effective - - ranked = sorted(scores.items(), key=lambda x: -x[1]) - out: list[SidecarHit] = [] - for did, score in ranked[:limit]: - out.append(self._doc_record(did, score=score)) - return out - - def _doc_title(self, doc_id: int) -> str: - """Read just the title field from the doc table (no allocation - of the rest of the doc record).""" - off = self.docs_offset + self._doc_offsets[doc_id] - off += 32 # document_root - (uri_len,) = struct.unpack_from(" int: - """Read doc_len from the doc table (varint after title).""" - cached = getattr(self, "_doc_len_cache", None) - if cached is not None and doc_id in cached: - return cached[doc_id] - off = self.docs_offset + self._doc_offsets[doc_id] - off += 32 # document_root - (uri_len,) = struct.unpack_from(" float: - """Walk every doc-table entry to compute the corpus average. - Cost ~0.5s for 1M docs (called once, cached).""" - total = 0 - N = self.docs_count - for i in range(N): - total += self._doc_length(i) - return (total / N) if N else 0.0 - - # --- doc-table lookup --- - - def _doc_record(self, doc_id: int, *, score: float = 0.0) -> SidecarHit: - if doc_id < 0 or doc_id >= self.docs_count: - raise IndexError(f"doc_id {doc_id} out of range") - off = self.docs_offset + self._doc_offsets[doc_id] - droot = self._data[off:off + 32].hex() - off += 32 - (uri_len,) = struct.unpack_from(" int: - return self.docs_count - - def term_count(self) -> int: - return self.dict_count - - def stats(self) -> dict: - return { - "file_bytes": len(self._data), - "dict_count": self.dict_count, - "docs_count": self.docs_count, - "dict_size_bytes": self.postings_offset - self.dict_offset, - "postings_size_bytes": self.docs_offset - self.postings_offset, - } diff --git a/bench/slim_fts_parity_bench.py b/bench/slim_fts_parity_bench.py new file mode 100644 index 0000000..bd834ba --- /dev/null +++ b/bench/slim_fts_parity_bench.py @@ -0,0 +1,263 @@ +"""Slim FTS5 sidecar parity bench — does the new cloud path agree with local? + +For each smoke fixture question, run: + legacy: arborist query (current local path) + corpus: arborist corpus-query (local shards via Corpus protocol) + slim-fts: FtsSidecarShardClient + run_query (NEW cloud path, exercised + locally by mocking the manifest with file:// sidecars + HTTP + chunk-content fallback to the real bucket .db) + +This proves the slim FTS5 cloud path produces the same primary-source +pick as the local paths BEFORE we upload anything. If slim-fts matches +corpus bit-for-bit (or better, matches legacy too), the bucket upload is +safe to do. + +Sidecars are expected at ~/.arborist/sidecar-fts/00*.idx.db. Build with: + make sidecar-build-fts-all +""" +from __future__ import annotations + +import argparse +import datetime as _dt +import json +import os +import subprocess +import sys +import time +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +ARBORIST = REPO / ".venv" / "bin" / "arborist" +DEFAULT_SHARDS = Path.home() / ".arborist" / "shards" +DEFAULT_SIDECAR_FTS = Path.home() / ".arborist" / "sidecar-fts" +DEFAULT_FIXTURE = REPO / "bench" / "qa_questions_smoke.txt" +DEFAULT_QWEN_ENDPOINT = "https://qwen.ai.unturf.com/v1" +DEFAULT_QWEN_MODEL = "Qwen3.6-27B-UD-Q4_K_XL.gguf" +DEFAULT_BUCKET_BLOB_BASE = "https://nyc3.digitaloceanspaces.com/arborist/blobs" +DEFAULT_LIVE_MANIFEST = ( + "https://nyc3.digitaloceanspaces.com/arborist/clones/manifest-sidecar.json" +) + + +def _load_fixture(path: Path) -> list[str]: + out = [] + for line in path.read_text().splitlines(): + s = line.strip() + if s and not s.startswith("#"): + out.append(s) + return out + + +def _run_cli(cmd: list[str], *, timeout_s: int) -> dict: + t0 = time.time() + try: + proc = subprocess.run( + cmd, capture_output=True, text=True, timeout=timeout_s, + env={**os.environ, "ARBORIST_PROGRESS": "0"}, + ) + except subprocess.TimeoutExpired: + return {"_error": f"timeout {timeout_s}s", "_elapsed_s": timeout_s} + dt = time.time() - t0 + if proc.returncode != 0: + return {"_error": f"exit {proc.returncode}", + "_stderr": proc.stderr[:300], "_elapsed_s": round(dt, 2)} + try: + d = json.loads(proc.stdout) + except json.JSONDecodeError as e: + return {"_error": f"parse: {e}", "_elapsed_s": round(dt, 2)} + d["_elapsed_s"] = round(dt, 2) + return d + + +def _legacy(q, *, shards_dir): + return _run_cli([ + str(ARBORIST), "--shards-dir", str(shards_dir), + "query", "--top-k", "8", "--json", "--burn", + "--answer-mode", "claim_lattice", q, + ], timeout_s=120) + + +def _corpus(q, *, shards_dir, endpoint, model): + return _run_cli([ + str(ARBORIST), "--shards-dir", str(shards_dir), + "corpus-query", q, + "--top-k", "4", "--json", + "--endpoint", endpoint, "--model", model, + ], timeout_s=120) + + +def _slim_fts( + q, *, sidecar_dir, live_manifest_url, endpoint, model, +): + """Slim FTS5 path: load the LIVE manifest (so chunk-content fallback + points at the real big-shard URLs) then splice in + ``fts_sidecar_url=file://...`` per shard so the slim FTS5 client + opens locally without a download. + + Pre-symlinks the file:// URL into the FTS-sidecar cache to avoid a + duplicate 2 GB copy through urllib. + """ + from arborist.wallet.bucket import ( + load_bucket_manifest, MultiShardSidecarCorpus, + _fts_sidecar_cache_path, + ) + from arborist.qa.corpus import SidecarBucketCorpus + from arborist.qa.client import OpenAICompatibleClient + from arborist.qa.corpus_query import run_query + + manifest = load_bucket_manifest(live_manifest_url) + # Splice fts_sidecar_url per shard by matching basename (e.g. 000.db + # → 000.idx.db); leave shard ``url`` + ``blob_base`` as the live + # values so the missing-blob fallback hits the real big shard. + for sh in manifest.shards: + basename = sh["url"].rsplit("/", 1)[-1] # "000.db" + n = basename.rsplit(".", 1)[0] # "000" + sidecar_path = sidecar_dir / f"{n}.idx.db" + if not sidecar_path.exists(): + continue + fts_sidecar_url = f"file://{sidecar_path}" + cache_path = _fts_sidecar_cache_path(fts_sidecar_url) + if not os.path.exists(cache_path): + os.symlink(sidecar_path, cache_path) + sh["fts_sidecar_url"] = fts_sidecar_url + # Drop shards without a slim sidecar (e.g. virtback) — parity bench + # only exercises the four wikipedia shards we built. + manifest.shards = [ + sh for sh in manifest.shards if sh.get("fts_sidecar_url") + ] + + t0 = time.time() + multi = MultiShardSidecarCorpus(manifest) + corpus = SidecarBucketCorpus(multi) + + client = OpenAICompatibleClient(base_url=endpoint) + result = run_query( + corpus, q, client, + model_id=model, top_k=4, max_context_chars=24_000, + ) + multi.close() + elapsed = time.time() - t0 + + out = dict(result) + out["_elapsed_s"] = round(elapsed, 2) + return out + + +def _primary(r: dict) -> dict: + srcs = r.get("sources") or [] + if not srcs: + return {"title": "", "uri": ""} + pri = next( + (s for s in srcs if s.get("source_role") == "primary_answer_source"), + srcs[0], + ) + return {"title": (pri.get("title") or "")[:55], + "uri": pri.get("document_uri") or ""} + + +def main(): + p = argparse.ArgumentParser() + p.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE) + p.add_argument("--shards-dir", type=Path, default=DEFAULT_SHARDS) + p.add_argument("--sidecar-dir", type=Path, default=DEFAULT_SIDECAR_FTS) + p.add_argument("--endpoint", default=DEFAULT_QWEN_ENDPOINT) + p.add_argument("--model", default=DEFAULT_QWEN_MODEL) + p.add_argument("--live-manifest", default=DEFAULT_LIVE_MANIFEST, + help="real bucket manifest URL — provides the shard " + "URLs used for the chunk-content fallback") + p.add_argument( + "--out-dir", type=Path, + default=REPO / "bench" / "slim_fts_parity_results", + ) + p.add_argument("--skip-legacy", action="store_true") + p.add_argument("--skip-corpus", action="store_true") + args = p.parse_args() + + questions = _load_fixture(args.fixture) + args.out_dir.mkdir(parents=True, exist_ok=True) + stamp = _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ") + out_path = args.out_dir / f"{stamp}.jsonl" + + print(f"# {len(questions)} questions from {args.fixture}", file=sys.stderr) + print(f"# shards: {args.shards_dir}", file=sys.stderr) + print(f"# sidecars-fts: {args.sidecar_dir}", file=sys.stderr) + print(f"# live manifest: {args.live_manifest}", file=sys.stderr) + print(f"# llm: {args.endpoint} / {args.model}", file=sys.stderr) + print(file=sys.stderr) + + n_agree = 0 + n_disagree = 0 + + with open(out_path, "w") as f: + for i, q in enumerate(questions, 1): + print(f"[{i}/{len(questions)}] {q}", file=sys.stderr) + legacy = {} if args.skip_legacy else _legacy( + q, shards_dir=args.shards_dir) + corpus = {} if args.skip_corpus else _corpus( + q, shards_dir=args.shards_dir, + endpoint=args.endpoint, model=args.model) + slim = _slim_fts( + q, + sidecar_dir=args.sidecar_dir, + live_manifest_url=args.live_manifest, + endpoint=args.endpoint, model=args.model, + ) + + row = {"question": q} + for label, r in ( + ("legacy", legacy), ("corpus", corpus), ("slim_fts", slim), + ): + row[label] = { + "audit_mode": r.get("audit_mode"), + "n_verified": r.get("n_verified"), + "n_quotes": r.get("n_quotes"), + "primary": _primary(r), + "elapsed_s": r.get("_elapsed_s") or ( + r.get("timings") or {}).get("total"), + "error": r.get("_error") or r.get("error"), + } + uris = { + lbl: row[lbl]["primary"]["uri"] + for lbl in ("legacy", "corpus", "slim_fts") + if row[lbl]["primary"]["uri"] + } + distinct = set(uris.values()) + row["agreement"] = { + "n_paths_with_source": len(uris), + "distinct_sources": len(distinct), + "all_agree": len(distinct) <= 1, + } + if row["agreement"]["all_agree"]: + n_agree += 1 + else: + n_disagree += 1 + + f.write(json.dumps(row, ensure_ascii=False) + "\n") + f.flush() + + for lbl in ("legacy", "corpus", "slim_fts"): + r = row[lbl] + badge = ( + " ←" if r["primary"]["uri"] + and len(distinct) > 1 + and r["primary"]["uri"] != min(uris.values()) else "" + ) + err = f" ERR={r['error']}" if r["error"] else "" + print(f" {lbl:9s} {(r['audit_mode'] or '?'):<22} " + f"{r['primary']['title']}{badge}{err}", + file=sys.stderr) + if not row["agreement"]["all_agree"]: + print(f" SOURCE DISAGREEMENT: {len(distinct)} distinct primaries", + file=sys.stderr) + print(file=sys.stderr) + + print(f"=== SUMMARY ===", file=sys.stderr) + print(f" {len(questions)} questions", file=sys.stderr) + print(f" {n_agree} all-paths agree on primary source", file=sys.stderr) + print(f" {n_disagree} paths-disagree", file=sys.stderr) + print(f" results: {out_path}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/upload_fts_sidecars.py b/scripts/upload_fts_sidecars.py new file mode 100644 index 0000000..8dafe85 --- /dev/null +++ b/scripts/upload_fts_sidecars.py @@ -0,0 +1,194 @@ +"""Upload slim FTS5 sidecars to the bucket + publish manifest-fts.json. + +Reads slim FTS5 sidecars from ~/.arborist/sidecar-fts/00*.idx.db, uploads +them to s3://${ARBORIST_COLD_BUCKET}/clones/sidecars-fts/.idx.db, and +publishes a new manifest at clones/manifest-fts.json that references both: + - the new fts_sidecar_url (slim FTS5) + - the existing shard URL from manifest-sidecar.json (chunk-content fallback) + +Idempotent: skips uploads whose remote size matches local. Existing +manifest-sidecar.json is NOT modified. + +Credentials via boto3 standard discovery (env vars / ~/.aws/credentials). +Operation Voyeur: credentials never appear in argv, logs, or this file. + +Usage: + .venv/bin/python3 scripts/upload_fts_sidecars.py [--dry-run] +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.request +from pathlib import Path + +SIDECAR_DIR = Path.home() / ".arborist" / "sidecar-fts" +PREFIX = "clones/sidecars-fts" +NEW_MANIFEST_KEY = "clones/manifest-fts.json" +EXISTING_MANIFEST_URL = ( + "https://nyc3.digitaloceanspaces.com/arborist/clones/manifest-sidecar.json" +) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--dry-run", action="store_true", + help="print what would happen; upload nothing") + args = ap.parse_args() + + bucket = os.environ.get("ARBORIST_COLD_BUCKET") + endpoint = os.environ.get("ARBORIST_COLD_ENDPOINT_URL") + if not bucket or not endpoint: + print("ERROR: set ARBORIST_COLD_BUCKET + ARBORIST_COLD_ENDPOINT_URL", + file=sys.stderr) + return 2 + + sidecars = sorted(SIDECAR_DIR.glob("[0-9][0-9][0-9].idx.db")) + if not sidecars: + print(f"ERROR: no sidecars at {SIDECAR_DIR}/[0-9][0-9][0-9].idx.db", + file=sys.stderr) + return 2 + + print(f"bucket: s3://{bucket}/", file=sys.stderr) + print(f"endpoint: {endpoint}", file=sys.stderr) + print(f"prefix: {PREFIX}/", file=sys.stderr) + print(f"sidecars: {len(sidecars)} files ({sum(p.stat().st_size for p in sidecars)/1e9:.2f} GB)", + file=sys.stderr) + print(file=sys.stderr) + + # Load the existing manifest's shard URLs so the new manifest reuses + # them for the chunk-content fallback. We don't touch the existing + # manifest itself — just borrow the URLs. + with urllib.request.urlopen(EXISTING_MANIFEST_URL, timeout=30) as resp: + existing = json.loads(resp.read().decode()) + # Map by trailing 3-digit shard index parsed from label + # ("genesis-shard-000" → "000", "genesis-shard-002-64k" → "002"). + # Drops non-genesis shards (virtback, etc.) so the basename collision + # virtback/000.db vs full-bench/000.db doesn't matter. + import re as _re + _GENESIS_IDX = _re.compile(r"genesis-shard-(\d{3})") + by_idx = {} + for sh in existing.get("shards", []): + m = _GENESIS_IDX.search(sh.get("label", "")) + if m: + by_idx[m.group(1)] = sh + + try: + import boto3 + from boto3.s3.transfer import TransferConfig + except ImportError: + print("ERROR: boto3 not installed. run: make bootstrap-object-store", + file=sys.stderr) + return 2 + + s3 = boto3.client("s3", endpoint_url=endpoint, region_name=os.environ.get( + "ARBORIST_COLD_REGION", "us-east-1")) + # 64 MB parts; parallel transfer. + cfg = TransferConfig( + multipart_threshold=64 * 1024 * 1024, + multipart_chunksize=64 * 1024 * 1024, + max_concurrency=4, + ) + + new_shards = [] + for sp in sidecars: + n = sp.stem.split(".")[0] # 000.idx → 000 + key = f"{PREFIX}/{n}.idx.db" + local_size = sp.stat().st_size + + # Idempotency: skip if remote size matches. + try: + head = s3.head_object(Bucket=bucket, Key=key) + remote_size = head["ContentLength"] + if remote_size == local_size: + print(f" skip {key} (already {local_size/1e9:.2f} GB)", + file=sys.stderr) + else: + print(f" remote size mismatch on {key} " + f"(local {local_size}, remote {remote_size}) — re-uploading", + file=sys.stderr) + if not args.dry_run: + s3.upload_file(str(sp), bucket, key, Config=cfg) + else: + print(f" [dry-run] would re-upload {key}", + file=sys.stderr) + except s3.exceptions.ClientError as e: + code = e.response.get("Error", {}).get("Code") + if code in ("404", "NoSuchKey", "NotFound"): + if args.dry_run: + print(f" [dry-run] would upload {key} ({local_size/1e9:.2f} GB)", + file=sys.stderr) + else: + print(f" upload {key} ({local_size/1e9:.2f} GB) ...", + file=sys.stderr) + s3.upload_file(str(sp), bucket, key, Config=cfg) + s3.put_object_acl(Bucket=bucket, Key=key, ACL="public-read") + print(f" done", file=sys.stderr) + else: + raise + + # Ensure ACL is public-read (idempotent for ACL-aware backends). + if not args.dry_run: + try: + s3.put_object_acl(Bucket=bucket, Key=key, ACL="public-read") + except Exception: + pass # endpoint may not honor ACLs + + # Build new manifest entry — reuse existing shard url + blob_base. + ex = by_idx.get(n) + if ex is None: + print(f" WARNING: no existing genesis-shard-{n} in live manifest; " + f"the new manifest will lack a content-fallback URL " + f"(slim FTS5 retrieval still works; blob fetch falls back " + f"to bucket blobs/ which 403s today)", + file=sys.stderr) + new_shard = { + "url": None, + "shard_idx": int(n), + "label": f"genesis-shard-{n}", + "fts_sidecar_url": f"{endpoint}/{bucket}/{key}", + } + else: + new_shard = dict(ex) + new_shard["fts_sidecar_url"] = f"{endpoint}/{bucket}/{key}" + new_shard.pop("sidecar_url", None) # this manifest is FTS5-only + new_shards.append(new_shard) + + new_manifest = { + "shards": new_shards, + "blob_base": existing.get("blob_base"), + "snapshot_root": existing.get("snapshot_root"), + "snapshot_ts": existing.get("snapshot_ts"), + "format": "fts5-slim-v1", + } + manifest_bytes = json.dumps(new_manifest, indent=2).encode() + print(file=sys.stderr) + print(f"new manifest: {NEW_MANIFEST_KEY} ({len(manifest_bytes)} bytes)", + file=sys.stderr) + print(f" shards: {len(new_shards)}", file=sys.stderr) + for sh in new_shards: + print(f" fts_sidecar={sh['fts_sidecar_url'].rsplit('/',1)[-1]:14s} " + f"shard_url={sh.get('url','(none)').rsplit('/',1)[-1]}", + file=sys.stderr) + + if args.dry_run: + print(file=sys.stderr) + print("[dry-run] manifest content:", file=sys.stderr) + print(manifest_bytes.decode(), file=sys.stderr) + return 0 + + s3.put_object( + Bucket=bucket, Key=NEW_MANIFEST_KEY, + Body=manifest_bytes, + ContentType="application/json", + ACL="public-read", + ) + print(f"published: {endpoint}/{bucket}/{NEW_MANIFEST_KEY}", + file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_qa_corpus.py b/tests/test_qa_corpus.py index 415055f..7eb8065 100644 --- a/tests/test_qa_corpus.py +++ b/tests/test_qa_corpus.py @@ -122,7 +122,8 @@ def test_sidecar_bucket_corpus_satisfies_protocol(): corpus = SidecarBucketCorpus(_StubMulti()) assert isinstance(corpus, Corpus) assert corpus.name == "sidecar-bucket" - assert corpus.higher_is_better is True + # FTS5 bm25 is negative; lower (more negative) = better. + assert corpus.higher_is_better is False # --------------------------------------------------------------------------- diff --git a/tests/test_qa_corpus_functional.py b/tests/test_qa_corpus_functional.py index f9a29cd..d60ad78 100644 --- a/tests/test_qa_corpus_functional.py +++ b/tests/test_qa_corpus_functional.py @@ -53,7 +53,7 @@ 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 +from arborist.wallet.fts_sidecar_build import build as build_fts_sidecar # --------------------------------------------------------------------------- @@ -101,14 +101,14 @@ def both_adapters(tmp_path: Path): c0.execute("PRAGMA wal_checkpoint(FULL)") c0.close() - # bucket layout + sidecar + # bucket layout + slim FTS5 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) + sidecar_path = clones / "000.idx.db" + build_fts_sidecar(str(bucket_shard), str(sidecar_path), verbose=False) # in-process Range-aware server cwd_before = os.getcwd() @@ -123,14 +123,14 @@ def both_adapters(tmp_path: Path): "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", + "fts_sidecar_url": f"{base_url}/clones/snap-1/000.idx.db", }], blob_base=f"{base_url}/blobs", snapshot_root=None, snapshot_ts=None, ) multi = MultiShardSidecarCorpus( - manifest, sidecar_cache_dir=str(tmp_path / "cache"), + manifest, fts_sidecar_cache_dir=str(tmp_path / "cache"), ) conn = connect(db) diff --git a/tests/test_qa_corpus_integration.py b/tests/test_qa_corpus_integration.py index ff30a40..1dd63dd 100644 --- a/tests/test_qa_corpus_integration.py +++ b/tests/test_qa_corpus_integration.py @@ -45,7 +45,7 @@ from arborist.wallet.bucket import ( BucketManifest, MultiShardSidecarCorpus, ) -from arborist.wallet.sidecar import build_sidecar +from arborist.wallet.fts_sidecar_build import build as build_fts_sidecar # --------------------------------------------------------------------------- @@ -90,8 +90,8 @@ def served_corpus(tmp_path: Path): import shutil bucket_shard = clones / "000.db" shutil.copy(db, bucket_shard) - sidecar_path = clones / "000.sidecar.bin" - build_sidecar(bucket_shard, sidecar_path) + sidecar_path = clones / "000.idx.db" + build_fts_sidecar(str(bucket_shard), str(sidecar_path), verbose=False) # Re-open the original for SqliteShardCorpus tests. conn = connect(db) @@ -110,13 +110,13 @@ def served_corpus(tmp_path: Path): "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", + "fts_sidecar_url": f"{base_url}/clones/snap-1/000.idx.db", }], blob_base=f"{base_url}/blobs", snapshot_root=None, snapshot_ts=None, ) - multi = MultiShardSidecarCorpus(manifest, sidecar_cache_dir=str(tmp_path / "cache")) + multi = MultiShardSidecarCorpus(manifest, fts_sidecar_cache_dir=str(tmp_path / "cache")) try: yield conn, multi, base_url