Corpus build pipeline

gutenberg_kg.build_corpus

build_corpus.py — Build a single consolidated DocKG over the whole Gutenberg corpus (or a chosen subset of genres), written to bundles/<name>/.dockg/.

This differs from :mod:gutenberg_kg.ingest, which builds one DocKG per book and federates them through the KGRAG registry. Here we walk corpus/ once and produce a single graph.sqlite + vectors.sqlite (sqlite-vec) index — the artifact that gets baked into the standalone "fat image" (Pepys-style: pull, run, query directly).

Genre is recoverable for free at query time: with the walk rooted at corpus/, every node's file_path is <genre>/<book>/<file>.md, so the genre is just the first path segment — no schema change, no tagging pass.

Chunk strategies are applied per-genre group. Sacred texts (Bible KJV, Quran, Torah, etc.) use the verse chunker which honours chapter:verse numbering. All other genres default to semantic. Override via --strategy genre:strategy or the strategy_overrides field on :class:BuildCorpusOptions.

DiaryKG indices (already built per-diary with temporal metadata) are not re-ingested through DocKG. Instead, each corpus/diaries/*/.diarykg/ index is copied verbatim into bundles/<name>/diaries/; the handler can register both the DocKG and the DiaryKG indices through KGRAG.

Output layout::

bundles/
  gutenberg-all/
    .dockg/
      graph.sqlite
      vectors.sqlite     (sqlite-vec; no lancedb/ — a stale one is purged)
      catalog.json
    diaries/
      pepys-complete/.diarykg/
      evelyn-vol1/.diarykg/
      …

bundles/ is gitignored and already in [tool.dockg].exclude, so the consolidated index is never re-ingested by a stray repo-root dockg build.

Author: Eric G. Suchanek, PhD License: Elastic 2.0

BuildCorpusOptions(output=None, similar_k=DEFAULT_SIMILAR_K, similar_max_degree=DEFAULT_SIMILAR_K, discover_similar=False, n_workers=4, embed_batch_size=64, embed_device='auto', wipe=True, update=False, dry_run=False, quiet=False, diaries_only=False, strategy_overrides=dict(), catalog_keys=None, diary_dirs=None, product_name=None, product_version=None, spec_path=None, force_overwrite_full=False) dataclass

Flags controlling a consolidated corpus build.

BuildError

Bases: RuntimeError

A book-filtered build's selection could not be honoured as asked.

assert_selection(files, catalog_keys)

Hard-error if the name-based exclude let an unselected book through.

Cheap — a set difference over the file list already walked — and it is what makes :func:derive_exclude's basename-only exclusion safe to ship: if two genres ever did share a book directory name, this is what turns that into a build failure instead of a bundle silently carrying content nobody asked for.

Parameters:

Name	Type	Description	Default
files	list[Path]	Paths returned by doc_kg.dockg.iter_text_files for the exclude set actually used.	required
catalog_keys	frozenset[str]	The resolved "<genre>/<book>" selection.	required

Raises:

Type	Description
BuildError	If any walked file lies outside the selection.

Source code in src/gutenberg_kg/build_corpus.py

def assert_selection(files: list[Path], catalog_keys: frozenset[str]) -> None:
    """Hard-error if the name-based exclude let an unselected book through.

    Cheap — a set difference over the file list already walked — and it is
    what makes :func:`derive_exclude`'s basename-only exclusion safe to ship:
    if two genres ever did share a book directory name, this is what turns
    that into a build failure instead of a bundle silently carrying content
    nobody asked for.

    :param files: Paths returned by ``doc_kg.dockg.iter_text_files`` for the
        exclude set actually used.
    :param catalog_keys: The resolved ``"<genre>/<book>"`` selection.
    :raises BuildError: If any walked file lies outside the selection.
    """
    stray = sorted({"/".join(f.relative_to(CORPUS_ROOT).parts[:2]) for f in files} - catalog_keys)
    if stray:
        raise BuildError(f"exclude leaked {len(stray)} unselected book(s): {stray[:5]}")

build_catalog(genres, out_dir, *, catalog_keys=None)

Write a catalog.json sidecar mapping <genre>/<book> → book metadata.

Parses each selected book's reference.md for author, title, and Gutenberg ID. The key matches the <genre>/<book> prefix of every node's file_path, so the handler can join author/title onto a hit at query time — no node-schema change, and the file bakes into the image alongside the index (it lives inside .dockg/).

Parameters:

Name	Type	Description	Default
genres	list[str]	Selected genres.	required
out_dir	Path	The bundle's .dockg/ directory (where the index lives).	required
catalog_keys	frozenset[str] | None	Restrict to these keys within genres; None catalogues every book in genres, as today.	None

Returns:

Type	Description
tuple[int, int]	(books_catalogued, books_with_author).

Source code in src/gutenberg_kg/build_corpus.py

def build_catalog(
    genres: list[str], out_dir: Path, *, catalog_keys: frozenset[str] | None = None
) -> tuple[int, int]:
    """Write a ``catalog.json`` sidecar mapping ``<genre>/<book>`` → book metadata.

    Parses each selected book's ``reference.md`` for author, title, and Gutenberg
    ID. The key matches the ``<genre>/<book>`` prefix of every node's
    ``file_path``, so the handler can join author/title onto a hit at query time
    — no node-schema change, and the file bakes into the image alongside the
    index (it lives inside ``.dockg/``).

    :param genres: Selected genres.
    :param out_dir: The bundle's ``.dockg/`` directory (where the index lives).
    :param catalog_keys: Restrict to these keys within ``genres``; ``None``
        catalogues every book in ``genres``, as today.
    :return: ``(books_catalogued, books_with_author)``.
    """
    from gutenberg_kg.authors import parse_reference

    catalog: dict[str, dict] = {}
    with_author = 0
    for genre in genres:
        for ref in sorted((CORPUS_ROOT / genre).glob("*/reference.md")):
            key = f"{genre}/{ref.parent.name}"  # matches file_path prefix
            if catalog_keys is not None and key not in catalog_keys:
                continue
            meta = parse_reference(ref)
            catalog[key] = {
                "genre": genre,
                "book": ref.parent.name,
                "title": meta.get("title"),
                "author": meta.get("author"),
                "author_birth": meta.get("author_birth"),
                "author_death": meta.get("author_death"),
                "author_url": meta.get("author_url"),
                "ebook_id": meta.get("ebook_id"),
            }
            if meta.get("author"):
                with_author += 1

    (out_dir / "catalog.json").write_text(
        json.dumps(catalog, indent=2, ensure_ascii=False), encoding="utf-8"
    )
    return len(catalog), with_author

bundle_diaries(out_dir, *, diary_dirs=None)

Copy existing .diarykg/ indices from corpus/diaries/ into the bundle.

DiaryKG indices carry temporal metadata (YAML timestamps, diary-specific chunks) that DocKG's semantic pipeline cannot reproduce. We copy them verbatim so the handler can register both KGKind.GUTENBERG (the DocKG index) and KGKind.DIARY (each DiaryKG) through KGRAG — without re-ingesting the diaries.

Destination layout::

bundles/<name>/diaries/<diary-name>/.diarykg/

Parameters:

Name	Type	Description	Default
out_dir	Path	The bundle's .dockg/ directory (sibling of diaries/).	required
diary_dirs	tuple[str, ...] | None	Diary slugs to include (matching export_swift._diary_slug), or None to copy every diary found — today's default. An empty tuple copies none, which is a selective bundle's own default (BundleSpec.diaries defaults to False) once a spec or --book/--genre selection is driving the build; it is not a change to the plain, flag-free command.	None

Returns:

Type	Description
int	Number of diary indices copied.

Source code in src/gutenberg_kg/build_corpus.py

def bundle_diaries(out_dir: Path, *, diary_dirs: tuple[str, ...] | None = None) -> int:
    """Copy existing ``.diarykg/`` indices from ``corpus/diaries/`` into the bundle.

    DiaryKG indices carry temporal metadata (YAML timestamps, diary-specific
    chunks) that DocKG's semantic pipeline cannot reproduce.  We copy them
    verbatim so the handler can register both ``KGKind.GUTENBERG`` (the DocKG
    index) and ``KGKind.DIARY`` (each DiaryKG) through KGRAG — without
    re-ingesting the diaries.

    Destination layout::

        bundles/<name>/diaries/<diary-name>/.diarykg/

    :param out_dir: The bundle's ``.dockg/`` directory (sibling of ``diaries/``).
    :param diary_dirs: Diary slugs to include (matching
        ``export_swift._diary_slug``), or ``None`` to copy every diary
        found — today's default. An empty tuple copies none, which is a
        selective bundle's own default (``BundleSpec.diaries`` defaults to
        ``False``) once a spec or ``--book``/``--genre`` selection is driving
        the build; it is not a change to the plain, flag-free command.
    :return: Number of diary indices copied.
    """
    diaries_root = CORPUS_ROOT / "diaries"
    if not diaries_root.exists():
        return 0

    allowed_slugs = frozenset(diary_dirs) if diary_dirs is not None else None
    bundle_diaries_dir = out_dir.parent / "diaries"
    n = 0
    for diary_dir in sorted(diaries_root.iterdir()):
        if not diary_dir.is_dir() or diary_dir.name.startswith("."):
            continue
        if allowed_slugs is not None:
            from gutenberg_kg.export_swift import _diary_slug  # noqa: PLC0415

            if _diary_slug(diary_dir.name) not in allowed_slugs:
                continue
        diarykg_dir = diary_dir / ".diarykg"
        if not diarykg_dir.exists():
            continue
        dest = bundle_diaries_dir / diary_dir.name / ".diarykg"
        dest.parent.mkdir(parents=True, exist_ok=True)
        if dest.exists() or dest.is_symlink():
            shutil.rmtree(dest)
        # Skip any lancedb/ a pre-0.94 diary-kg left behind — the served store
        # is .diarykg/vectors.sqlite and nothing reads LanceDB any more, so
        # copying the old one only bloats the bundle and the image.
        shutil.copytree(
            str(diarykg_dir),
            str(dest),
            symlinks=True,
            ignore=shutil.ignore_patterns("lancedb"),
        )
        n += 1
    return n

derive_exclude(genres, *, only_books=None)

Return the set of directory names to prune from the corpus/ walk.

Excludes every genre not selected, the non-genre corpus dirs, and the built-in DocKG SKIP_DIRS (.git, .dockg, .venv, …). Pruning the unselected genres lets us keep the walk rooted at corpus/ so node file_paths stay genre-prefixed regardless of the subset chosen.

Parameters:

Name	Type	Description	Default
genres	list[str]	Selected genres.	required
only_books	set[str] | None	Book directory names to keep within the selected genres; every sibling not in this set is added to the exclude set too. None keeps every book in every selected genre, as today.	None

Returns:

Type

Description

set[str]

Directory names to exclude at every level of the walk. This is a flat set of basenames, pruned wherever they occur in the walk — it cannot tell one genre's "Republic" from another's. That is a real gap in principle and a closed one in practice: every book directory name in the corpus is unique across all 253 books today. :func:assert_selection is the backstop if that ever stops being true, converting a silent wrong-contents build into a hard failure before any embedding happens, rather than preventing the collision at this level.

Source code in src/gutenberg_kg/build_corpus.py

def derive_exclude(genres: list[str], *, only_books: set[str] | None = None) -> set[str]:
    """Return the set of directory names to prune from the ``corpus/`` walk.

    Excludes every genre *not* selected, the non-genre corpus dirs, and the
    built-in DocKG ``SKIP_DIRS`` (``.git``, ``.dockg``, ``.venv``, …). Pruning
    the unselected genres lets us keep the walk rooted at ``corpus/`` so node
    ``file_path``s stay genre-prefixed regardless of the subset chosen.

    :param genres: Selected genres.
    :param only_books: Book directory names to keep within the selected
        genres; every sibling not in this set is added to the exclude set
        too. ``None`` keeps every book in every selected genre, as today.
    :return: Directory names to exclude at every level of the walk.

        This is a flat set of *basenames*, pruned wherever they occur in the
        walk — it cannot tell one genre's "Republic" from another's. That is
        a real gap in principle and a closed one in practice: every book
        directory name in the corpus is unique across all 253 books today.
        :func:`assert_selection` is the backstop if that ever stops being
        true, converting a silent wrong-contents build into a hard failure
        before any embedding happens, rather than preventing the collision
        at this level.
    """
    from doc_kg.dockg import SKIP_DIRS

    unselected = set(ALL_GENRES) - set(genres)
    exclude = unselected | NON_GENRE_DIRS | set(SKIP_DIRS)
    if only_books is not None:
        for genre in genres:
            genre_dir = CORPUS_ROOT / genre
            if not genre_dir.is_dir():
                continue
            for child in genre_dir.iterdir():
                if (
                    child.is_dir()
                    and child.name not in only_books
                    and not child.name.startswith(".")
                ):
                    exclude.add(child.name)
    return exclude

derive_output_name(genres, override)

Return the bundle directory name for a genre selection.

Parameters:

Name	Type	Description	Default
genres	list[str]	Selected genres (sorted by caller; full set == all genres).	required
override	str | None	Explicit --output name; wins if given.	required

Returns:

Type	Description
str	gutenberg-all for the full corpus, gutenberg-<genre> for a single genre, else a hyphen-joined gutenberg-<g1>-<g2>… slug.

Source code in src/gutenberg_kg/build_corpus.py

def derive_output_name(genres: list[str], override: str | None) -> str:
    """Return the bundle directory name for a genre selection.

    :param genres: Selected genres (sorted by caller; full set == all genres).
    :param override: Explicit ``--output`` name; wins if given.
    :return: ``gutenberg-all`` for the full corpus, ``gutenberg-<genre>`` for a
             single genre, else a hyphen-joined ``gutenberg-<g1>-<g2>…`` slug.
    """
    if override:
        return override
    if set(genres) == set(ALL_GENRES):
        return "gutenberg-all"
    if len(genres) == 1:
        return f"gutenberg-{genres[0]}"
    return "gutenberg-" + "-".join(genres)

ensure_diaries_built(dry_run=False, quiet=False)

Reconstruct any unbuilt diary indices so :func:bundle_diaries finds them.

Runs the DiaryKG pipeline — chunk-diaries (.md → .diary/) then build-diaries (.diary/ → .diarykg/) — idempotently: diaries that already have a .diary/ and .diarykg/ are skipped. This makes gutenkg build-corpus self-sufficient on a clean clone instead of relying on the Makefile to sequence build-diaries first.

Parameters:

Name	Type	Description	Default
dry_run	bool	Print the plan without executing.	False
quiet	bool	Suppress per-stage DiaryKG build progress.	False

Returns:

Type	Description
int	0 on success, 1 if chunking or building any diary failed.

Source code in src/gutenberg_kg/build_corpus.py

def ensure_diaries_built(dry_run: bool = False, quiet: bool = False) -> int:
    """Reconstruct any unbuilt diary indices so :func:`bundle_diaries` finds them.

    Runs the DiaryKG pipeline — ``chunk-diaries`` (``.md`` → ``.diary/``) then
    ``build-diaries`` (``.diary/`` → ``.diarykg/``) — idempotently: diaries that
    already have a ``.diary/`` and ``.diarykg/`` are skipped.  This makes
    ``gutenkg build-corpus`` self-sufficient on a clean clone instead of relying
    on the Makefile to sequence ``build-diaries`` first.

    :param dry_run: Print the plan without executing.
    :param quiet: Suppress per-stage DiaryKG build progress.
    :return: 0 on success, 1 if chunking or building any diary failed.
    """
    from gutenberg_kg.build_diaries import BuildDiariesOptions, run_build_diaries
    from gutenberg_kg.diary.chunk import ChunkDiariesOptions, run_chunk_diaries

    rc = run_chunk_diaries([], ChunkDiariesOptions(force=False, dry_run=dry_run))
    if rc != 0:
        return rc
    rc, _results = run_build_diaries(
        [], BuildDiariesOptions(force=False, dry_run=dry_run, quiet=quiet)
    )
    return rc

fmt_duration(seconds)

Human-readable duration string.

Source code in src/gutenberg_kg/build_corpus.py

def fmt_duration(seconds: float) -> str:
    """Human-readable duration string."""
    if seconds < 60:
        return f"{seconds:.1f}s"
    m, s = divmod(int(seconds), 60)
    if m < 60:
        return f"{m}m {s:02d}s"
    h, m = divmod(m, 60)
    return f"{h}h {m:02d}m {s:02d}s"

run_build_corpus(genres, opts)

Build one consolidated DocKG over the selected genres.

Genres are grouped by their effective chunk strategy, then each group is walked and parsed into the same graph.sqlite (first group wipes, the rest append). A single embedding pass and a single vector index pass run over the combined node table. Finally, any DiaryKG indices found under corpus/diaries/ are copied verbatim into the bundle.

Parameters:

Name	Type	Description	Default
genres	list[str]	Genre names to include (already validated; empty == all).	required
opts	BuildCorpusOptions	Build option flags.	required

Returns:

Type	Description
int	0 on success, 1 on failure.

Source code in src/gutenberg_kg/build_corpus.py

def run_build_corpus(genres: list[str], opts: BuildCorpusOptions) -> int:
    """Build one consolidated DocKG over the selected genres.

    Genres are grouped by their effective chunk strategy, then each group is
    walked and parsed into the *same* ``graph.sqlite`` (first group wipes, the
    rest append).  A single embedding pass and a single vector index pass run
    over the combined node table.  Finally, any DiaryKG indices found under
    ``corpus/diaries/`` are copied verbatim into the bundle.

    :param genres: Genre names to include (already validated; empty == all).
    :param opts: Build option flags.
    :return: 0 on success, 1 on failure.
    """
    genres = sorted(genres) if genres else list(ALL_GENRES)
    name = derive_output_name(genres, opts.output)
    if opts.catalog_keys is not None and name == "gutenberg-all" and not opts.force_overwrite_full:
        print(
            "[x] a book-filtered build targeting 'gutenberg-all' needs --force-overwrite-full "
            "(or pass --output to name it something else)."
        )
        return 1
    only_books = (
        {key.split("/", 1)[1] for key in opts.catalog_keys}
        if opts.catalog_keys is not None
        else None
    )
    out_dir = BUNDLES_ROOT / name / ".dockg"
    sqlite_path = out_dir / "graph.sqlite"
    vectors_path = out_dir / "vectors.sqlite"
    # Only ever used to find and delete a pre-migration store (see the purge
    # below). It is deliberately NOT passed to DocKG: since doc-kg 0.20.0 the
    # sqlite-vec location is given directly via `vectors_path`, so `lancedb_dir`
    # no longer has to serve as the anchor the sidecar path was derived from.
    lancedb_path = out_dir / "lancedb"

    # Effective strategy: defaults merged with caller overrides.
    effective_strategy: dict[str, str] = {**GENRE_STRATEGY, **opts.strategy_overrides}

    # Group genres by chunk strategy.
    strategy_groups: dict[str, list[str]] = defaultdict(list)
    for genre in genres:
        strategy = effective_strategy.get(genre, "semantic")
        strategy_groups[strategy].append(genre)

    n_books = (
        len(opts.catalog_keys)
        if opts.catalog_keys is not None
        else sum(
            1
            for g in genres
            if (CORPUS_ROOT / g).is_dir()
            for p in (CORPUS_ROOT / g).iterdir()
            if p.is_dir() and not p.name.startswith(".")
        )
    )

    print("=== gutenkg build-corpus ===")
    print(f"  bundle        : {name}")
    print(f"  output        : {out_dir}")
    print(
        f"  genres        : {len(genres)} ({'all' if set(genres) == set(ALL_GENRES) else ', '.join(genres)})"
    )
    print(f"  books         : ~{n_books}")
    print(f"  similar_k     : {opts.similar_k if opts.discover_similar else 'disabled'}")
    _mode = {
        "cpu": f"parallel (CPU multiprocessing, {opts.n_workers} workers)",
        "mps": "streaming (single-process; GPU can't be shared across workers)",
        "cuda": "streaming (single-process; GPU can't be shared across workers)",
    }.get(
        opts.embed_device,
        "auto → streaming MPS (single-process)"
        if _mps_available()
        else f"auto → parallel CPU ({opts.n_workers} workers)",
    )
    print(f"  embed mode    : {_mode}")
    print(f"  embed batch   : {opts.embed_batch_size}")
    print(f"  embed device  : {opts.embed_device}")
    print()
    print("  strategy groups:")
    for strategy, sg_genres in sorted(strategy_groups.items()):
        print(f"    {strategy:16s} : {', '.join(sorted(sg_genres))}")
    print()

    if opts.diaries_only:
        print("[diaries-only] skipping phases 1-3; building + bundling diary indices …")
        if ensure_diaries_built(dry_run=opts.dry_run, quiet=opts.quiet) != 0:
            print("[x] diary build failed")
            return 1
        try:
            n_diaries = bundle_diaries(out_dir, diary_dirs=opts.diary_dirs)
        except Exception as exc:  # noqa: BLE001
            print(f"[x] bundle_diaries failed: {exc}")
            return 1
        if n_diaries:
            print(f"  copied {n_diaries} diary index(es) → bundles/{name}/diaries/")
        else:
            print("  (no .diarykg indices found under corpus/diaries/)")
        return 0

    if opts.dry_run:
        print("[dry-run] phase 1: build graph — sequential per strategy group:")
        for strategy, sg_genres in sorted(strategy_groups.items()):
            sg_exclude = derive_exclude(sg_genres, only_books=only_books)
            print(f"  strategy={strategy}, genres={sorted(sg_genres)}")
            print(f"    exclude={sorted(sg_exclude)}")
        print(f"[dry-run] phase 2: embed all nodes → {sqlite_path}")
        print(f"[dry-run] phase 3: sqlite-vec index + SIMILAR_TO → {vectors_path}")
        if lancedb_path.exists():
            print(
                f"[dry-run] would remove stale LanceDB store "
                f"({_dir_size_mb(lancedb_path):.0f} MB) → {lancedb_path}"
            )
        print("[dry-run] phase 4: build (if needed) + bundle DiaryKG indices → bundles dir")
        print(f"[dry-run] would write {out_dir / 'catalog.json'} (author/title per book)")
        return 0

    try:
        from doc_kg.index import make_embedder
        from doc_kg.kg import DocKG
    except ImportError as exc:
        print(f"[x] doc_kg not installed (needed for build-corpus): {exc}")
        return 1

    out_dir.mkdir(parents=True, exist_ok=True)

    # Purge a LanceDB store left over from a pre-migration build: ~2.8 GB of
    # index nothing reads any more (LanceDB support was removed fleet-wide in
    # 0.29.0), which a blanket `COPY bundles/` baked into the
    # image byte for byte.
    #
    # It used to be a correctness issue too, not just weight: through doc-kg
    # 0.19 the default backend was "auto", resolved by probing the filesystem,
    # so a stale lancedb/ with no vectors.sqlite yet — the state a --wipe
    # rebuild starts from — silently selected the retired backend. doc-kg 0.20.0
    # pins the default to sqlite-vec and we now pass it explicitly besides, so
    # this is purely about size. Delete it anyway; nothing should ship it.
    if lancedb_path.exists():
        stale_mb = _dir_size_mb(lancedb_path)
        shutil.rmtree(lancedb_path)
        print(f"  [clean] removed stale LanceDB store ({stale_mb:.0f} MB) → {lancedb_path}")

    t0 = time.perf_counter()
    # Build the embedder on the requested device. The consolidated build embeds
    # 700k+ nodes in one pass; on Apple MPS the unified-memory watermark contends
    # with the system file cache and OOMs on "other allocations", so
    # `--embed-device cpu` is the reliable choice for the full corpus.
    if opts.embed_device == "mps" and not _mps_available():
        print("[x] --embed-device mps requested but MPS is not available")
        return 1
    if opts.embed_device == "cuda" and not _cuda_available():
        print("[x] --embed-device cuda requested but CUDA is not available")
        return 1
    try:
        embedder = make_embedder(device=opts.embed_device)
    except Exception as exc:  # noqa: BLE001
        print(f"[x] failed to build embedder on device {opts.embed_device!r}: {exc}")
        return 1
    print(f"  [embedder] {embedder!r}  (device={opts.embed_device})\n")

    try:
        # ------------------------------------------------------------------
        # Phase 1: parse each strategy group into the shared SQLite graph.
        # First group wipes (if opts.wipe); subsequent groups append.
        # ------------------------------------------------------------------
        n_groups = len(strategy_groups)
        print(f"[1/4] parsing corpus → SQLite ({n_groups} strategy group(s)) …")
        first = True
        for strategy, sg_genres in sorted(strategy_groups.items()):
            sg_exclude = derive_exclude(sg_genres, only_books=only_books)
            if opts.catalog_keys is not None:
                # Before any embedding happens: convert a silent
                # wrong-contents build into a hard failure now, at the cost
                # of one more corpus walk.
                from doc_kg.dockg import iter_text_files  # noqa: PLC0415

                walked = iter_text_files(CORPUS_ROOT, exclude=sg_exclude)
                assert_selection(walked, opts.catalog_keys)
            print(f"  [{strategy}] {', '.join(sorted(sg_genres))}")
            kg = DocKG(
                corpus_root=CORPUS_ROOT,
                db_path=sqlite_path,
                exclude=sg_exclude,
                embedder=embedder,
                chunk_strategy=strategy,
                # State the store explicitly in both phases rather than letting
                # either infer it. doc-kg 0.20.0 defaults `vector_backend` to
                # sqlite-vec (0.19 and earlier defaulted to "auto", which
                # resolved from whatever was on disk — a stale lancedb/ dir was
                # enough to keep a migrated corpus on the retired backend), so
                # this is belt-and-braces; it also means the call says what it
                # writes instead of relying on a default we don't control.
                vector_backend="sqlite-vec",
                vectors_path=vectors_path,
            )
            kg.build_graph(wipe=first and opts.wipe, quiet=opts.quiet)
            kg.close()
            first = False

        # ------------------------------------------------------------------
        # Phase 2: embed all nodes in the combined graph.
        # ------------------------------------------------------------------
        print("[2/4] embedding nodes …")
        kg_all = DocKG(
            corpus_root=CORPUS_ROOT,
            db_path=sqlite_path,
            embedder=embedder,
            # Served bundle is sqlite-vec (exact, ~10x smaller), written to the
            # path named here rather than derived from a LanceDB anchor.
            vector_backend="sqlite-vec",
            vectors_path=vectors_path,
        )
        # Resolve the effective device to pick the embedding path: CPU can fan
        # out across processes safely; MPS/CUDA cannot (one shared allocator).
        # `auto` prefers a GPU when available (MPS on Macs, CUDA elsewhere —
        # e.g. RunPod), falling back to parallel CPU when no GPU is present.
        #
        # CAUTION: unlike `dockg build-index` (SemanticIndex.build(), which
        # hard-caps its encode sub-batch at 128 — doc_kg PR #7 / kgmodule-utils
        # 0.4.6), this streaming path (DocKG.build_embeddings ->
        # precompute_embeddings -> _precompute_embeddings_jsonl_stream) has NO
        # internal cap: --embed-batch-size is passed straight to
        # model.encode() uncapped. It's safe today only because our own
        # --embed-batch-size default (64) is small. Raising it for throughput
        # while on MPS/CUDA for the full 700k+ node corpus reintroduces the
        # batch x seq^2 unified-memory OOM PR #7 fixed elsewhere -- that fix
        # does not reach this code path.
        if opts.embed_device == "auto":
            if _mps_available():
                effective_device = "mps"
            elif _cuda_available():
                effective_device = "cuda"
            else:
                effective_device = "cpu"
        else:
            effective_device = opts.embed_device

        if effective_device == "cpu" or opts.update:
            # CPU (or any --update) → PARALLEL embedding via doc_kg's multi-process
            # CorpusEmbedder (`.json` cache). This is the only path that supports
            # incremental `only_missing` embedding. Pin every worker via
            # KG_EMBED_DEVICE — even when the resolved device is MPS (--update):
            # without the pin each worker auto-selects MPS and N workers stack
            # N GPU allocations → OOM.
            os.environ["KG_EMBED_DEVICE"] = "cpu"
            # .jsonl streams vectors to disk as shards complete (CorpusEmbedder.
            # embed_to_cache) instead of holding the whole corpus in RAM — peak memory
            # is bounded by shard size, which is what kept the 689k-node build off swap.
            cache_path = out_dir / "embeddings.jsonl"
            mode = "incremental — embed only new/changed nodes" if opts.update else "full"
            print(f"  [parallel] CPU multiprocessing ({mode})")
            kg_all.build_embeddings(
                out=cache_path,
                n_workers=opts.n_workers,  # each worker loads its own model copy (~1.2 GB)
                batch_size=opts.embed_batch_size,
                only_missing=opts.update,
                quiet=opts.quiet,
            )
        else:
            # MPS/CUDA → single-process streaming to a `.jsonl` cache. The GPU
            # allocator can't be shared across spawn workers without OOM, so
            # parallelism is off here by design; streaming keeps GPU + host
            # memory flat across the whole corpus (700k+ nodes).
            cache_path = out_dir / "embeddings.jsonl"
            print(f"  [stream] single-process on {effective_device}")
            kg_all.build_embeddings(
                out=cache_path,
                n_workers=None,
                batch_size=opts.embed_batch_size,
                quiet=opts.quiet,
            )

        # ------------------------------------------------------------------
        # Phase 3: build the sqlite-vec index + SIMILAR_TO edges.
        # ------------------------------------------------------------------
        # On --update we UPSERT into the existing vector index (wipe=False) and skip
        # SIMILAR_TO — the served handler is semantic-first and never traverses those
        # edges, so recomputing 800k+ of them every update is pure waste.
        index_wipe = opts.wipe and not opts.update
        discover_similar = opts.discover_similar and not opts.update
        print(
            f"[3/4] building sqlite-vec index{' + SIMILAR_TO edges' if discover_similar else ' (upsert)' if opts.update else ''} …"
        )
        stats = kg_all.build_index_from_cache(
            cache_path,
            wipe=index_wipe,
            discover_similar=discover_similar,
            similar_k=opts.similar_k,
            similar_max_degree=opts.similar_max_degree,
            quiet=opts.quiet,
        )

        # Rebuild the FTS5 lexical index over the *full* consolidated graph so
        # the handler's hybrid (dense + BM25) retrieval activates. build_graph()
        # rebuilds it per strategy group, but we do it once more here as an
        # explicit guard: it is the single artifact the lexical channel depends
        # on, and an absent nodes_fts silently degrades retrieval to dense-only.
        print("[3/4] building FTS5 lexical index (nodes_fts) over the full graph …")
        n_fts = kg_all.store.rebuild_fts(quiet=opts.quiet)
        print(f"  lexical index: {n_fts:,} chunks")

        # Incremental adds only; drop index vectors for nodes that left the graph
        # (removed/renamed books) so they can't return stale hits.
        if opts.update:
            print("[3/4] pruning orphan vectors (removed/renamed nodes) …")
            kg_all.prune_index(quiet=opts.quiet)

        kg_all.close()
        # The embedding cache (.json / .jsonl) is a build-only intermediate — it can
        # be several GB and must never ship in the bundle. Remove BOTH variants so a
        # stale cache from a prior or killed run can't bloat the bundle (and the image
        # COPY). .dockerignore guards the same at pack time as a belt-and-suspenders.
        for _cache in ("embeddings.json", "embeddings.jsonl"):
            (out_dir / _cache).unlink(missing_ok=True)

        # ------------------------------------------------------------------
        # Phase 4: build (if needed) + bundle DiaryKG indices.
        # ------------------------------------------------------------------
        print("[4/4] building + bundling DiaryKG indices …")
        if ensure_diaries_built(dry_run=opts.dry_run, quiet=opts.quiet) != 0:
            print("[x] diary build failed")
            return 1
        n_diaries = bundle_diaries(out_dir, diary_dirs=opts.diary_dirs)
        if n_diaries:
            print(f"  copied {n_diaries} diary index(es) → bundles/{name}/diaries/")
        else:
            print("  (no .diarykg indices found under corpus/diaries/)")

        print("[+] writing catalog.json (author/title from reference.md) …")
        n_cat, n_auth = build_catalog(genres, out_dir, catalog_keys=opts.catalog_keys)

        product_path = None
        if opts.catalog_keys is not None:
            print("[+] writing product.json (frozen selection + checksums) …")
            product_path = write_product_json(
                BUNDLES_ROOT / name,
                name=opts.product_name or name,
                version=opts.product_version,
                catalog_keys=opts.catalog_keys,
                diary_dirs=_diary_dirs_for_product(opts.diary_dirs, n_diaries),
                spec_path=opts.spec_path,
            )

    except Exception as exc:  # noqa: BLE001
        print(f"[x] build failed: {exc}")
        msg = str(exc)
        if "MPS backend out of memory" in msg:
            print("[hint] Apple MPS OOM during embedding pass.")
            print("[hint] Retry with CPU embeddings and a smaller embed batch:")
            print("       gutenkg build-corpus --embed-device cpu --embed-batch-size 16")
        return 1

    elapsed = time.perf_counter() - t0
    size_mb = _dir_size_mb(out_dir)

    print()
    print("=== build complete ===")
    print(f"  bundle        : {name}")
    print(f"  nodes         : {stats.total_nodes:,}")
    print(f"  edges         : {stats.total_edges:,}")
    if stats.similar_edges_added is not None:
        print(f"  SIMILAR_TO    : {stats.similar_edges_added:,}")
    print(f"  catalog       : {n_cat} books ({n_auth} with author)")
    print(f"  diaries       : {n_diaries} index(es) bundled")
    print(f"  index size    : {size_mb:,.1f} MB")
    print(f"  elapsed       : {fmt_duration(elapsed)}")
    print(f"  written to    : {out_dir}")
    if product_path is not None:
        print(f"  product.json  : {product_path}")
    print()
    return 0

write_product_json(bundle_dir, *, name, version, catalog_keys, diary_dirs, materialize='rebuild', spec_path=None)

Freeze a selective build's resolved selection and checksums to disk.

A rollback and audit record only — neither the worker nor the Swift reader reads it. Written only for a book-filtered build; the plain gutenberg-all build has no selection to freeze.

Parameters:

Name	Type	Description	Default
bundle_dir	Path	The bundle root, bundles/<name>/ (parent of .dockg/).	required
name	str	Product name — the spec's name, or the bundle directory name for an ad-hoc --book/--genre build with no spec.	required
version	str | None	Product version, or None when there is no spec to supply one.	required
catalog_keys	frozenset[str]	The resolved "<genre>/<book>" selection.	required
diary_dirs	tuple[str, ...]	The resolved diary slugs, possibly empty.	required
materialize	str	Always "rebuild" here — this function only runs after a build, never for materialize = "none".	'rebuild'
spec_path	Path | None	The spec file used, if any; hashed into spec_sha256.	None

Returns:

Type	Description
Path	Path to the written product.json.

Source code in src/gutenberg_kg/build_corpus.py

def write_product_json(
    bundle_dir: Path,
    *,
    name: str,
    version: str | None,
    catalog_keys: frozenset[str],
    diary_dirs: tuple[str, ...],
    materialize: str = "rebuild",
    spec_path: Path | None = None,
) -> Path:
    """Freeze a selective build's resolved selection and checksums to disk.

    A rollback and audit record only — neither the worker nor the Swift
    reader reads it. Written only for a book-filtered build; the plain
    ``gutenberg-all`` build has no selection to freeze.

    :param bundle_dir: The bundle root, ``bundles/<name>/`` (parent of
        ``.dockg/``).
    :param name: Product name — the spec's ``name``, or the bundle directory
        name for an ad-hoc ``--book``/``--genre`` build with no spec.
    :param version: Product version, or ``None`` when there is no spec to
        supply one.
    :param catalog_keys: The resolved ``"<genre>/<book>"`` selection.
    :param diary_dirs: The resolved diary slugs, possibly empty.
    :param materialize: Always ``"rebuild"`` here — this function only runs
        after a build, never for ``materialize = "none"``.
    :param spec_path: The spec file used, if any; hashed into
        ``spec_sha256``.
    :returns: Path to the written ``product.json``.
    """
    dockg_dir = bundle_dir / ".dockg"
    dockg_sha256 = {
        fname: _sha256(dockg_dir / fname)
        for fname in ("graph.sqlite", "vectors.sqlite", "catalog.json")
        if (dockg_dir / fname).exists()
    }
    document = {
        "name": name,
        "version": version,
        "spec_sha256": _sha256(spec_path) if spec_path is not None else None,
        "materialize": materialize,
        "catalog_keys": sorted(catalog_keys),
        "book_count": len(catalog_keys),
        "diary_dirs": list(diary_dirs),
        "created": datetime.now(UTC).isoformat(timespec="seconds"),
        "dockg_sha256": dockg_sha256,
    }
    path = bundle_dir / "product.json"
    path.write_text(json.dumps(document, indent=2, ensure_ascii=False), encoding="utf-8")
    return path