Swift corpus packs

gutenberg_kg.export_swift

Build the on-device corpus packs the native app searches locally.

Phase 0 of analysis/APP_ARCHITECTURE.md: turn bundles/gutenberg-all/ — a 2.9 GB DocKG graph.sqlite plus a 1.1 GB vector store plus four diary KGs — into three SQLite files small enough to sit on a phone, carrying exactly what the query path reads and nothing else.

Layout produced

::

<out>/
  manifest.json      pack versions, checksums, the embedder the packs require
  core.pack          catalog: genres, books, the Browse entry points
  gutenberg.pack     passages + FTS5 for the 241 books
  gutenberg.vectors  their embeddings, row-major, memory-mappable
  diaries.pack       the same, for the four diaries, with timestamps
  diaries.vectors
  golden.json        reference top-k per query — the Swift parity gate

What is dropped, and why it is safe

The served query path (serve/handler.py:_semantic_search) is a dense kNN plus an FTS5/BM25 query plus RRF over kind IN ('chunk','section'), with content hydrated from SQLite afterwards. It never hops the graph. So the packs carry chunk and section nodes only: the 324 K topic/entity/keyword nodes, every edge, and the structured embed-text duplicate of each passage all stay behind. Vectors are re-encoded from fp32 to int8, which is where most of the remaining size goes.

Why the vectors sit beside the pack, not inside it

A vec0 virtual table cannot be read without the sqlite-vec C extension compiled into the reader, and iOS ships stock SQLite — so a pack built that way would not open on the device it was built for. Vendoring ten thousand lines of C would buy nothing either: vec0's search is exhaustive, and so is the dot product the app does instead over a memory-mapped file, with SIMD and no allocation.

The sidecar is therefore a 32-byte header (magic, dtype, dim, count) followed by row-major vectors. passages.vector_index is a dense row number into it, so a passage whose vector the source store lacks leaves no hole in the file. The header lets a reader reject a truncated download instead of interpreting whatever bytes it finds as embeddings.

Schema notes

Two title columns are deliberate: title is the work's title, which the hit cards show, while node_title is the node's own — a section's chapter name, which the Browse tab lists. Collapsing them loses the chapter list.

The FTS5 index is rebuilt here over clean passage text rather than copied, because the worker's nodes_fts indexes the embed-text form. Lexical results will therefore not match the worker's token for token — which is why golden.json is generated from the pack itself. It is the contract the Swift engine must reproduce, not a record of what the worker happened to return.

Reading the source

Column sets differ across the fleet's stores (a diary carries timestamp, a book carries chapter), and they change as KGs are rebuilt, so every read here is driven by PRAGMA table_info rather than a hardcoded SELECT. Missing columns become NULL in the pack; they never become a crash three hundred thousand rows into an export.

Usage

::

gutenkg export-swift                       # bundles/gutenberg-all → bundles/gutenberg-all/swift
gutenkg export-swift --dtype float         # fp32 vectors, ~3x larger, exact
gutenkg export-swift --verify              # measure int8 recall against the source

BundlePaths(root, graph, catalog, vectors, diaries=()) dataclass

Everything the export reads, resolved and checked to exist.

DiarySource(slug, graph, vectors) dataclass

One DiaryKG inside the bundle.

ExportError

Bases: RuntimeError

A bundle could not be read, or a required dependency is missing.

ExportOptions(bundle=DEFAULT_BUNDLE, out=None, dtype='int8', max_chars=0, include_diaries=True, with_vectors=True, golden=True, golden_k=10, verify=False, force=False, catalog_keys=None, diary_dirs=None, golden_queries=None, product_name=None, product_version=None) dataclass

Inputs to :func:export_swift.

catalog_keys, diary_dirs, golden_queries, product_name, and product_version are the selective-bundle-export additions (analysis/SELECTIVE_BUNDLE_EXPORT_PLAN.md phase 2). Every one of them defaults to None, and None reproduces the export's original, unfiltered behaviour exactly — an existing caller that never sets them is unaffected.

resolved_out()

Output directory; defaults to <bundle>/swift.

Source code in src/gutenberg_kg/export_swift.py

def resolved_out(self) -> Path:
    """Output directory; defaults to ``<bundle>/swift``."""
    return Path(self.out) if self.out else Path(self.bundle) / "swift"

ExportReport(out, packs=list(), golden=None, verification=None, elapsed_s=0.0) dataclass

What an export produced.

PackStats(path, passages=0, vectors=0, missing_vectors=0, bytes=0, sha256='', sidecar=None, sidecar_bytes=0, sidecar_sha256='') dataclass

What a finished pack contains.

total_bytes property

Pack plus sidecar — what actually has to reach the device.

build_core_pack(dest, *, bundle, catalog, diary_count, catalog_keys=None)

Write core.pack — the catalog the Browse tab and the scope picker read.

Parameters:

Name	Type	Description	Default
dest	Path	Output path.	required
bundle	BundlePaths	Resolved bundle paths.	required
catalog	dict[str, dict]	Parsed catalog.json.	required
diary_count	int	Diaries included in this export.	required
catalog_keys	frozenset[str] | None	Restrict to these "<genre>/<book>" keys; None keeps every book in catalog.	None

Returns:

Type	Description
PackStats	Stats for the finished pack.

Source code in src/gutenberg_kg/export_swift.py

def build_core_pack(
    dest: Path,
    *,
    bundle: BundlePaths,
    catalog: dict[str, dict],
    diary_count: int,
    catalog_keys: frozenset[str] | None = None,
) -> PackStats:
    """Write ``core.pack`` — the catalog the Browse tab and the scope picker read.

    :param dest: Output path.
    :param bundle: Resolved bundle paths.
    :param catalog: Parsed ``catalog.json``.
    :param diary_count: Diaries included in this export.
    :param catalog_keys: Restrict to these ``"<genre>/<book>"`` keys; ``None``
        keeps every book in ``catalog``.
    :returns: Stats for the finished pack.
    """
    if catalog_keys is not None:
        catalog = {key: meta for key, meta in catalog.items() if key in catalog_keys}
    con = _fresh_db(dest, _CORE_SCHEMA)
    try:
        file_paths = _document_paths(bundle.graph)
        rows = []
        genre_counts: dict[str, int] = {}
        for key, meta in sorted(catalog.items()):
            genre = meta.get("genre") or key.split("/")[0]
            book = meta.get("book") or key.split("/")[-1]
            genre_counts[genre] = genre_counts.get(genre, 0) + 1
            rows.append(
                (
                    key,
                    genre,
                    book,
                    meta.get("title"),
                    meta.get("author"),
                    meta.get("ebook_id"),
                    file_paths.get(key),
                )
            )
        con.executemany(
            "INSERT OR REPLACE INTO books"
            "(key, genre, book, title, author, ebook_id, file_path) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            rows,
        )
        con.executemany(
            "INSERT OR REPLACE INTO genres(genre, book_count) VALUES (?, ?)",
            sorted(genre_counts.items()),
        )

        nodes, edges = _graph_totals(bundle.graph)
        con.execute(
            "INSERT INTO corpus_stats(books, genres, diaries, nodes, edges, embed_model) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (len(rows), len(genre_counts), diary_count, nodes, edges, EMBED_MODEL),
        )
        _stamp(
            con,
            {
                "pack": "core",
                "pack_version": PACK_VERSION,
                "generated": datetime.now(UTC).isoformat(timespec="seconds"),
                "embed_model": EMBED_MODEL,
            },
        )
        con.commit()
    finally:
        con.close()
    return _finalise(dest, PackStats(path=dest, passages=len(catalog)))

build_golden(packs, *, k, dtype, queries=GOLDEN_QUERIES, progress=None)

Record each golden query's top-k, as the packs themselves answer it.

The Swift engine is correct when it reproduces this file: same node ids in roughly the same order, same scores to two decimal places. Divergence localises to one of two places — the WordPiece tokenizer in front of the Core ML embedder, or the int8 quantisation — which is what makes the gate worth having.

Parameters:

Name	Type	Description	Default
packs	dict[str, Path]	{"gutenberg": path, "diaries": path}; missing keys skipped.	required
k	int	Hits to record per query.	required
dtype	str	The packs' vector dtype.	required
queries	Sequence[str]	Queries to record.	GOLDEN_QUERIES
progress		Optional status callable.	None

Returns:

Type	Description
dict	The golden document, ready to serialise.

Source code in src/gutenberg_kg/export_swift.py

def build_golden(
    packs: dict[str, Path],
    *,
    k: int,
    dtype: str,
    queries: Sequence[str] = GOLDEN_QUERIES,
    progress=None,
) -> dict:
    """Record each golden query's top-k, as the packs themselves answer it.

    The Swift engine is correct when it reproduces this file: same node ids in
    roughly the same order, same scores to two decimal places. Divergence
    localises to one of two places — the WordPiece tokenizer in front of the
    Core ML embedder, or the int8 quantisation — which is what makes the gate
    worth having.

    :param packs: ``{"gutenberg": path, "diaries": path}``; missing keys skipped.
    :param k: Hits to record per query.
    :param dtype: The packs' vector dtype.
    :param queries: Queries to record.
    :param progress: Optional status callable.
    :returns: The golden document, ready to serialise.
    """
    say = progress or (lambda _message: None)
    embedder = _make_embedder()
    say(f"  embedding {len(queries)} golden queries…")
    vectors = embedder.embed_texts(list(queries))

    def ranked(hits: Sequence[dict]) -> list[dict]:
        return [
            {"rank": rank, "node_id": hit["node_id"], "score": hit["score"]}
            for rank, hit in enumerate(hits)
        ]

    entries = []
    for query, vector in zip(queries, vectors, strict=False):
        record: dict = {"query": query, "packs": {}}
        per_pack: dict[str, list[dict]] = {}
        rescued: set[str] = set()
        for name, path in packs.items():
            hits, found = search_pack_with_provenance(path, vector, query, k=k)
            per_pack[name] = hits
            rescued |= found
            record["packs"][name] = ranked(hits)
        # The ``corpus=all`` ranking, from the same per-pack lists recorded
        # above, folded by the worker's own merge.  This is what makes the
        # golden file a reference for the cross-pack merge and not only for
        # each pack alone -- until it was recorded, nothing checked that the
        # app and the worker fold the two lists the same way.
        if "gutenberg" in per_pack and "diaries" in per_pack:
            record["all"] = ranked(
                merge_by_rank(
                    per_pack["gutenberg"],
                    per_pack["diaries"],
                    k,
                    rrf_k=RRF_K,
                    lexically_rescued=rescued,
                )
            )
        entries.append(record)

    return {
        "generated": datetime.now(UTC).isoformat(timespec="seconds"),
        "embed_model": EMBED_MODEL,
        "embed_dim": EMBED_DIM,
        "vector_dtype": dtype,
        "rrf_k": RRF_K,
        "k": k,
        # Recorded so the app can assert its own constant equals this one,
        # by name, rather than discover a drift as ranking symptoms.
        "rescue_tolerance": RESCUE_TOLERANCE,
        # max_rank_drift bounds how far a shared hit may move between the
        # reference and the Swift engine. It is not zero because the dense
        # channels differ in their last bits — numpy's argsort over
        # dequantised int8 against Accelerate's dot product — so near-ties
        # come back in either order, and a tie swapped is not a ranking
        # fault. It is small because losing the fused order altogether (say,
        # by re-sorting on cosine) moves a lexically-promoted hit much
        # further than a place or two.
        "tolerance": {"rank_overlap": 0.9, "score_delta": 0.02, "max_rank_drift": 2},
        "queries": entries,
    }

build_passage_pack(dest, sources, *, catalog, dtype='int8', max_chars=0, with_vectors=True, browse_window=BROWSE_WINDOW_TRIGGER_CHARS, progress=None)

Write a passage pack: content, an FTS5 index over it, and the vectors.

Passages are written first so SQLite assigns their rowids; vectors are then streamed in against those same rowids. A passage whose vector is missing from the source store is kept — it stays findable lexically and readable in Browse — and counted in missing_vectors so a partial store is visible rather than silent.

The pack id is <kg_name>:<node_id>, not the source node id verbatim. Diary node ids (chunk:entry_0000_chunk_0.md:0000) name the entry file within a book but not the book, and every diary numbers its entries from entry_0000 — so merging four diaries into one table keyed by the raw id silently dropped every later diary's colliding rows (4,601 of 27,462 passages, found 2026-09-01). The prefix restores uniqueness while keeping the id an opaque string, which is all the search code here and the Swift engine ever treat it as. The insert is a plain INSERT, so any residual collision fails the export instead of shrinking the pack.

Parameters:

Name	Type	Description	Default
dest	Path	Output path.	required
sources	Sequence[_PassageSource]	KGs to merge into this pack (one for books, four for diaries).	required
catalog	dict[str, dict]	Parsed catalog.json, for per-book title/author.	required
dtype	str	"int8" (default, ~3x smaller) or "float".	'int8'
max_chars	int	Per-passage truncation; 0 keeps passages whole.	0
with_vectors	bool	Skip the vector stage entirely when False.	True
progress		Optional callable(str) for status lines.	None

Returns:

Type	Description
PackStats	Stats for the finished pack.

Source code in src/gutenberg_kg/export_swift.py

def build_passage_pack(
    dest: Path,
    sources: Sequence[_PassageSource],
    *,
    catalog: dict[str, dict],
    dtype: str = "int8",
    max_chars: int = 0,
    with_vectors: bool = True,
    browse_window: int = BROWSE_WINDOW_TRIGGER_CHARS,
    progress=None,
) -> PackStats:
    """Write a passage pack: content, an FTS5 index over it, and the vectors.

    Passages are written first so SQLite assigns their rowids; vectors are then
    streamed in against those same rowids. A passage whose vector is missing
    from the source store is kept — it stays findable lexically and readable in
    Browse — and counted in ``missing_vectors`` so a partial store is visible
    rather than silent.

    The pack id is ``<kg_name>:<node_id>``, not the source node id verbatim.
    Diary node ids (``chunk:entry_0000_chunk_0.md:0000``) name the entry file
    within a book but not the book, and every diary numbers its entries from
    ``entry_0000`` — so merging four diaries into one table keyed by the raw
    id silently dropped every later diary's colliding rows (4,601 of 27,462
    passages, found 2026-09-01). The prefix restores uniqueness while keeping
    the id an opaque string, which is all the search code here and the Swift
    engine ever treat it as. The insert is a plain INSERT, so any residual
    collision fails the export instead of shrinking the pack.

    :param dest: Output path.
    :param sources: KGs to merge into this pack (one for books, four for diaries).
    :param catalog: Parsed ``catalog.json``, for per-book title/author.
    :param dtype: ``"int8"`` (default, ~3x smaller) or ``"float"``.
    :param max_chars: Per-passage truncation; 0 keeps passages whole.
    :param with_vectors: Skip the vector stage entirely when False.
    :param progress: Optional ``callable(str)`` for status lines.
    :returns: Stats for the finished pack.
    """
    say = progress or (lambda _message: None)
    diary_meta = _diary_meta()

    con = _fresh_db(dest, _PASSAGE_SCHEMA)
    rowid_by_id: dict[str, int] = {}
    stats = PackStats(path=dest)
    try:
        insert = (
            "INSERT INTO passages"
            "(id, kg_name, kg_kind, kind, name, title, node_title, author, genre, "
            " book, file_path, char_start, chapter, timestamp, content) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        )
        for source in sources:
            fixed = source.fixed_meta or diary_meta.get(source.kg_name)
            written = 0
            batch: list[tuple] = []
            for record in iter_source_passages(
                source.graph, max_chars=max_chars, allowed_keys=source.allowed_keys
            ):
                genre, book = split_source_path(record["file_path"])
                meta = catalog.get(f"{genre}/{book}", {}) if genre and book else {}
                batch.append(
                    (
                        f"{source.kg_name}:{record['id']}",
                        source.kg_name,
                        source.kg_kind,
                        record["kind"],
                        record["name"],
                        (fixed or {}).get("title") or meta.get("title") or book,
                        record["title"],
                        (fixed or {}).get("author") or meta.get("author"),
                        (fixed or {}).get("genre") or meta.get("genre") or genre,
                        book,
                        record["file_path"],
                        record["char_start"],
                        record["chapter"],
                        record["timestamp"],
                        record["text"],
                    )
                )
                if len(batch) >= 2000:
                    con.executemany(insert, batch)
                    written += len(batch)
                    batch = []
            if batch:
                con.executemany(insert, batch)
                written += len(batch)
            con.commit()
            stats.passages += written
            say(f"  {source.kg_name}: {written:,} passages")

        windowed = window_oversized_sections(con, trigger_chars=browse_window)
        if windowed:
            stats.passages += windowed
            say(f"  {windowed:,} synthetic Browse sections (split over {browse_window:,} chars)")

        # Synthetic sections are deliberately excluded: they have no source
        # vector, and counting them would report thousands of phantom misses.
        for node_id, rowid in con.execute(
            "SELECT id, rowid FROM passages WHERE id NOT LIKE ?", (f"%{_PART_MARKER}%",)
        ):
            rowid_by_id[node_id] = rowid

        say("  building the FTS5 index over clean passage text…")
        con.executescript(_FTS_SCHEMA)
        con.execute("INSERT INTO passages_fts(passages_fts) VALUES('rebuild')")
        con.commit()
        _stamp(
            con,
            {
                "pack": dest.stem,
                "pack_version": PACK_VERSION,
                "generated": datetime.now(UTC).isoformat(timespec="seconds"),
                "embed_model": EMBED_MODEL,
                "embed_dim": EMBED_DIM,
                "vector_dtype": dtype if with_vectors else "none",
                "rrf_k": RRF_K,
                "kgs": ",".join(s.kg_name for s in sources),
            },
        )
        con.commit()
    finally:
        con.close()

    if with_vectors:
        stats.sidecar, stats.vectors, stats.missing_vectors = write_vector_sidecar(
            dest, sources, rowid_by_id, dtype=dtype, progress=say
        )
        if stats.sidecar is not None:
            stats.sidecar_bytes = stats.sidecar.stat().st_size
            stats.sidecar_sha256 = _sha256(stats.sidecar)

    return _finalise(dest, stats)

encode_vector(vector, dtype)

Encode one vector for a vec0 column.

Vectors are L2-normalised before int8 quantisation. Cosine distance is scale-invariant, so this changes no ranking — but it does guarantee the * 127 scaling lands inside int8 range instead of clipping whatever the embedder happened to emit.

Parameters:

Name	Type	Description	Default
vector		A numpy float32 array.	required
dtype	str	"float" or "int8".	required

Returns:

Type	Description
bytes	The packed bytes for the column.

Source code in src/gutenberg_kg/export_swift.py

def encode_vector(vector, dtype: str) -> bytes:
    """Encode one vector for a ``vec0`` column.

    Vectors are L2-normalised before int8 quantisation. Cosine distance is
    scale-invariant, so this changes no ranking — but it does guarantee the
    ``* 127`` scaling lands inside int8 range instead of clipping whatever the
    embedder happened to emit.

    :param vector: A numpy float32 array.
    :param dtype: ``"float"`` or ``"int8"``.
    :returns: The packed bytes for the column.
    """
    numpy = _load_numpy()
    array = numpy.asarray(vector, dtype=numpy.float32)
    if dtype == "float":
        return array.tobytes()
    norm = float(numpy.linalg.norm(array))
    if norm > 0:
        array = array / norm
    return numpy.clip(numpy.round(array * 127.0), -128, 127).astype(numpy.int8).tobytes()

export_swift(options, *, progress=None)

Build the on-device packs from a bundle.

Parameters:

Name	Type	Description	Default
options	ExportOptions	Export inputs.	required
progress		Optional callable(str) for status lines.	None

Returns:

Type	Description
ExportReport	A report naming every file written.

Raises:

Type	Description
ExportError	On a missing bundle, an unreadable store, or an existing output directory without force.

Source code in src/gutenberg_kg/export_swift.py

def export_swift(options: ExportOptions, *, progress=None) -> ExportReport:
    """Build the on-device packs from a bundle.

    :param options: Export inputs.
    :param progress: Optional ``callable(str)`` for status lines.
    :returns: A report naming every file written.
    :raises ExportError: On a missing bundle, an unreadable store, or an
        existing output directory without ``force``.
    """
    say = progress or (lambda _message: None)
    started = time.perf_counter()

    if options.dtype not in ("int8", "float"):
        raise ExportError(f"dtype must be 'int8' or 'float', got {options.dtype!r}")

    bundle = locate_bundle(Path(options.bundle))
    out = options.resolved_out()
    if out.exists() and any(out.iterdir()):
        if not options.force:
            raise ExportError(f"{out} is not empty — pass --force to overwrite it.")
        # Wipe rather than write over: a spec whose book set shrank must not
        # leave the previous run's orphaned pack files sitting next to the
        # new, smaller ones.
        #
        # Only this export's own artifacts, though. `gutenkg export-embedder`
        # writes BGEEmbedder.mlpackage, vocab.txt and embedder.json into this
        # same directory by default, so clearing every child destroyed the
        # Core ML model the packs are useless without -- silently, and needing
        # a separate toolchain to rebuild. The orphan the wipe exists to catch
        # is always one of the names below.
        for child in out.iterdir():
            if child.is_file() and (
                child.suffix in (".pack", ".vectors")
                or child.name in ("manifest.json", "golden.json")
            ):
                child.unlink()
    out.mkdir(parents=True, exist_ok=True)

    catalog = load_catalog(bundle.catalog)
    if not catalog:
        say("  WARNING: no catalog.json — books will carry no title or author")

    if options.diary_dirs is not None:
        allowed_diary_slugs = frozenset(options.diary_dirs)
        diaries = tuple(d for d in bundle.diaries if d.slug in allowed_diary_slugs)
    else:
        diaries = bundle.diaries if options.include_diaries else ()
    report = ExportReport(out=out)

    book_count = len(options.catalog_keys) if options.catalog_keys is not None else len(catalog)
    say(f"core.pack  ← {book_count} books")
    report.packs.append(
        build_core_pack(
            out / "core.pack",
            bundle=bundle,
            catalog=catalog,
            diary_count=len(diaries),
            catalog_keys=options.catalog_keys,
        )
    )

    books_source = _PassageSource(
        graph=bundle.graph,
        kg_name="gutenberg",
        kg_kind="KGKind.GUTENBERG",
        vectors=bundle.vectors,
        allowed_keys=options.catalog_keys,
    )
    say("gutenberg.pack")
    report.packs.append(
        build_passage_pack(
            out / "gutenberg.pack",
            [books_source],
            catalog=catalog,
            dtype=options.dtype,
            max_chars=options.max_chars,
            with_vectors=options.with_vectors,
            progress=say,
        )
    )

    if diaries:
        say("diaries.pack")
        report.packs.append(
            build_passage_pack(
                out / "diaries.pack",
                [
                    _PassageSource(
                        graph=diary.graph,
                        kg_name=diary.slug,
                        kg_kind="KGKind.DIARY",
                        vectors=diary.vectors,
                    )
                    for diary in diaries
                ],
                catalog=catalog,
                dtype=options.dtype,
                max_chars=options.max_chars,
                with_vectors=options.with_vectors,
                progress=say,
            )
        )

    pack_paths = {pack.path.stem: pack.path for pack in report.packs if pack.path.stem != "core"}

    if options.golden and options.with_vectors:
        say("golden.json")
        report.golden = build_golden(
            pack_paths,
            k=options.golden_k,
            dtype=options.dtype,
            queries=options.golden_queries or GOLDEN_QUERIES,
            progress=say,
        )
        (out / "golden.json").write_text(
            json.dumps(report.golden, indent=2, ensure_ascii=False), encoding="utf-8"
        )

    if options.verify and options.with_vectors and bundle.has_vectors:
        say(f"verifying {options.dtype} recall against exact fp32 ground truth")
        report.verification = verify_pack(
            out / "gutenberg.pack",
            books_source,
            k=options.golden_k,
            dtype=options.dtype,
            progress=say,
        )

    report.elapsed_s = time.perf_counter() - started
    (out / "manifest.json").write_text(
        json.dumps(_manifest(report, bundle, options), indent=2, ensure_ascii=False),
        encoding="utf-8",
    )
    return report

fts_match_expression(query)

The any-term FTS5 expression — the recall fallback behind the phrase.

Every term is quoted and the terms are OR-ed, so an apostrophe or a question mark in the query cannot become FTS5 syntax — and a query is never rejected wholesale for one odd character.

Parameters:

Name	Type	Description	Default
query	str	The user's query text.	required

Returns:

Type	Description
str	An FTS5 MATCH expression, or "" when nothing is searchable.

Source code in src/gutenberg_kg/export_swift.py

def fts_match_expression(query: str) -> str:
    """The any-term FTS5 expression — the recall fallback behind the phrase.

    Every term is quoted and the terms are OR-ed, so an apostrophe or a
    question mark in the query cannot become FTS5 syntax — and a query is never
    rejected wholesale for one odd character.

    :param query: The user's query text.
    :returns: An FTS5 MATCH expression, or ``""`` when nothing is searchable.
    """
    return " OR ".join(f'"{term}"' for term in fts_terms(query))

fts_phrase_expression(query)

The exact-phrase FTS5 expression: every term, adjacent and in order.

Tried before :func:fts_match_expression, because a rare phrase diluted in a long chunk is exactly what the dense channel buries — "pillar of salt" ranks the Lot's-wife verse 604th on cosine alone, and only a phrase match recovers it. doc_kg.store.GraphStore.search_lexical does the same two steps in the same order; this is the pack's copy of that behaviour.

Parameters:

Name	Type	Description	Default
query	str	The user's query text.	required

Returns:

Type	Description
str	A quoted phrase expression, or "" when nothing is searchable.

Source code in src/gutenberg_kg/export_swift.py

def fts_phrase_expression(query: str) -> str:
    """The exact-phrase FTS5 expression: every term, adjacent and in order.

    Tried *before* :func:`fts_match_expression`, because a rare phrase diluted
    in a long chunk is exactly what the dense channel buries — "pillar of salt"
    ranks the Lot's-wife verse 604th on cosine alone, and only a phrase match
    recovers it. ``doc_kg.store.GraphStore.search_lexical`` does the same two
    steps in the same order; this is the pack's copy of that behaviour.

    :param query: The user's query text.
    :returns: A quoted phrase expression, or ``""`` when nothing is searchable.
    """
    terms = fts_terms(query)
    return '"' + " ".join(terms) + '"' if terms else ""

fts_terms(query)

Tokenise a query into bare alphanumeric terms for FTS5.

Strips apostrophes, quotes and punctuation that would otherwise be read as FTS5 query syntax, so Lot's becomes lot, s. Mirrors doc_kg.store._fts_terms.

Parameters:

Name	Type	Description	Default
query	str	The user's query text.	required

Returns:

Type	Description
list[str]	The searchable terms, in order; empty when nothing is searchable.

Source code in src/gutenberg_kg/export_swift.py

def fts_terms(query: str) -> list[str]:
    """Tokenise a query into bare alphanumeric terms for FTS5.

    Strips apostrophes, quotes and punctuation that would otherwise be read as
    FTS5 query syntax, so ``Lot's`` becomes ``lot``, ``s``. Mirrors
    ``doc_kg.store._fts_terms``.

    :param query: The user's query text.
    :returns: The searchable terms, in order; empty when nothing is searchable.
    """
    return [t for t in ("".join(c if c.isalnum() else " " for c in query)).split() if t]

iter_source_passages(graph, *, max_chars=0, allowed_keys=None)

Yield searchable passages from a graph.sqlite.

Applies the worker's own scope — kind IN ('chunk','section'), and never a book's reference.md metadata sheet — so the pack holds the same set the served query ranks.

Empty chunks are dropped. Empty sections are not: they carry no prose but they are the chapter markers get_chapters lists and get_chapter slices between, so dropping them would take the Browse tab with them.

:yields: Row dicts keyed by the columns in :data:_WANTED_NODE_COLUMNS, with absent columns set to None.

Parameters:

Name	Type	Description	Default
graph	Path	Path to a DocKG/DiaryKG graph.sqlite.	required
max_chars	int	Truncate content to this many characters, at a word boundary; 0 keeps the passage whole.	0
allowed_keys	frozenset[str] | None	Restrict to nodes whose file_path starts with one of these "<genre>/<book>" keys. None keeps every book — callers pass this for a book-filtered bundle export, never for a diary source (diaries are scoped by which sources are passed in at all, not by this filter).	None

Source code in src/gutenberg_kg/export_swift.py

def iter_source_passages(
    graph: Path,
    *,
    max_chars: int = 0,
    allowed_keys: frozenset[str] | None = None,
) -> Iterator[dict]:
    """Yield searchable passages from a ``graph.sqlite``.

    Applies the worker's own scope — ``kind IN ('chunk','section')``, and never
    a book's ``reference.md`` metadata sheet — so the pack holds the same set
    the served query ranks.

    Empty chunks are dropped. Empty *sections* are not: they carry no prose but
    they are the chapter markers ``get_chapters`` lists and ``get_chapter``
    slices between, so dropping them would take the Browse tab with them.

    :param graph: Path to a DocKG/DiaryKG ``graph.sqlite``.
    :param max_chars: Truncate ``content`` to this many characters, at a word
        boundary; 0 keeps the passage whole.
    :param allowed_keys: Restrict to nodes whose ``file_path`` starts with one
        of these ``"<genre>/<book>"`` keys. ``None`` keeps every book —
        callers pass this for a book-filtered bundle export, never for a
        diary source (diaries are scoped by which sources are passed in at
        all, not by this filter).
    :yields: Row dicts keyed by the columns in :data:`_WANTED_NODE_COLUMNS`,
        with absent columns set to ``None``.
    """
    con = sqlite3.connect(f"file:{graph}?mode=ro", uri=True)
    try:
        present = table_columns(con, "nodes")
        if not present:
            raise ExportError(f"{graph} has no `nodes` table")
        if "text" not in present:
            raise ExportError(
                f"{graph} has no `nodes.text` column — this store predates the "
                "clean-content schema and cannot be exported."
            )
        selected = [c for c in _WANTED_NODE_COLUMNS if c in present]

        kinds = ", ".join("?" for _ in SEARCHED_KINDS)
        sql = f"SELECT {', '.join(selected)} FROM nodes WHERE kind IN ({kinds})"
        params: list[object] = list(SEARCHED_KINDS)
        if "file_path" in present:
            sql += " AND (file_path IS NULL OR file_path NOT LIKE '%reference.md')"
            if allowed_keys is not None:
                scope_sql, scope_params = _file_path_scope(allowed_keys)
                sql += f" AND {scope_sql}"
                params.extend(scope_params)
        sql += " ORDER BY rowid"

        con.row_factory = sqlite3.Row
        for row in con.execute(sql, params):
            record = {column: None for column in _WANTED_NODE_COLUMNS}
            for column in selected:
                record[column] = row[column]
            content = (record["text"] or "").strip()
            # A chunk with no text is nothing; a section with no text is still
            # a chapter marker, and Browse builds its chapter list from those.
            if not content and record["kind"] != "section":
                continue
            record["text"] = _truncate(content, max_chars) if max_chars else content
            yield record
    finally:
        con.close()

iter_source_vectors(vectors, *, allowed_keys=None)

Stream (node_id, vector) from the bundle's sqlite-vec store.

Streams rather than materialising: the consolidated store holds 688 K vectors, and only the ~364 K searchable ones end up in the pack.

:yields: (id, numpy float32 array) pairs.

Parameters:

Name	Type	Description	Default
vectors	Path | None	The vectors.sqlite store, or None.	required
allowed_keys	frozenset[str] | None	Restrict to vec_meta rows whose file_path starts with one of these "<genre>/<book>" keys, applied alongside kind IN ('chunk','section') so enrichment-kind rows (topic, entity, keyword, document) never reach the scan either.	None

Raises:

Type	Description
ExportError	If there is no store.

Source code in src/gutenberg_kg/export_swift.py

def iter_source_vectors(
    vectors: Path | None,
    *,
    allowed_keys: frozenset[str] | None = None,
) -> Iterator[tuple[str, object]]:
    """Stream ``(node_id, vector)`` from the bundle's sqlite-vec store.

    Streams rather than materialising: the consolidated store holds 688 K
    vectors, and only the ~364 K searchable ones end up in the pack.

    :param vectors: The ``vectors.sqlite`` store, or None.
    :param allowed_keys: Restrict to ``vec_meta`` rows whose ``file_path``
        starts with one of these ``"<genre>/<book>"`` keys, applied alongside
        ``kind IN ('chunk','section')`` so enrichment-kind rows (topic,
        entity, keyword, document) never reach the scan either.
    :yields: ``(id, numpy float32 array)`` pairs.
    :raises ExportError: If there is no store.
    """
    numpy = _load_numpy()
    if vectors is not None:
        con = _connect_with_vec(vectors, read_only=True)
        try:
            scope_sql, scope_params = _file_path_scope(allowed_keys, column="vec_meta.file_path")
            kinds = ", ".join(f"'{kind}'" for kind in SEARCHED_KINDS)
            sql = (
                "SELECT vec_meta.id, vec_nodes.embedding "
                "FROM vec_nodes JOIN vec_meta ON vec_meta.rowid = vec_nodes.rowid "
                f"WHERE vec_meta.kind IN ({kinds}) AND {scope_sql}"
                if allowed_keys is not None
                else "SELECT vec_meta.id, vec_nodes.embedding "
                "FROM vec_nodes JOIN vec_meta ON vec_meta.rowid = vec_nodes.rowid"
            )
            rows = con.execute(sql, scope_params if allowed_keys is not None else [])
            for node_id, blob in rows:
                yield node_id, numpy.frombuffer(blob, dtype=numpy.float32)
        finally:
            con.close()
        return

    raise ExportError(
        "no vector store found in the bundle (.dockg/vectors.sqlite); a lancedb/ "
        "store is no longer read -- rebuild it with `gutenkg build-corpus`"
    )

load_catalog(path)

Load catalog.json: "<genre>/<book>" -> {genre, title, author, ...}.

Parameters:

Name	Type	Description	Default
path	Path | None	Path to the catalog, or None.	required

Returns:

Type	Description
dict[str, dict]	The catalog, or {} when it is missing or unreadable.

Source code in src/gutenberg_kg/export_swift.py

def load_catalog(path: Path | None) -> dict[str, dict]:
    """Load ``catalog.json``: ``"<genre>/<book>" -> {genre, title, author, ...}``.

    :param path: Path to the catalog, or None.
    :returns: The catalog, or ``{}`` when it is missing or unreadable.
    """
    if path is None or not path.exists():
        return {}
    try:
        with open(path, encoding="utf-8") as handle:
            data = json.load(handle)
        return data if isinstance(data, dict) else {}
    except (OSError, json.JSONDecodeError):
        return {}

locate_bundle(root)

Resolve a bundle directory into the files the export reads.

Parameters:

Name	Type	Description	Default
root	Path	Bundle root, e.g. bundles/gutenberg-all.	required

Returns:

Type	Description
BundlePaths	Resolved paths, with diaries discovered under diaries/.

Raises:

Type	Description
ExportError	If the root or its consolidated graph.sqlite is missing — the one failure worth reporting before any work starts.

Source code in src/gutenberg_kg/export_swift.py

def locate_bundle(root: Path) -> BundlePaths:
    """Resolve a bundle directory into the files the export reads.

    :param root: Bundle root, e.g. ``bundles/gutenberg-all``.
    :returns: Resolved paths, with diaries discovered under ``diaries/``.
    :raises ExportError: If the root or its consolidated ``graph.sqlite`` is
        missing — the one failure worth reporting before any work starts.
    """
    root = Path(root)
    if not root.is_dir():
        raise ExportError(
            f"bundle not found: {root}\nBuild one first with `make build-corpus`, or pass --bundle."
        )
    dockg = root / ".dockg"
    graph = dockg / "graph.sqlite"
    if not graph.exists():
        raise ExportError(f"no consolidated DocKG at {graph} — is {root} a bundle root?")

    vectors = _resolve_vectors_path(dockg)
    catalog = dockg / "catalog.json"

    diaries: list[DiarySource] = []
    diaries_root = root / "diaries"
    if diaries_root.is_dir():
        for directory in sorted(p for p in diaries_root.iterdir() if p.is_dir()):
            store = directory / ".diarykg"
            diary_graph = store / "graph.sqlite"
            if not diary_graph.exists():
                continue
            diaries.append(
                DiarySource(
                    slug=_diary_slug(directory.name),
                    graph=diary_graph,
                    vectors=_resolve_vectors_path(store),
                )
            )

    return BundlePaths(
        root=root,
        graph=graph,
        catalog=catalog if catalog.exists() else None,
        vectors=vectors,
        diaries=tuple(diaries),
    )

read_vector_sidecar(sidecar)

Memory-map a sidecar and return (matrix, dtype).

Parameters:

Name	Type	Description	Default
sidecar	Path	A .vectors file.	required

Returns:

Type	Description
	An (count, dim) numpy array viewing the file, and its dtype name.

Raises:

Type	Description
ExportError	If the magic, dimension or length do not agree — a truncated download is the failure this catches.

Source code in src/gutenberg_kg/export_swift.py

def read_vector_sidecar(sidecar: Path):
    """Memory-map a sidecar and return ``(matrix, dtype)``.

    :param sidecar: A ``.vectors`` file.
    :returns: An ``(count, dim)`` numpy array viewing the file, and its dtype
        name.
    :raises ExportError: If the magic, dimension or length do not agree — a
        truncated download is the failure this catches.
    """
    numpy = _load_numpy()
    with open(sidecar, "rb") as handle:
        header = handle.read(VECTOR_HEADER_BYTES)
    if len(header) < VECTOR_HEADER_BYTES or header[0:8] != VECTOR_MAGIC:
        raise ExportError(f"{sidecar} is not a GutenbergKG vector sidecar")
    codes = {code: name for name, code in _DTYPE_CODES.items()}
    dtype = codes.get(header[8])
    if dtype is None:
        raise ExportError(f"{sidecar} has an unknown vector dtype ({header[8]})")
    dim = int.from_bytes(header[12:16], "little")
    count = int.from_bytes(header[16:24], "little")
    expected = VECTOR_HEADER_BYTES + count * dim * _DTYPE_ITEMSIZE[dtype]
    actual = sidecar.stat().st_size
    if actual != expected:
        raise ExportError(
            f"{sidecar} is {actual} bytes, expected {expected} for "
            f"{count} x {dim} {dtype} — truncated?"
        )
    numpy_dtype = numpy.int8 if dtype == "int8" else numpy.float32
    matrix = numpy.memmap(
        sidecar, dtype=numpy_dtype, mode="r", offset=VECTOR_HEADER_BYTES, shape=(count, dim)
    )
    return matrix, dtype

resolve_catalog_selectors(*, books, genres, catalog)

Resolve --book/--genre selectors against a built bundle's catalog.

A sibling to bundle_spec.resolve_selection, deliberately not shared with it: that resolver reads corpus/<genre>/<book>/reference.md from the source tree used to build a bundle, while export-swift --book filters a bundle that may be all that is present — a downloaded bundle with no corpus/ beside it still has to work. Both apply the same three rules in the same order (explicit key, ebook_id, unique bare name), just against different data.

Parameters:

Name	Type	Description	Default
books	Sequence[str]	Raw --book values.	required
genres	Sequence[str]	Raw --genre values.	required
catalog	dict[str, dict]	Parsed catalog.json.	required

Returns:

Type	Description
tuple[frozenset[str], tuple[str, ...]]	(catalog_keys, missing) — missing holds every selector that did not resolve to exactly one book or a non-empty genre.

Source code in src/gutenberg_kg/export_swift.py

def resolve_catalog_selectors(
    *, books: Sequence[str], genres: Sequence[str], catalog: dict[str, dict]
) -> tuple[frozenset[str], tuple[str, ...]]:
    """Resolve ``--book``/``--genre`` selectors against a *built* bundle's catalog.

    A sibling to ``bundle_spec.resolve_selection``, deliberately not shared
    with it: that resolver reads ``corpus/<genre>/<book>/reference.md`` from
    the source tree used to *build* a bundle, while ``export-swift --book``
    filters a bundle that may be all that is present — a downloaded bundle
    with no ``corpus/`` beside it still has to work. Both apply the same
    three rules in the same order (explicit key, ebook_id, unique bare name),
    just against different data.

    :param books: Raw ``--book`` values.
    :param genres: Raw ``--genre`` values.
    :param catalog: Parsed ``catalog.json``.
    :returns: ``(catalog_keys, missing)`` — ``missing`` holds every selector
        that did not resolve to exactly one book or a non-empty genre.
    """
    keys: set[str] = set()
    missing: list[str] = []

    for genre in genres:
        matched = [
            key
            for key, meta in catalog.items()
            if (meta.get("genre") or key.split("/", 1)[0]) == genre
        ]
        if not matched:
            missing.append(genre)
        keys.update(matched)

    for selector in books:
        if selector in catalog:
            keys.add(selector)
            continue
        if selector.isdigit():
            ebook_id = int(selector)
            found = [key for key, meta in catalog.items() if meta.get("ebook_id") == ebook_id]
        else:
            found = [key for key in catalog if key.split("/", 1)[-1] == selector]
        if len(found) == 1:
            keys.add(found[0])
        else:
            missing.append(selector)

    return frozenset(keys), tuple(missing)

rrf_fuse(dense_ids, lexical_ids, k)

Blend two rank lists with reciprocal rank fusion.

Byte-for-byte the arithmetic in handler._rrf_fuse, at the same :data:RRF_K. A hit ranked well by either channel floats up; one ranked by both wins.

Parameters:

Name	Type	Description	Default
dense_ids	Sequence[str]	Ids best-first by cosine.	required
lexical_ids	Sequence[str]	Ids best-first by BM25.	required
k	int	How many fused ids to return.	required

Returns:

Type	Description
list[str]	Fused ids, best-first.

Source code in src/gutenberg_kg/export_swift.py

def rrf_fuse(dense_ids: Sequence[str], lexical_ids: Sequence[str], k: int) -> list[str]:
    """Blend two rank lists with reciprocal rank fusion.

    Byte-for-byte the arithmetic in ``handler._rrf_fuse``, at the same
    :data:`RRF_K`. A hit ranked well by either channel floats up; one ranked by
    both wins.

    :param dense_ids: Ids best-first by cosine.
    :param lexical_ids: Ids best-first by BM25.
    :param k: How many fused ids to return.
    :returns: Fused ids, best-first.
    """
    scores: dict[str, float] = {}
    # Lexical first, so a cross-channel tie goes to the exact match.
    for rank, node_id in enumerate(lexical_ids):
        scores[node_id] = scores.get(node_id, 0.0) + 1.0 / (RRF_K + rank)
    for rank, node_id in enumerate(dense_ids):
        scores[node_id] = scores.get(node_id, 0.0) + 1.0 / (RRF_K + rank)
    return sorted(scores, key=lambda i: -scores[i])[:k]

search_pack(pack, query_vector, query_text, *, k=10, genre=None, min_score=0.0, lexical=True)

Run the packed corpus's own hybrid search — dense + BM25, fused by RRF.

:func:search_pack_with_provenance without the rescued set; see it for the parameters.

Source code in src/gutenberg_kg/export_swift.py

def search_pack(
    pack: Path,
    query_vector,
    query_text: str,
    *,
    k: int = 10,
    genre: str | None = None,
    min_score: float = 0.0,
    lexical: bool = True,
) -> list[dict]:
    """Run the packed corpus's own hybrid search — dense + BM25, fused by RRF.

    :func:`search_pack_with_provenance` without the rescued set; see it for the
    parameters.
    """
    return search_pack_with_provenance(
        pack, query_vector, query_text, k=k, genre=genre, min_score=min_score, lexical=lexical
    )[0]

search_pack_with_provenance(pack, query_vector, query_text, *, k=10, genre=None, min_score=0.0, lexical=True)

Run the packed corpus's own hybrid search — dense + BM25, fused by RRF.

A direct translation of handler._semantic_search onto the pack: same k * 3 oversampling in both channels, same :data:RRF_K, same round(1 - distance, 4) score, and the same hydrate-missing-cosine step so every fused hit carries an honest score.

The dense channel is a dot product over the memory-mapped sidecar rather than a vec0 query, which is what the Swift engine does with Accelerate. Both are exhaustive, so this is the same ranking by a simpler route.

Parameters:

Name	Type	Description	Default
pack	Path	A passage pack (its sidecar sits beside it).	required
query_vector		The query embedding (numpy float32, 384-d).	required
query_text	str	The raw query, for the lexical channel.	required
k	int	Hits to return.	10
genre	str | None	Restrict to one genre, as the worker's genre_filter does.	None
min_score	float	Drop hits below this cosine score.	0.0
lexical	bool	Include the BM25 channel. False gives the dense channel alone, which is what :func:verify_pack compares against dense ground truth — fusing BM25 into one side of that comparison made the quantisation gate report hybrid-vs-dense divergence as recall loss.	True

Returns:

Type	Description
tuple[list[dict], set[str]]	Hit dicts in the worker's shape, best-first, and the node ids among them the lexical channel rescued -- found by BM25 outside the dense top k. :func:build_golden needs the second to record the corpus=all merge the way the worker and the app perform it.

Source code in src/gutenberg_kg/export_swift.py

def search_pack_with_provenance(
    pack: Path,
    query_vector,
    query_text: str,
    *,
    k: int = 10,
    genre: str | None = None,
    min_score: float = 0.0,
    lexical: bool = True,
) -> tuple[list[dict], set[str]]:
    """Run the packed corpus's own hybrid search — dense + BM25, fused by RRF.

    A direct translation of ``handler._semantic_search`` onto the pack: same
    ``k * 3`` oversampling in both channels, same :data:`RRF_K`, same
    ``round(1 - distance, 4)`` score, and the same hydrate-missing-cosine step
    so every fused hit carries an honest score.

    The dense channel is a dot product over the memory-mapped sidecar rather
    than a vec0 query, which is what the Swift engine does with Accelerate.
    Both are exhaustive, so this is the same ranking by a simpler route.

    :param pack: A passage pack (its sidecar sits beside it).
    :param query_vector: The query embedding (numpy float32, 384-d).
    :param query_text: The raw query, for the lexical channel.
    :param k: Hits to return.
    :param genre: Restrict to one genre, as the worker's ``genre_filter`` does.
    :param min_score: Drop hits below this cosine score.
    :param lexical: Include the BM25 channel. False gives the dense channel
        alone, which is what :func:`verify_pack` compares against dense
        ground truth — fusing BM25 into one side of that comparison made the
        quantisation gate report hybrid-vs-dense divergence as recall loss.
    :returns: Hit dicts in the worker's shape, best-first, and the node ids
        among them the lexical channel rescued -- found by BM25 outside the
        dense top ``k``.  :func:`build_golden` needs the second to record the
        ``corpus=all`` merge the way the worker and the app perform it.
    """
    numpy = _load_numpy()
    con = sqlite3.connect(f"file:{pack}?mode=ro", uri=True)
    con.row_factory = sqlite3.Row
    try:
        scope = _scope_clause(genre=genre)
        params: list[object] = [genre] if genre else []

        dense_ids: list[str] = []
        distance_by_id: dict[str, float] = {}
        sidecar = pack.with_suffix(".vectors")
        if sidecar.exists():
            matrix, dtype = read_vector_sidecar(sidecar)
            rows = con.execute(
                f"SELECT id, vector_index FROM passages WHERE vector_index IS NOT NULL AND {scope}",
                params,
            ).fetchall()
            if rows:
                ids = [row["id"] for row in rows]
                indices = numpy.fromiter(
                    (row["vector_index"] for row in rows), dtype=numpy.int64, count=len(rows)
                )
                # int8 rows are unit vectors scaled by 127; the scale cancels
                # once the similarity is renormalised, so ranking is unaffected.
                candidates = numpy.asarray(matrix[indices], dtype=numpy.float32)
                norms = numpy.linalg.norm(candidates, axis=1).clip(min=1e-12)
                query = numpy.asarray(query_vector, dtype=numpy.float32)
                query = query / max(float(numpy.linalg.norm(query)), 1e-12)
                similarity = (candidates @ query) / norms
                order = numpy.argsort(-similarity)[: k * 3]
                dense_ids = [ids[i] for i in order]
                distance_by_id = {ids[i]: 1.0 - float(similarity[i]) for i in order}
                # Keep every candidate's score available for the lexical
                # hydration step below, without re-reading the sidecar.
                _all_scores = {ids[i]: float(similarity[i]) for i in range(len(ids))}
            else:
                _all_scores = {}
        else:
            _all_scores = {}

        lexical_ids: list[str] = []
        if lexical:
            lex_sql = (
                "SELECT p.id AS id FROM passages_fts f "
                "JOIN passages p ON p.rowid = f.rowid "
                f"WHERE passages_fts MATCH ? AND {_scope_clause(genre=genre, prefix='p.')} "
                "ORDER BY bm25(passages_fts) LIMIT ?"
            )

            def run_lexical(expression: str) -> list[str]:
                if not expression:
                    return []
                try:
                    return [r["id"] for r in con.execute(lex_sql, [expression, *params, k * 3])]
                except sqlite3.Error:
                    return []

            # Phrase first, then any-term for recall — the order
            # ``GraphStore.search_lexical`` uses. A phrase hit is a literal
            # match and deserves to seed the fusion; the OR form is what keeps
            # a query with no adjacent match from returning nothing at all.
            lexical_ids = run_lexical(fts_phrase_expression(query_text))
            if not lexical_ids:
                lexical_ids = run_lexical(fts_match_expression(query_text))

        # Hydrate cosine for lexical-only ids so every fused hit carries an
        # honest score — handler._semantic_search does exactly this.
        for node_id in lexical_ids:
            if node_id not in distance_by_id and node_id in _all_scores:
                distance_by_id[node_id] = 1.0 - _all_scores[node_id]
        lexical_ids = [i for i in lexical_ids if i in distance_by_id]

        ordered = rrf_fuse(dense_ids, lexical_ids, k) if lexical_ids else dense_ids[:k]
        # What the dense channel would have returned on its own.  A lexical hit
        # outside it is one cosine did not find -- the definition
        # ``LocalRetrieval.search`` uses, so the golden ``all`` ranking and the
        # app's are built from the same set.
        dense_top = set(dense_ids[:k])
        rescued = {i for i in lexical_ids if i not in dense_top}
        if not ordered:
            return [], set()

        placeholders = ", ".join("?" for _ in ordered)
        by_id = {
            row["id"]: dict(row)
            for row in con.execute(f"SELECT * FROM passages WHERE id IN ({placeholders})", ordered)
        }
        hits = []
        for node_id in ordered:
            row = by_id.get(node_id)
            if row is None:
                continue
            score = round(1.0 - distance_by_id.get(node_id, 1.0), 4)
            if score < min_score:
                continue
            hits.append(
                {
                    "kg_name": row["kg_name"],
                    "kg_kind": row["kg_kind"],
                    "node_id": node_id,
                    "name": row["name"] or row["title"] or "",
                    "kind": row["kind"],
                    "score": score,
                    "summary": row["content"],
                    "source_path": row["file_path"] or "",
                    "content": row["content"],
                    "timestamp": row["timestamp"],
                    "genre": row["genre"],
                    "title": row["title"],
                    "author": row["author"],
                }
            )
        return hits, rescued & {h["node_id"] for h in hits}
    finally:
        con.close()

split_source_path(file_path)

Split a node file_path into (genre, book).

Paths are "<genre>/<book>/<document>.md". This is the same split handler._enrich_catalog does to key into the catalog.

Parameters:

Name	Type	Description	Default
file_path	str | None	A node's file path, or None.	required

Returns:

Type	Description
tuple[str | None, str | None]	(genre, book); either may be None for an unexpected shape.

Source code in src/gutenberg_kg/export_swift.py

def split_source_path(file_path: str | None) -> tuple[str | None, str | None]:
    """Split a node ``file_path`` into ``(genre, book)``.

    Paths are ``"<genre>/<book>/<document>.md"``. This is the same split
    ``handler._enrich_catalog`` does to key into the catalog.

    :param file_path: A node's file path, or None.
    :returns: ``(genre, book)``; either may be None for an unexpected shape.
    """
    if not file_path:
        return None, None
    parts = file_path.split("/")
    if len(parts) < 2:
        return None, None
    return parts[0], parts[1]

table_columns(con, table)

Column names present on table.

Every read in this module is shaped by this rather than by a hardcoded column list: DocKG and DiaryKG differ, and both change as they are rebuilt.

Parameters:

Name	Type	Description	Default
con	Connection	An open connection.	required
table	str	Table name.	required

Returns:

Type	Description
set[str]	The column names, or an empty set when the table is absent.

Source code in src/gutenberg_kg/export_swift.py

def table_columns(con: sqlite3.Connection, table: str) -> set[str]:
    """Column names present on *table*.

    Every read in this module is shaped by this rather than by a hardcoded
    column list: DocKG and DiaryKG differ, and both change as they are rebuilt.

    :param con: An open connection.
    :param table: Table name.
    :returns: The column names, or an empty set when the table is absent.
    """
    try:
        return {row[1] for row in con.execute(f"PRAGMA table_info({table})")}
    except sqlite3.Error:
        return set()

verify_pack(pack, source, *, k, dtype, queries=GOLDEN_QUERIES, progress=None)

Measure what quantisation cost, against exact fp32 ground truth.

Computes the true top-k by brute force over the source vectors — the same method benchmarks/bench_sqlite_vec.py uses — and compares it with what the pack's dense channel returns (search_pack(..., lexical=False); the BM25 channel is deliberately off, because ground truth here is dense-only and fusing lexical hits into one side reported the divergence as recall loss — measured at 0.567 apparent vs 0.958 actual on 2026-09-01). This is a build-time gate: an int8 recall that has quietly fallen is something to see here, not in a user's answer.

Parameters:

Name	Type	Description	Default
pack	Path	The finished pack.	required
source	_PassageSource	The KG whose source vectors are the ground truth.	required
k	int	Depth to compare at.	required
dtype	str	The pack's vector dtype.	required
queries	Sequence[str]	Queries to check.	GOLDEN_QUERIES
progress		Optional status callable.	None

Returns:

Type	Description
dict	{"recall_at_k", "mean_score_delta", "queries": [...]}.

Source code in src/gutenberg_kg/export_swift.py

def verify_pack(
    pack: Path,
    source: _PassageSource,
    *,
    k: int,
    dtype: str,
    queries: Sequence[str] = GOLDEN_QUERIES,
    progress=None,
) -> dict:
    """Measure what quantisation cost, against exact fp32 ground truth.

    Computes the true top-k by brute force over the *source* vectors — the same
    method ``benchmarks/bench_sqlite_vec.py`` uses — and compares it with what
    the pack's dense channel returns (``search_pack(..., lexical=False)``; the
    BM25 channel is deliberately off, because ground truth here is dense-only
    and fusing lexical hits into one side reported the divergence as recall
    loss — measured at 0.567 apparent vs 0.958 actual on 2026-09-01). This is
    a build-time gate: an int8 recall that has quietly fallen is something to
    see here, not in a user's answer.

    :param pack: The finished pack.
    :param source: The KG whose source vectors are the ground truth.
    :param k: Depth to compare at.
    :param dtype: The pack's vector dtype.
    :param queries: Queries to check.
    :param progress: Optional status callable.
    :returns: ``{"recall_at_k", "mean_score_delta", "queries": [...]}``.
    """
    say = progress or (lambda _message: None)
    numpy = _load_numpy()
    embedder = _make_embedder()

    say("  loading source vectors for exact ground truth…")
    ids: list[str] = []
    rows: list = []
    for node_id, vector in iter_source_vectors(source.vectors, allowed_keys=source.allowed_keys):
        ids.append(f"{source.kg_name}:{node_id}")
        rows.append(vector)
    if not rows:
        raise ExportError("no source vectors to verify against")
    matrix = numpy.vstack(rows).astype(numpy.float32)
    matrix /= numpy.linalg.norm(matrix, axis=1, keepdims=True).clip(min=1e-12)

    # Ground truth is over the pack's own id set, so a passage the pack
    # deliberately omits never counts as a miss.
    con = sqlite3.connect(f"file:{pack}?mode=ro", uri=True)
    try:
        packed_ids = {row[0] for row in con.execute("SELECT id FROM passages")}
    finally:
        con.close()
    keep = [i for i, node_id in enumerate(ids) if node_id in packed_ids]
    matrix = matrix[keep]
    ids = [ids[i] for i in keep]

    vectors = embedder.embed_texts(list(queries))
    recalls: list[float] = []
    deltas: list[float] = []
    per_query = []
    for query, vector in zip(queries, vectors, strict=False):
        qvec = numpy.asarray(vector, dtype=numpy.float32)
        qvec = qvec / max(float(numpy.linalg.norm(qvec)), 1e-12)
        exact_scores = matrix @ qvec
        top = numpy.argsort(-exact_scores)[:k]
        expected = {ids[i]: float(exact_scores[i]) for i in top}

        got = search_pack(pack, vector, query, k=k, lexical=False)
        got_ids = [hit["node_id"] for hit in got]
        overlap = len(set(got_ids) & set(expected)) / max(len(expected), 1)
        recalls.append(overlap)

        query_deltas = [
            abs(hit["score"] - expected[hit["node_id"]])
            for hit in got
            if hit["node_id"] in expected
        ]
        deltas.extend(query_deltas)
        per_query.append(
            {
                "query": query,
                "recall": round(overlap, 4),
                "mean_score_delta": round(sum(query_deltas) / len(query_deltas), 5)
                if query_deltas
                else None,
            }
        )
        say(f"  {overlap:.2f}  {query}")

    return {
        "k": k,
        "vector_dtype": dtype,
        "recall_at_k": round(sum(recalls) / len(recalls), 4) if recalls else 0.0,
        "mean_score_delta": round(sum(deltas) / len(deltas), 5) if deltas else None,
        "queries": per_query,
    }

window_oversized_sections(con, *, trigger_chars, part_chars=BROWSE_WINDOW_PART_CHARS)

Split sections too long to read as one chapter into synthetic parts.

Browse lists kind='section' rows by char_start and slices the chunks between consecutive markers, so a book whose whole text sits under one heading renders as a single scroll — 373 K characters for Franklin's Autobiography, 1.2 M for Boswell's Johnson. Works like those have no divisions the converter can find, so the markers have to be manufactured here.

Only sections past trigger_chars are touched, and those are divided into equal parts of roughly part_chars, cut at chunk boundaries. The two numbers have to differ: cutting everything down to part size fragments the real chapters of books that were already correct, which is a worse result than the monolith it fixes.

Measuring in characters rather than chunks keeps a part the same length to read across genres, since chunk sizes vary (a tenth are under 268 characters, a tenth over 510) and the verse chunker's are smaller again.

The extra rows carry no content and get no vector, which keeps this to the Browse tab: retrieval ranks chunks, and a marker with no text and no embedding can never be a hit. The parent section keeps its id, so anything holding one still resolves.

Parameters:

Name	Type	Description	Default
con	Connection	Open connection to a passage pack.	required
trigger_chars	int	Only sections longer than this are split. Values <= 0 disable windowing.	required
part_chars	int	Length each part aims for.	BROWSE_WINDOW_PART_CHARS

Returns:

Type	Description
int	Number of synthetic sections inserted.

Source code in src/gutenberg_kg/export_swift.py

def window_oversized_sections(
    con: sqlite3.Connection,
    *,
    trigger_chars: int,
    part_chars: int = BROWSE_WINDOW_PART_CHARS,
) -> int:
    """Split sections too long to read as one chapter into synthetic parts.

    Browse lists ``kind='section'`` rows by ``char_start`` and slices the chunks
    between consecutive markers, so a book whose whole text sits under one
    heading renders as a single scroll — 373 K characters for Franklin's
    Autobiography, 1.2 M for Boswell's *Johnson*. Works like those have no
    divisions the converter can find, so the markers have to be manufactured
    here.

    Only sections past *trigger_chars* are touched, and those are divided into
    equal parts of roughly *part_chars*, cut at chunk boundaries. The two
    numbers have to differ: cutting everything down to part size fragments the
    real chapters of books that were already correct, which is a worse result
    than the monolith it fixes.

    Measuring in characters rather than chunks keeps a part the same length to
    read across genres, since chunk sizes vary (a tenth are under 268
    characters, a tenth over 510) and the verse chunker's are smaller again.

    The extra rows carry no content and get no vector, which keeps this to the
    Browse tab: retrieval ranks chunks, and a marker with no text and no
    embedding can never be a hit. The parent section keeps its id, so anything
    holding one still resolves.

    :param con: Open connection to a passage pack.
    :param trigger_chars: Only sections longer than this are split. Values
        <= 0 disable windowing.
    :param part_chars: Length each part aims for.
    :returns: Number of synthetic sections inserted.
    """
    if trigger_chars <= 0 or part_chars <= 0:
        return 0

    chunks: dict[str, list[tuple[int, int]]] = {}
    for file_path, char_start, length in con.execute(
        "SELECT file_path, char_start, length(content) FROM passages "
        "WHERE kind = 'chunk' AND file_path IS NOT NULL AND char_start IS NOT NULL "
        "ORDER BY file_path, char_start"
    ):
        chunks.setdefault(file_path, []).append((char_start, length or 0))

    sections: dict[str, list[sqlite3.Row]] = {}
    con.row_factory = sqlite3.Row
    for row in con.execute(
        "SELECT rowid, id, kg_name, kg_kind, name, title, node_title, author, genre, "
        "       book, file_path, char_start "
        "  FROM passages "
        " WHERE kind = 'section' AND file_path IS NOT NULL AND char_start IS NOT NULL "
        " ORDER BY file_path, char_start"
    ):
        sections.setdefault(row["file_path"], []).append(row)
    con.row_factory = None

    insert = (
        "INSERT INTO passages"
        "(id, kg_name, kg_kind, kind, name, title, node_title, author, genre, "
        " book, file_path, char_start, chapter, timestamp, content) "
        "VALUES (?, ?, ?, 'section', ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, '')"
    )
    added: list[tuple] = []
    retitled: list[tuple[str, int]] = []

    for file_path, rows in sections.items():
        spans = chunks.get(file_path)
        if not spans:
            continue
        starts = [start for start, _length in spans]
        for index, row in enumerate(rows):
            begin = bisect.bisect_left(starts, row["char_start"])
            if index + 1 < len(rows):
                stop = bisect.bisect_left(starts, rows[index + 1]["char_start"])
            else:
                stop = len(starts)
            if stop - begin < 2:
                continue
            total = sum(length for _start, length in spans[begin:stop])
            if total <= trigger_chars:
                continue

            parts = math.ceil(total / part_chars)
            target = total / parts
            base = row["node_title"] or row["name"] or row["book"] or "Untitled"

            # Cut where the running length crosses each part's share, so parts
            # come out even rather than leaving a stub at the end.
            boundaries: list[int] = []
            running = 0
            wanted = 1
            for offset in range(begin, stop):
                running += spans[offset][1]
                if wanted < parts and running >= target * wanted:
                    boundaries.append(offset + 1)
                    wanted += 1
            boundaries = [b for b in boundaries if begin < b < stop][: parts - 1]
            if not boundaries:
                continue

            parts = len(boundaries) + 1
            retitled.append((f"{base} (Part 1 of {parts})", row["rowid"]))
            for number, offset in enumerate(boundaries, start=2):
                added.append(
                    (
                        f"{row['id']}{_PART_MARKER}{number}",
                        row["kg_name"],
                        row["kg_kind"],
                        row["name"],
                        row["title"],
                        f"{base} (Part {number} of {parts})",
                        row["author"],
                        row["genre"],
                        row["book"],
                        file_path,
                        starts[offset],
                    )
                )

    if retitled:
        con.executemany("UPDATE passages SET node_title = ? WHERE rowid = ?", retitled)
    if added:
        con.executemany(insert, added)
    con.commit()
    return len(added)

write_vector_sidecar(pack, sources, rowid_by_id, *, dtype, progress)

Write <pack>.vectors and point each passage at its row.

The sidecar is a header plus row-major vectors — nothing else. The device memory-maps it and multiplies straight out of the mapping, so a query never allocates and the OS pages in only the part it touches. That is also why the vectors do not live in the pack as a vec0 table: reading one needs the sqlite-vec C extension compiled into the app, and vec0's search is brute force regardless, so the extension would buy nothing.

passages.vector_index is dense (0..N-1) rather than the passage rowid, because a passage whose vector the source store lacks must not leave a hole in the file.

Parameters:

Name	Type	Description	Default
pack	Path	The pack whose passages were just written.	required
sources	Sequence[_PassageSource]	KGs whose vector stores to stream.	required
rowid_by_id	dict[str, int]	Node id → the rowid its passage got.	required
dtype	str	"int8" or "float".	required
progress		Status callable.	required

Returns:

Type	Description
tuple[Path | None, int, int]	(sidecar_path, vectors_written, passages_without_a_vector).

Source code in src/gutenberg_kg/export_swift.py

def write_vector_sidecar(
    pack: Path,
    sources: Sequence[_PassageSource],
    rowid_by_id: dict[str, int],
    *,
    dtype: str,
    progress,
) -> tuple[Path | None, int, int]:
    """Write ``<pack>.vectors`` and point each passage at its row.

    The sidecar is a header plus row-major vectors — nothing else. The device
    memory-maps it and multiplies straight out of the mapping, so a query never
    allocates and the OS pages in only the part it touches. That is also why
    the vectors do not live in the pack as a ``vec0`` table: reading one needs
    the sqlite-vec C extension compiled into the app, and vec0's search is
    brute force regardless, so the extension would buy nothing.

    ``passages.vector_index`` is dense (0..N-1) rather than the passage rowid,
    because a passage whose vector the source store lacks must not leave a hole
    in the file.

    :param pack: The pack whose passages were just written.
    :param sources: KGs whose vector stores to stream.
    :param rowid_by_id: Node id → the rowid its passage got.
    :param dtype: ``"int8"`` or ``"float"``.
    :param progress: Status callable.
    :returns: ``(sidecar_path, vectors_written, passages_without_a_vector)``.
    """
    _load_numpy()  # fail before writing a partial file, not halfway through one
    sidecar = pack.with_suffix(".vectors")
    itemsize = _DTYPE_ITEMSIZE[dtype]

    con = sqlite3.connect(str(pack))
    written = 0
    try:
        con.executescript("PRAGMA journal_mode=OFF; PRAGMA synchronous=OFF;")
        with open(sidecar, "wb") as handle:
            handle.write(b"\0" * VECTOR_HEADER_BYTES)  # backfilled once count is known
            seen: set[int] = set()
            assignments: list[tuple[int, int]] = []
            for source in sources:
                if source.vectors is None:
                    progress(f"  {source.kg_name}: no vector store — lexical search only")
                    continue
                source_start = written
                for node_id, vector in iter_source_vectors(
                    source.vectors, allowed_keys=source.allowed_keys
                ):
                    # The pack id is namespaced by kg_name, so a vector can only
                    # ever land on a row from its own KG — a diary's vector
                    # cannot claim another diary's colliding node id.
                    rowid = rowid_by_id.get(f"{source.kg_name}:{node_id}")
                    if rowid is None or rowid in seen:
                        continue
                    seen.add(rowid)
                    encoded = encode_vector(vector, dtype)
                    if len(encoded) != EMBED_DIM * itemsize:
                        raise ExportError(
                            f"{node_id} has {len(encoded) // itemsize} dimensions, "
                            f"expected {EMBED_DIM}"
                        )
                    handle.write(encoded)
                    assignments.append((written, rowid))
                    written += 1
                progress(
                    f"  {source.kg_name}: {written - source_start:,} vectors written ({dtype})"
                )

            handle.flush()
            handle.seek(0)
            handle.write(_vector_header(dtype, EMBED_DIM, written))

        with con:
            con.executemany("UPDATE passages SET vector_index = ? WHERE rowid = ?", assignments)
    finally:
        con.close()

    if written == 0:
        sidecar.unlink(missing_ok=True)
        return None, 0, len(rowid_by_id)
    return sidecar, written, len(rowid_by_id) - written