Index

Two search backends, plus the store both are rebuilt from. Neither backend persists its own on-disk form: BM25Index.from_store() and a freshly populated in-memory vector index are both derived, at open time, from SQLiteMetadataStore — the durable truth for documents and chunks (ADR-0002). That is a deliberate anti-pattern guard, not an optimization deferred: no in-process index state may shadow persisted state (SPEC.md §2), so postings and vectors can never drift out of sync with what SQLite actually holds.

Lexical (BM25)

Ported near-verbatim from ARP (ADR-0001) — pure Python, zero dependencies. BM25Index.search returns (Chunk, score) pairs rather than constructing a RetrievalResult itself; joining a chunk back to its document's source path is the retrieval layer's job; see Retrieval.

groundkit.index.bm25

Persisted BM25 lexical index (Phase 1). Library choice justified in an ADR.

Ported from ARP's agentic_v2.rag.retrieval.BM25Index (ADR-0001: pure-Python BM25, promoted near-verbatim — zero deps, already pinned by ARP's behavioral tests). Two deliberate changes for groundkit's layering:

• :meth:BM25Index.search returns (Chunk, score) pairs instead of constructing RetrievalResult directly. RetrievalResult.source has no counterpart on Chunk (only Document carries the source path/URL), so assembling a citation-bearing result means joining against the metadata store — that join is the retrieval layer's job, not this module's.
• :meth:BM25Index.from_store rebuilds a fresh in-memory index from :class:~groundkit.index.metadata.SQLiteMetadataStore (ADR-0002): SQLite is the durable truth for chunks; BM25 postings are never pickled to disk.

BM25Index

BM25Index(*, k1: float = 1.5, b: float = 0.75)

Pure-Python in-memory BM25 keyword index.

Builds an inverted index from chunks — a term -> [chunk index] postings map (_postings) beside the document-frequency counter (_doc_freqs) and the length statistics — tokenizes on lowercased word characters, and scores candidates with Okapi BM25.

ADR-0002 decision 2 named all three of those from its first version, and until GK-018 only two of them existed — so :meth:search scored every indexed chunk however selective the query was, making query cost a function of corpus size. The postings map closes that, recorded as an erratum on the decision itself rather than silently. What it buys is score-identical, not an approximation: a chunk holding no query term skips every term inside :meth:_score_document, scores exactly 0.0, and was already dropped by that method's caller, so narrowing the walk to the union of the query terms' postings removes work and never a result — the low-scoring-but-nonzero tail included, which is what an approximate candidate scheme would have cost and why none was taken.

Not safe to mutate while a search is running. :class:~groundkit.retrieval.search.Retriever dispatches :meth:search to a worker thread, where it reads the parallel per-chunk lists and the postings map below live; :meth:index_chunks extends them in separate statements, so an append landing mid-scan would be observed torn — a posting, or len(self._chunks), past the matching _doc_lengths or _doc_term_freqs entry, raising IndexError from :meth:_score_document. Build the index fully (normally via :meth:from_store) and treat it as frozen thereafter; rebuilding means constructing a new instance, which is what :class:~groundkit.runtime.CollectionRuntime does.

Parameters:

Name	Type	Description	Default
k1	float	Term-frequency saturation parameter.	1.5
b	float	Length-normalization parameter.	0.75

Source code in src/groundkit/index/bm25.py

def __init__(self, *, k1: float = 1.5, b: float = 0.75) -> None:
    self._k1 = k1
    self._b = b
    self._chunks: list[Chunk] = []
    self._tie_keys: list[str] = []
    self._doc_freqs: dict[str, int] = defaultdict(int)
    # term -> the indices of every chunk holding it, strictly ascending
    # and duplicate-free (see index_chunks). Not a second copy of
    # _doc_freqs: that counter answers "how many chunks hold this term"
    # for IDF, this answers "which ones" for candidate selection, and
    # ``len(self._postings[term]) == self._doc_freqs[term]`` holds by
    # construction because both are written from the same first-sighting
    # branch.
    self._postings: dict[str, list[int]] = defaultdict(list)
    self._doc_term_freqs: list[dict[str, int]] = []
    self._doc_lengths: list[int] = []
    self._avg_doc_length: float = 0.0

size property

size: int

Number of chunks currently indexed.

from_store async classmethod

from_store(
    store: MetadataStoreProtocol,
    *,
    k1: float = 1.5,
    b: float = 0.75,
) -> BM25Index

Rebuild a fresh in-memory BM25 index from persisted chunks.

ADR-0002: BM25 postings are never pickled — SQLite is durable truth, and the in-memory index is reconstructed from it at process start. Cost is O(corpus size); acceptable for v1 and reconsidered if rebuild time is ever measured to be a problem.

Parameters:

Name	Type	Description	Default
store	MetadataStoreProtocol	The metadata store to read every persisted chunk from.	required
k1	float	Term-frequency saturation parameter for the new index.	1.5
b	float	Length-normalization parameter for the new index.	0.75

Returns:

Type	Description
BM25Index	A new :class:BM25Index populated with every chunk currently
BM25Index	persisted in store.

Source code in src/groundkit/index/bm25.py

@classmethod
async def from_store(
    cls, store: MetadataStoreProtocol, *, k1: float = 1.5, b: float = 0.75
) -> BM25Index:
    """Rebuild a fresh in-memory BM25 index from persisted chunks.

    ADR-0002: BM25 postings are never pickled — SQLite is durable truth,
    and the in-memory index is reconstructed from it at process start.
    Cost is O(corpus size); acceptable for v1 and reconsidered if
    rebuild time is ever measured to be a problem.

    Args:
        store: The metadata store to read every persisted chunk from.
        k1: Term-frequency saturation parameter for the new index.
        b: Length-normalization parameter for the new index.

    Returns:
        A new :class:`BM25Index` populated with every chunk currently
        persisted in ``store``.
    """
    index = cls(k1=k1, b=b)
    chunks = await store.get_chunks()
    index.index_chunks(chunks)
    return index

index_chunks

index_chunks(chunks: list[Chunk]) -> None

Incrementally add chunks to the BM25 index.

Existing indexed chunks are preserved — this is an accumulation, not a replacement. Rebuilding from scratch means constructing a new :class:BM25Index (see :meth:from_store).

A chunk's index joins the postings list of every term it holds, on that term's first occurrence in the chunk rather than on every occurrence — the same branch that increments the document-frequency counter, so the two cannot disagree about which chunks hold a term. Appending in chunk order then leaves each postings list strictly ascending and duplicate-free, which is the property :meth:search relies on to visit candidates in the order a whole-corpus scan would have.

Parameters:

Name	Type	Description	Default
chunks	list[Chunk]	Chunks to add.	required

Source code in src/groundkit/index/bm25.py

def index_chunks(self, chunks: list[Chunk]) -> None:
    """Incrementally add chunks to the BM25 index.

    Existing indexed chunks are preserved — this is an accumulation,
    not a replacement. Rebuilding from scratch means constructing a new
    :class:`BM25Index` (see :meth:`from_store`).

    A chunk's index joins the postings list of every term it holds, on
    that term's *first* occurrence in the chunk rather than on every
    occurrence — the same branch that increments the document-frequency
    counter, so the two cannot disagree about which chunks hold a term.
    Appending in chunk order then leaves each postings list strictly
    ascending and duplicate-free, which is the property :meth:`search`
    relies on to visit candidates in the order a whole-corpus scan
    would have.

    Args:
        chunks: Chunks to add.
    """
    for chunk in chunks:
        doc_idx = len(self._chunks)
        self._chunks.append(chunk)
        self._tie_keys.append(chunk.content_hash)
        tokens = _tokenize(chunk.content)
        self._doc_lengths.append(len(tokens))

        term_freq: dict[str, int] = defaultdict(int)
        seen_terms: set[str] = set()
        for token in tokens:
            term_freq[token] += 1
            if token not in seen_terms:
                self._doc_freqs[token] += 1
                self._postings[token].append(doc_idx)
                seen_terms.add(token)
        self._doc_term_freqs.append(dict(term_freq))

    total_length = sum(self._doc_lengths)
    self._avg_doc_length = total_length / len(self._chunks) if self._chunks else 0.0
    logger.debug("BM25 index now holds %d chunks", len(self._chunks))

search

search(
    query: str, *, top_k: int = 5
) -> list[tuple[Chunk, float]]

Search the index with a BM25 query.

Only chunks holding at least one query term are scored (GK-018), so the cost is proportional to the number of matching chunks rather than to the corpus. The returned list is exactly what scoring every chunk produced, ties included — see the class docstring for why that is an identity rather than an approximation.

Parameters:

Name	Type	Description	Default
query	str	The search query string.	required
top_k	int	Maximum number of results to return.	5

Returns:

Type	Description
list[tuple[Chunk, float]]	(chunk, score) pairs ranked by descending BM25 score, each
list[tuple[Chunk, float]]	score >= 0.0. An empty query or an empty index returns
list[tuple[Chunk, float]]	[], as does a query whose terms match nothing. Ties are
list[tuple[Chunk, float]]	broken by ascending content_hash so that ranking is stable
list[tuple[Chunk, float]]	across re-ingestion of the same corpus (see the sort below).
list[tuple[Chunk, float]]	The one case this can't disambiguate: two byte-identical chunks
list[tuple[Chunk, float]]	share a content_hash and fall back to insertion order — no
list[tuple[Chunk, float]]	content-derived key can distinguish them.

Source code in src/groundkit/index/bm25.py

def search(self, query: str, *, top_k: int = 5) -> list[tuple[Chunk, float]]:
    """Search the index with a BM25 query.

    Only chunks holding at least one query term are scored (GK-018), so
    the cost is proportional to the number of matching chunks rather than
    to the corpus. The returned list is exactly what scoring every chunk
    produced, ties included — see the class docstring for why that is an
    identity rather than an approximation.

    Args:
        query: The search query string.
        top_k: Maximum number of results to return.

    Returns:
        ``(chunk, score)`` pairs ranked by descending BM25 score, each
        score ``>= 0.0``. An empty query or an empty index returns
        ``[]``, as does a query whose terms match nothing. Ties are
        broken by ascending ``content_hash`` so that ranking is stable
        across re-ingestion of the same corpus (see the sort below).
        The one case this can't disambiguate: two byte-identical chunks
        share a ``content_hash`` and fall back to insertion order — no
        content-derived key can distinguish them.
    """
    query_tokens = _tokenize(query)
    if not query_tokens or not self._chunks:
        return []

    candidates: set[int] = set()
    for token in query_tokens:
        candidates.update(self._postings.get(token, _NO_POSTINGS))
    if not candidates:
        return []

    # Ascending, so the walk visits candidates in the order a
    # whole-corpus scan visited them. ``list.sort`` below is stable, so
    # that order is what two chunks tied on *both* score and
    # ``content_hash`` — byte-identical chunks, the one case the
    # tie-break cannot resolve — still fall back to. Iterating
    # ``candidates`` directly would hand that fallback to set layout.
    scored: list[tuple[float, int]] = []
    for doc_idx in sorted(candidates):
        score = self._score_document(query_tokens, doc_idx)
        if score > 0.0:
            scored.append((score, doc_idx))

    # Tie-break on content_hash, not document_id/chunk_id (contracts.py:38,64):
    # both are uuid4, regenerated on every ingest, so using either as a
    # secondary key would make tied ordering just as unstable as no
    # secondary key at all. Negate the score (rather than reverse=True)
    # so both sort keys are ascending — reverse=True would also flip the
    # tie-break to descending hash order, which isn't the intent.
    scored.sort(key=lambda pair: (-pair[0], self._tie_keys[pair[1]]))
    return [(self._chunks[doc_idx], score) for score, doc_idx in scored[:top_k]]

Dense (vector)

InMemoryVectorStore is the offline/dev double, importable with no optional extra installed. LanceDBVectorStore is the persisted backend behind the dense extra, with a lazily-imported lancedb/pyarrow so a BM25-only install never pays for either. Both implementations close two ported ARP defects (ADR-0001): hazard 3, a metadata filter accepted through **kwargs and silently never applied, and hazard 5, a document ID reaching an unescaped LanceDB delete predicate. Filtering happens before the top-k truncation on both stores, not after — filter-then-truncate, so a filtered query can't lose real matches to unfiltered ones that happened to score higher.

groundkit.index.dense

Dense vector index behind :class:~groundkit.index.protocols.VectorStoreProtocol (Phase 3, ADR-0001, ADR-0004).

Two implementations land together, never separately (SPEC.md §5.3: "Metadata filtering is implemented on both in-memory and LanceDB paths from the first dense-store commit, with regression tests on both paths"):

• :class:InMemoryVectorStore — pure Python, no lancedb import. The dev/test double for offline work and CI where the optional dense extra may be absent.
• :class:LanceDBVectorStore — persisted, backed by LanceDB. The lancedb/pyarrow import is lazy (see :func:_import_lancedb): a BM25-only install can still import groundkit.index.dense freely, and only constructing a :class:LanceDBVectorStore without the dense extra installed raises a clear :class:~groundkit.errors.StorageError instead of a bare ImportError surfacing from deep inside this module.

Both implementations close two ported ARP defects (ADR-0001):

• Hazard 3 — silent filter drop. ARP's search accepted a filter argument via **kwargs and never read it; any call spelling, keyword or positional, silently returned unfiltered results. VectorStoreProtocol already closes the spelling half of this (no **kwargs, so a misspelled keyword is a TypeError at the call site) — the implementation half is this module's job: :func:_matches_filter is the single place a metadata_filter is ever applied, on both stores, with identical semantics (subset match: every (key, value) pair in the filter must be present and equal in the chunk's metadata).
• Hazard 5 — unescaped delete predicate. LanceDB deletes through a SQL expression string built by the caller. ADR-0004 decision 6: a document ID is validated against a strict identifier pattern (:func:_validate_document_id, mirroring index/metadata.py's validate_collection_name in spirit) before it can reach :func:_build_delete_predicate — the single function that builds that string. Escaping is defence in depth, applied after validation, never the primary defence. Verified live against this repo's pinned LanceDB 0.37.1: t.delete("document_id = 'nope\' OR \'1\'=\'1'") wipes every row in the table; the regression test in tests/test_dense.py asserts that input cannot reach a predicate at all.

Filter-then-truncate. Filtering a vector search result after asking the backend for top_k nearest neighbors returns fewer than top_k whenever some of those neighbors don't match the filter — a silently short result set, which SPEC.md §5.3 and ADR-0001 hazard 3 both call a defect, not a quirk. Both stores over-fetch instead: when metadata_filter is given, every stored vector's similarity is scored (:class:InMemoryVectorStore already does this — it is a linear scan by construction) and the filter is applied before truncating to top_k, so a top_k slice is only ever short when fewer than top_k chunks actually match. This is the same "correctness over cleverness, revisit if it's ever measured to matter" tradeoff index/bm25.py already accepts for its O(corpus) rebuild (ADR-0002) — filtered dense search costs O(corpus) instead of O(log corpus) per query; unfiltered search stays a cheap top-k vector query.

Score mapping. LanceDB reports a distance, not a similarity, and RetrievalResult.score is Field(ge=0.0) (contracts.py). Both stores compute cosine similarity and then floor it at zero via :func:_clamp_score — similarity is queried from LanceDB as 1 - cosine_distance (verified live: an identical vector pair reports _distance == 0.0, orthogonal reports 1.0, opposite reports 2.0, matching distance = 1 - similarity), or computed directly in :func:_cosine_similarity for the in-memory path. A negative similarity (near-opposite vectors) floors to 0.0 rather than crashing the contract or being silently coerced — ranking order among genuinely relevant, positive-similarity results is unaffected; only the below-zero tail collapses to a shared floor.

Determinism. Ties are broken by ascending Chunk.content_hash, the exact convention index/bm25.py already pins (its module docstring: tied scores fall back to insertion order only for two byte-identical chunks, which share a hash and can't be told apart by content alone). Both stores reconstruct full :class:~groundkit.contracts.Chunk objects internally (from stored rows, for LanceDB) specifically so this tie-break key and Pydantic's own offset/length re-validation come "for free" at read time.

No citation building here. search returns (Chunk, score) pairs, exactly as BM25Index.search does, and leaves the document-source join to retrieval/search.py. This is a correctness requirement, not symmetry for its own sake.

ingestion/chunking.py does seed every chunk's metadata with its document's source (metadata={"source": document.source, ...}), so a store could read source from chunk.metadata. It must not. That value is a snapshot taken at chunk time, while documents.source in SQLite is the durable truth (ADR-0002) that Retriever.search already joins against and fails closed on. Re-ingest a document from a new path and the two disagree — a dense hit would cite the stale path while a BM25 hit for the same document cites the current one, in a single response. Routing both seams through the same join makes that divergence unrepresentable.

It also keeps fusion simple: Wave C's RRF consumes two lists of the same shape, and the source join happens once, after fusion, on the surviving results.

InMemoryVectorStore

InMemoryVectorStore()

Pure-Python in-memory vector store — the dev/test double (ADR-0001 promote decision).

No lancedb import anywhere in this class; safe to use in a BM25-only install. Every search is an exhaustive linear scan over all stored vectors — by construction, not an optimization to revisit, since that is what makes its filter-then-truncate behavior trivially correct (see the module docstring). Appropriate for the corpus sizes this repo targets pre-Phase-6 scale work, matching index/bm25.py's identical O(corpus) tradeoff.

Satisfies :class:~groundkit.index.protocols.VectorStoreProtocol.

Source code in src/groundkit/index/dense.py

def __init__(self) -> None:
    self._chunks: list[Chunk] = []
    self._vectors: list[list[float]] = []
    self._dimensions: int | None = None

add async

add(
    chunks: list[Chunk], embeddings: list[list[float]]
) -> None

Store chunks with their embeddings.

Parameters:

Name	Type	Description	Default
chunks	list[Chunk]	Chunks to store.	required
embeddings	list[list[float]]	One embedding per chunk, same order.	required

Raises:

Type	Description
StorageError	chunks/embeddings length mismatch, embeddings in this call are not all the same width, or that width disagrees with a width already established by an earlier add() call on this store.

Source code in src/groundkit/index/dense.py

async def add(self, chunks: list[Chunk], embeddings: list[list[float]]) -> None:
    """Store chunks with their embeddings.

    Args:
        chunks: Chunks to store.
        embeddings: One embedding per chunk, same order.

    Raises:
        StorageError: ``chunks``/``embeddings`` length mismatch, embeddings
            in this call are not all the same width, or that width
            disagrees with a width already established by an earlier
            ``add()`` call on this store.
    """
    width = _validate_add_shapes(chunks, embeddings)
    if width is None:
        return
    if self._dimensions is not None and width != self._dimensions:
        raise _dimension_mismatch_error("embedding", width, self._dimensions)
    self._dimensions = width
    self._chunks.extend(chunks)
    self._vectors.extend(embeddings)

delete async

delete(document_id: str) -> int

Delete all vectors for document_id.

Parameters:

Name	Type	Description	Default
document_id	str	The document whose chunks should be removed.	required

Returns:

Type	Description
int	The number of vectors actually removed (counted, not assumed).

Raises:

Type	Description
StorageError	document_id fails :func:_validate_document_id.

Source code in src/groundkit/index/dense.py

async def delete(self, document_id: str) -> int:
    """Delete all vectors for ``document_id``.

    Args:
        document_id: The document whose chunks should be removed.

    Returns:
        The number of vectors actually removed (counted, not assumed).

    Raises:
        StorageError: ``document_id`` fails :func:`_validate_document_id`.
    """
    _validate_document_id(document_id)
    kept_chunks: list[Chunk] = []
    kept_vectors: list[list[float]] = []
    removed = 0
    for chunk, vector in zip(self._chunks, self._vectors, strict=True):
        if chunk.document_id == document_id:
            removed += 1
        else:
            kept_chunks.append(chunk)
            kept_vectors.append(vector)
    self._chunks = kept_chunks
    self._vectors = kept_vectors
    return removed

search async

search(
    query_embedding: list[float],
    top_k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[Chunk, float]]

Return the most similar chunks, most similar first.

The exhaustive scan runs in a worker thread, not on the caller's event loop. This method was async def with no await in its body: an await on it therefore never yielded, and a pure-Python cosine over every stored vector ran to completion before any other coroutine got the loop back. That is the same defect :meth:~groundkit.retrieval.search.Retriever._bm25_search documents, on the other always-in-process store.

The (chunk, vector) pairing is materialized here, on the calling thread, and only the scoring is dispatched. :meth:add extends the two backing lists in two separate statements, so a worker thread that zipped them itself could observe the first extended and the second not — turning a benign interleaving into a strict=True length mismatch. Pairing before dispatch keeps the snapshot exactly as consistent as the fully-inline version was, and it is cheap next to what moves: tuple construction per chunk versus a full-width dot product per chunk.

Parameters:

Name	Type	Description	Default
query_embedding	list[float]	The query vector.	required
top_k	int	Maximum number of results.	5
metadata_filter	dict[str, Any] | None	Keep only chunks whose metadata contains every given key/value pair; None means no filter. Applied before truncating to top_k (see the module docstring's "Filter-then-truncate" section) so a filtered search still returns top_k results whenever that many chunks match.	None

Returns:

Type	Description
list[tuple[Chunk, float]]	Up to top_k (chunk, score) pairs,
list[tuple[Chunk, float]]	ranked by descending score, ties broken by ascending
list[tuple[Chunk, float]]	Chunk.content_hash. top_k <= 0 or an empty store
list[tuple[Chunk, float]]	returns [].

Raises:

Type	Description
StorageError	query_embedding's width disagrees with this store's established width.

Source code in src/groundkit/index/dense.py

async def search(
    self,
    query_embedding: list[float],
    top_k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[Chunk, float]]:
    """Return the most similar chunks, most similar first.

    The exhaustive scan runs in a worker thread, not on the caller's
    event loop. This method was ``async def`` with no ``await`` in its
    body: an ``await`` on it therefore never yielded, and a pure-Python
    cosine over every stored vector ran to completion before any other
    coroutine got the loop back. That is the same defect
    :meth:`~groundkit.retrieval.search.Retriever._bm25_search` documents,
    on the other always-in-process store.

    The ``(chunk, vector)`` pairing is materialized *here*, on the
    calling thread, and only the scoring is dispatched. :meth:`add`
    extends the two backing lists in two separate statements, so a
    worker thread that zipped them itself could observe the first
    extended and the second not — turning a benign interleaving into a
    ``strict=True`` length mismatch. Pairing before dispatch keeps the
    snapshot exactly as consistent as the fully-inline version was,
    and it is cheap next to what moves: tuple construction per chunk
    versus a full-width dot product per chunk.

    Args:
        query_embedding: The query vector.
        top_k: Maximum number of results.
        metadata_filter: Keep only chunks whose metadata contains every
            given key/value pair; ``None`` means no filter. Applied
            before truncating to ``top_k`` (see the module docstring's
            "Filter-then-truncate" section) so a filtered search still
            returns ``top_k`` results whenever that many chunks match.

    Returns:
        Up to ``top_k`` ``(chunk, score)`` pairs,
        ranked by descending score, ties broken by ascending
        ``Chunk.content_hash``. ``top_k <= 0`` or an empty store
        returns ``[]``.

    Raises:
        StorageError: ``query_embedding``'s width disagrees with this
            store's established width.
    """
    if top_k <= 0 or not self._chunks:
        return []
    if self._dimensions is not None and len(query_embedding) != self._dimensions:
        raise _dimension_mismatch_error(
            "query embedding", len(query_embedding), self._dimensions
        )

    entries = list(zip(self._chunks, self._vectors, strict=True))
    return await asyncio.to_thread(
        self._score_sync, entries, query_embedding, top_k, metadata_filter
    )

LanceDBVectorStore

LanceDBVectorStore(
    db: Any,
    pa_module: Any,
    table_name: str,
    table: Any,
    dimensions: int | None,
)

Persisted vector store backed by LanceDB (ADR-0001 adapt decision).

Construct via :meth:open, not directly — mirrors index/metadata.py's SQLiteMetadataStore.open() convention. All LanceDB calls are synchronous; every one is run via asyncio.to_thread and serialized through one asyncio.Lock held across the awaited call, for the same reason SQLiteMetadataStore does this: concurrent coroutines can genuinely run these calls on different threads simultaneously, and the underlying table object is not documented as safe for that.

Satisfies :class:~groundkit.index.protocols.VectorStoreProtocol.

Source code in src/groundkit/index/dense.py

def __init__(
    self,
    db: Any,
    pa_module: Any,
    table_name: str,
    table: Any,
    dimensions: int | None,
) -> None:
    self._db = db
    self._pa = pa_module
    self._table_name = table_name
    self._table = table
    self._dimensions = dimensions
    self._lock = asyncio.Lock()

add async

add(
    chunks: list[Chunk], embeddings: list[list[float]]
) -> None

Store chunks with their embeddings.

Creates the backing table on the first call, with a schema fixed to this call's embedding width — LanceDB tables are fixed-width, so every later add() must agree with it.

Parameters:

Name	Type	Description	Default
chunks	list[Chunk]	Chunks to store.	required
embeddings	list[list[float]]	One embedding per chunk, same order.	required

Raises:

Type	Description
StorageError	chunks/embeddings length mismatch, embeddings in this call are not all the same width, or that width disagrees with this store's already-established width.

Source code in src/groundkit/index/dense.py

async def add(self, chunks: list[Chunk], embeddings: list[list[float]]) -> None:
    """Store chunks with their embeddings.

    Creates the backing table on the first call, with a schema fixed to
    this call's embedding width — LanceDB tables are fixed-width, so
    every later ``add()`` must agree with it.

    Args:
        chunks: Chunks to store.
        embeddings: One embedding per chunk, same order.

    Raises:
        StorageError: ``chunks``/``embeddings`` length mismatch, embeddings
            in this call are not all the same width, or that width
            disagrees with this store's already-established width.
    """
    width = _validate_add_shapes(chunks, embeddings)
    if width is None:
        return

    rows = [
        _row_from_chunk(chunk, vector) for chunk, vector in zip(chunks, embeddings, strict=True)
    ]
    async with self._lock:
        if self._table is None:
            self._table = await asyncio.to_thread(self._create_table_sync, width)
            self._dimensions = width
        elif self._dimensions is not None and width != self._dimensions:
            raise _dimension_mismatch_error("embedding", width, self._dimensions)
        await asyncio.to_thread(self._table.add, rows)

delete async

delete(document_id: str) -> int

Delete all vectors for document_id.

Parameters:

Name	Type	Description	Default
document_id	str	The document whose chunks should be removed.	required

Returns:

Type	Description
int	The number of vectors actually removed, computed by counting
int	rows before and after the delete (counted, not assumed) — never
int	read off LanceDB's own delete-result object, which is not
int	guaranteed present across this repo's supported lancedb
int	version range (>=0.18,<1).

Raises:

Type	Description
StorageError	document_id fails :func:_validate_document_id (ADR-0004 decision 6 — validated before it can reach :func:_build_delete_predicate).

Source code in src/groundkit/index/dense.py

async def delete(self, document_id: str) -> int:
    """Delete all vectors for ``document_id``.

    Args:
        document_id: The document whose chunks should be removed.

    Returns:
        The number of vectors actually removed, computed by counting
        rows before and after the delete (counted, not assumed) — never
        read off LanceDB's own delete-result object, which is not
        guaranteed present across this repo's supported ``lancedb``
        version range (``>=0.18,<1``).

    Raises:
        StorageError: ``document_id`` fails :func:`_validate_document_id`
            (ADR-0004 decision 6 — validated before it can reach
            :func:`_build_delete_predicate`).
    """
    predicate = _build_delete_predicate(document_id)
    async with self._lock:
        if self._table is None:
            return 0
        before = await asyncio.to_thread(self._table.count_rows)
        await asyncio.to_thread(self._table.delete, predicate)
        after = await asyncio.to_thread(self._table.count_rows)
    return int(before) - int(after)

explain_search async

explain_search(
    query_embedding: list[float],
    top_k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> str

Return LanceDB's physical plan for exactly the query :meth:search would run.

Built through :meth:_resolve_fetch_limit and :meth:_build_query, the same two calls :meth:search makes, so the plan cannot describe a different query than the app executes.

Raises:

Type	Description
StorageError	The table does not exist yet, query_embedding's width disagrees with this store's established width, or the call is one :meth:search answers with [] without running any query (top_k <= 0, or a filtered search on an empty table). There is no plan to report for those, and inventing one would describe a scan that never happens.

Source code in src/groundkit/index/dense.py

async def explain_search(
    self,
    query_embedding: list[float],
    top_k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> str:
    """Return LanceDB's physical plan for exactly the query :meth:`search` would run.

    Built through :meth:`_resolve_fetch_limit` and :meth:`_build_query`,
    the same two calls :meth:`search` makes, so the plan cannot describe
    a different query than the app executes.

    Raises:
        StorageError: The table does not exist yet, ``query_embedding``'s
            width disagrees with this store's established width, or the
            call is one :meth:`search` answers with ``[]`` without running
            any query (``top_k <= 0``, or a filtered search on an empty
            table). There is no plan to report for those, and inventing
            one would describe a scan that never happens.
    """
    async with self._lock:
        if self._table is None:
            raise StorageError("cannot explain a search on a store with no table yet")
        if self._dimensions is not None and len(query_embedding) != self._dimensions:
            raise _dimension_mismatch_error(
                "query embedding", len(query_embedding), self._dimensions
            )
        if top_k <= 0:
            raise StorageError(
                f"top_k must be positive to explain a search; search(top_k={top_k}) "
                "returns [] without running a query"
            )
        limit, exhaustive = await self._resolve_fetch_limit(top_k, metadata_filter)
        if limit <= 0:
            raise StorageError(
                "search would return [] without running a query (empty table); "
                "there is no plan to explain"
            )
        query = self._build_query(query_embedding, limit, exhaustive=exhaustive)
        plan: str = await asyncio.to_thread(query.explain_plan, True)
    return plan

open async classmethod

open(
    db_path: str | Path,
    table_name: str = _DEFAULT_TABLE_NAME,
) -> LanceDBVectorStore

Open (creating if absent) the LanceDB-backed store at db_path.

Parameters:

Name	Type	Description	Default
db_path	str | Path	Directory LanceDB should use for this collection's table(s). Created on first write if it does not exist.	required
table_name	str	Table name within that directory.	_DEFAULT_TABLE_NAME

Returns:

Type	Description
LanceDBVectorStore	A ready-to-use store. If a table named table_name already
LanceDBVectorStore	exists at db_path (a prior session's data), it is opened
LanceDBVectorStore	and this store's dimensions are inferred from its schema;
LanceDBVectorStore	otherwise the table is created lazily on the first :meth:add.

Raises:

Type	Description
StorageError	The optional dense extra (lancedb, pyarrow) is not installed, or the LanceDB connection could not be opened.

Source code in src/groundkit/index/dense.py

@classmethod
async def open(
    cls, db_path: str | Path, table_name: str = _DEFAULT_TABLE_NAME
) -> LanceDBVectorStore:
    """Open (creating if absent) the LanceDB-backed store at ``db_path``.

    Args:
        db_path: Directory LanceDB should use for this collection's
            table(s). Created on first write if it does not exist.
        table_name: Table name within that directory.

    Returns:
        A ready-to-use store. If a table named ``table_name`` already
        exists at ``db_path`` (a prior session's data), it is opened
        and this store's dimensions are inferred from its schema;
        otherwise the table is created lazily on the first :meth:`add`.

    Raises:
        StorageError: The optional ``dense`` extra (``lancedb``,
            ``pyarrow``) is not installed, or the LanceDB connection
            could not be opened.
    """
    lancedb, pa = _import_lancedb()

    def _open_sync() -> tuple[Any, Any, int | None]:
        db = lancedb.connect(str(db_path))
        try:
            table = db.open_table(table_name)
        except ValueError:
            # LanceDB raises ValueError, not a missing-file OSError, for
            # "no table with this name yet" — verified live against the
            # pinned 0.37.1 client. Treated as "create lazily on add()",
            # not an error.
            table = None
        dimensions = _infer_dimensions(table) if table is not None else None
        return db, table, dimensions

    try:
        db, table, dimensions = await asyncio.to_thread(_open_sync)
    except OSError as exc:
        raise StorageError(f"failed to open LanceDB store at {db_path}") from exc
    return cls(db, pa, table_name, table, dimensions)

search async

search(
    query_embedding: list[float],
    top_k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[Chunk, float]]

Return the most similar chunks, most similar first.

Parameters:

Name	Type	Description	Default
query_embedding	list[float]	The query vector.	required
top_k	int	Maximum number of results.	5
metadata_filter	dict[str, Any] | None	Keep only chunks whose metadata contains every given key/value pair; None means no filter. When set, every stored row is fetched and filtered in Python before truncating to top_k (see the module docstring's "Filter-then-truncate" section) — metadata is an opaque JSON blob column here, not structured columns, so pushing an arbitrary filter into a LanceDB WHERE predicate would mean building that predicate from caller-controlled values, the exact class of hazard ADR-0004 decision 6 exists to close for document IDs. Over-fetching and filtering in Python avoids ever constructing one.	None

Returns:

Type	Description
list[tuple[Chunk, float]]	Up to top_k (chunk, score) pairs,
list[tuple[Chunk, float]]	ranked by descending score, ties broken by ascending
list[tuple[Chunk, float]]	Chunk.content_hash. top_k <= 0 or an empty/absent table
list[tuple[Chunk, float]]	returns [].

Raises:

Type	Description
StorageError	query_embedding's width disagrees with this store's established width.

Source code in src/groundkit/index/dense.py

async def search(
    self,
    query_embedding: list[float],
    top_k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[Chunk, float]]:
    """Return the most similar chunks, most similar first.

    Args:
        query_embedding: The query vector.
        top_k: Maximum number of results.
        metadata_filter: Keep only chunks whose metadata contains every
            given key/value pair; ``None`` means no filter. When set,
            every stored row is fetched and filtered in Python before
            truncating to ``top_k`` (see the module docstring's
            "Filter-then-truncate" section) — metadata is an opaque
            JSON blob column here, not structured columns, so pushing
            an arbitrary filter into a LanceDB ``WHERE`` predicate would
            mean building that predicate from caller-controlled values,
            the exact class of hazard ADR-0004 decision 6 exists to
            close for document IDs. Over-fetching and filtering in
            Python avoids ever constructing one.

    Returns:
        Up to ``top_k`` ``(chunk, score)`` pairs,
        ranked by descending score, ties broken by ascending
        ``Chunk.content_hash``. ``top_k <= 0`` or an empty/absent table
        returns ``[]``.

    Raises:
        StorageError: ``query_embedding``'s width disagrees with this
            store's established width.
    """
    if top_k <= 0:
        return []

    async with self._lock:
        if self._table is None:
            return []
        if self._dimensions is not None and len(query_embedding) != self._dimensions:
            raise _dimension_mismatch_error(
                "query embedding", len(query_embedding), self._dimensions
            )
        fetch_limit, exhaustive = await self._resolve_fetch_limit(top_k, metadata_filter)
        if fetch_limit <= 0:
            return []
        rows = await asyncio.to_thread(
            partial(self._search_sync, exhaustive=exhaustive), query_embedding, fetch_limit
        )

    scored: list[tuple[Chunk, float]] = []
    for row in rows:
        chunk = _chunk_from_row(row)
        if not _matches_filter(chunk.metadata, metadata_filter):
            continue
        scored.append((chunk, _clamp_score(1.0 - float(row["_distance"]))))

    ranked = _sort_by_score(scored)
    return ranked[:top_k]

vector_index_stats async

vector_index_stats() -> tuple[str, int] | None

(distance_type, unindexed_rows) for the vector column's index, or None.

distance_type is lower-cased ("cosine", "l2", ...). unindexed_rows counts rows added after the index was built: they are scored by brute force and unioned with the ANN results — correct, but a growing flat-scanned tail. Incremental ingest produces exactly this.

Source code in src/groundkit/index/dense.py

async def vector_index_stats(self) -> tuple[str, int] | None:
    """``(distance_type, unindexed_rows)`` for the vector column's index, or ``None``.

    ``distance_type`` is lower-cased (``"cosine"``, ``"l2"``, ...).
    ``unindexed_rows`` counts rows added after the index was built: they
    are scored by brute force and unioned with the ANN results — correct,
    but a growing flat-scanned tail. Incremental ingest produces exactly
    this.
    """
    async with self._lock:
        if self._table is None:
            return None
        indices = await asyncio.to_thread(self._table.list_indices)
        for index in indices:
            if _VECTOR_COLUMN in index.columns:
                stats = await asyncio.to_thread(self._table.index_stats, index.name)
                return str(stats.distance_type).lower(), int(stats.num_unindexed_rows)
    return None

VectorIndexFallbackError

Bases: StorageError

An unfiltered dense search would not be served by the vector index.

Raised by :func:assert_search_uses_vector_index. A flat scan returns correct results, so nothing else notices: latency grows with the corpus and no test or log line fails. This error is the loud version.

assert_search_uses_vector_index async

assert_search_uses_vector_index(
    store: LanceDBVectorStore,
    query_embedding: list[float],
    top_k: int = 5,
    *,
    max_unindexed_rows: int = 0,
) -> str

Fail loudly unless :meth:LanceDBVectorStore.search would use the ANN index.

Checks the plan of the query the app actually runs (via :meth:LanceDBVectorStore.explain_search), not a hand-written one. Three conditions must hold:

1. The vector index, if any, was built for the query's cosine metric. Checked from the index's own stats, not the plan, because a mismatch fails differently across the supported LanceDB range: 0.37 logs a warning and brute-force scans, but 0.18 uses an L2 index for a cosine query and ranks results by the wrong metric — verified live, and the plan line there does not name the metric at all.
2. The plan contains an ANN index operator. It will not when no index exists, or on a metric mismatch under newer LanceDB.
3. At most max_unindexed_rows rows sit outside the index. Rows added after the index was built are brute-force scored and unioned in, so a stale index degrades gradually rather than all at once.

Only the unfiltered path is checked. A filtered search is an exhaustive scan by design (see the module docstring's "Filter-then-truncate" section and :meth:LanceDBVectorStore._build_query).

The plan and the index stats are read under separate lock acquisitions, so a concurrent add() can land between them and the two can describe slightly different table states. That is harmless for a diagnostic meant to run against a quiescent collection (CI, a deploy gate).

Parameters:

Name	Type	Description	Default
store	LanceDBVectorStore	An opened LanceDB store with a populated table.	required
query_embedding	list[float]	A representative query vector of the store's width.	required
top_k	int	The top_k the app searches with.	5
max_unindexed_rows	int	Unindexed-row allowance before the check fails.	0

Returns:

Type	Description
str	The verified plan, for logging.

Raises:

Type	Description
VectorIndexFallbackError	Either condition above fails. The message carries the full plan.
StorageError	The store has no table, the query width is wrong, top_k is not positive, or max_unindexed_rows is negative. These are invalid inputs, never a fallback verdict: a negative allowance would report zero unindexed rows as stale.

Source code in src/groundkit/index/dense.py

async def assert_search_uses_vector_index(
    store: LanceDBVectorStore,
    query_embedding: list[float],
    top_k: int = 5,
    *,
    max_unindexed_rows: int = 0,
) -> str:
    """Fail loudly unless :meth:`LanceDBVectorStore.search` would use the ANN index.

    Checks the plan of the query the app actually runs (via
    :meth:`LanceDBVectorStore.explain_search`), not a hand-written one. Three
    conditions must hold:

    1. The vector index, if any, was built for the query's cosine metric.
       Checked from the index's own stats, not the plan, because a mismatch
       fails differently across the supported LanceDB range: 0.37 logs a
       warning and brute-force scans, but 0.18 *uses* an L2 index for a
       cosine query and ranks results by the wrong metric — verified live,
       and the plan line there does not name the metric at all.
    2. The plan contains an ANN index operator. It will not when no index
       exists, or on a metric mismatch under newer LanceDB.
    3. At most ``max_unindexed_rows`` rows sit outside the index. Rows added
       after the index was built are brute-force scored and unioned in, so
       a stale index degrades gradually rather than all at once.

    Only the unfiltered path is checked. A filtered search is an exhaustive
    scan by design (see the module docstring's "Filter-then-truncate"
    section and :meth:`LanceDBVectorStore._build_query`).

    The plan and the index stats are read under separate lock acquisitions,
    so a concurrent ``add()`` can land between them and the two can describe
    slightly different table states. That is harmless for a diagnostic
    meant to run against a quiescent collection (CI, a deploy gate).

    Args:
        store: An opened LanceDB store with a populated table.
        query_embedding: A representative query vector of the store's width.
        top_k: The ``top_k`` the app searches with.
        max_unindexed_rows: Unindexed-row allowance before the check fails.

    Returns:
        The verified plan, for logging.

    Raises:
        VectorIndexFallbackError: Either condition above fails. The message
            carries the full plan.
        StorageError: The store has no table, the query width is wrong,
            ``top_k`` is not positive, or ``max_unindexed_rows`` is negative.
            These are invalid inputs, never a fallback verdict: a negative
            allowance would report zero unindexed rows as stale.
    """
    if max_unindexed_rows < 0:
        raise StorageError(f"max_unindexed_rows must be >= 0, got {max_unindexed_rows}")
    plan = await store.explain_search(query_embedding, top_k=top_k)
    stats = await store.vector_index_stats()
    if stats is not None and stats[0] != _DISTANCE_METRIC:
        raise VectorIndexFallbackError(
            f"Vector index on {_VECTOR_COLUMN!r} was built for metric {stats[0]!r}, but "
            f"search queries with {_DISTANCE_METRIC!r}. Depending on the LanceDB "
            "version this either brute-force scans every row or ranks results by the "
            f"wrong metric. Rebuild the index with {_DISTANCE_METRIC!r}. Plan:\n{plan}"
        )
    if stats is None:
        raise VectorIndexFallbackError(
            "Dense search would brute-force scan every row instead of using a vector "
            f"index: no index exists on the {_VECTOR_COLUMN!r} column. Plan:\n{plan}"
        )
    if _ANN_OPERATOR_RE.search(plan) is None:
        raise VectorIndexFallbackError(
            f"A vector index exists on {_VECTOR_COLUMN!r} (metric {stats[0]!r}, "
            f"{stats[1]} unindexed row(s)) but the query plan has no {_ANN_PLAN_MARKER} "
            "operator, so search is not served by it. Plan:\n" + plan
        )
    unindexed = stats[1]
    if unindexed > max_unindexed_rows:
        raise VectorIndexFallbackError(
            f"Vector index is stale: {unindexed} row(s) are outside it and are "
            f"brute-force scanned on every search (allowance: {max_unindexed_rows}). "
            f"Rebuild or optimize the index. Plan:\n{plan}"
        )
    return plan

verify_dense_side_present async

verify_dense_side_present(
    store: MetadataStoreProtocol,
    vector_store: VectorStoreProtocol,
    dimensions: int,
    *,
    manifest: CollectionManifest | None,
) -> None

Fail closed when a dense-bound collection has lost its vectors.

The dense analogue of the incremental skip key's blind spot. SQLite is the durable truth for documents and chunks (ADR-0002) and its content_hash decides what gets re-indexed — but that hash records only whether the content changed, never whether the vectors derived from it still exist. A collection whose SQLite file persisted while its vectors did not therefore reports every document as unchanged, re-embeds nothing, and returns nothing from the dense side, permanently and with no error: exactly the silent-absence class SPEC.md §2 fails closed against.

Two ways to reach that state. The library one is pairing an :class:InMemoryVectorStore with a file-backed :class:~groundkit.index.metadata.SQLiteMetadataStore, where the dense side empties on every restart by construction (the grk CLI cannot: it wires :class:LanceDBVectorStore unconditionally). The operational one, and the likelier one, is the collection's .lance directory being deleted, moved, or restored from a backup that omitted it.

ADR-0004's manifest is what makes the check unambiguous: it is written on the collection's first dense write, so a bound manifest proves vectors once existed. Manifest bound + documents in SQLite + an empty vector store is a lost dense side, and is distinguishable from the legitimate cases — an unbound manifest means the collection was only ever BM25-only (whose upgrade path is the documented no-backfill limitation, not a defect), and no documents means an empty collection.

Emptiness is probed with search(top_k=1) rather than a count: a dense search never drops zero-scored results, so an empty result means an empty store, and this needs no addition to :class:~groundkit.index.protocols.VectorStoreProtocol.

Parameters:

Name	Type	Description	Default
store	MetadataStoreProtocol	The collection's metadata store.	required
vector_store	VectorStoreProtocol	The dense store paired with it.	required
dimensions	int	Embedding width, used to shape the probe vector.	required
manifest	CollectionManifest | None	The collection's manifest as the caller last read it, or None if it has none — keyword-only, and deliberately passed in rather than read here. Every caller reaches this function immediately after store.verify_manifest, which returns the manifest it checked; reusing that value keeps the whole open/ingest preamble reasoning about one manifest snapshot. A second read here could observe a newly bound manifest whose first vectors have not landed yet and report a healthy collection as a lost dense side.	required

Raises:

Type	Description
StorageError	The collection is manifest-bound and holds documents, but its vector store is empty.

Source code in src/groundkit/index/dense.py

async def verify_dense_side_present(
    store: MetadataStoreProtocol,
    vector_store: VectorStoreProtocol,
    dimensions: int,
    *,
    manifest: CollectionManifest | None,
) -> None:
    """Fail closed when a dense-bound collection has lost its vectors.

    The dense analogue of the incremental skip key's blind spot. SQLite is
    the durable truth for documents and chunks (ADR-0002) and its
    ``content_hash`` decides what gets re-indexed — but that hash records
    only whether the *content* changed, never whether the vectors derived
    from it still exist. A collection whose SQLite file persisted while its
    vectors did not therefore reports every document as unchanged, re-embeds
    nothing, and returns nothing from the dense side, permanently and with
    no error: exactly the silent-absence class SPEC.md §2 fails closed
    against.

    Two ways to reach that state. The library one is pairing an
    :class:`InMemoryVectorStore` with a file-backed
    :class:`~groundkit.index.metadata.SQLiteMetadataStore`, where the dense
    side empties on every restart by construction (the ``grk`` CLI cannot:
    it wires :class:`LanceDBVectorStore` unconditionally). The operational
    one, and the likelier one, is the collection's ``.lance`` directory
    being deleted, moved, or restored from a backup that omitted it.

    ADR-0004's manifest is what makes the check unambiguous: it is written
    on the collection's *first dense write*, so a bound manifest proves
    vectors once existed. Manifest bound + documents in SQLite + an empty
    vector store is a lost dense side, and is distinguishable from the
    legitimate cases — an unbound manifest means the collection was only
    ever BM25-only (whose upgrade path is the documented no-backfill
    limitation, not a defect), and no documents means an empty collection.

    Emptiness is probed with ``search(top_k=1)`` rather than a count: a
    dense search never drops zero-scored results, so an empty result means
    an empty store, and this needs no addition to
    :class:`~groundkit.index.protocols.VectorStoreProtocol`.

    Args:
        store: The collection's metadata store.
        vector_store: The dense store paired with it.
        dimensions: Embedding width, used to shape the probe vector.
        manifest: The collection's manifest as the caller last read it, or
            ``None`` if it has none — keyword-only, and deliberately passed
            in rather than read here. Every caller reaches this function
            immediately after ``store.verify_manifest``, which returns the
            manifest it checked; reusing that value keeps the whole
            open/ingest preamble reasoning about one manifest snapshot. A
            second read here could observe a *newly* bound manifest whose
            first vectors have not landed yet and report a healthy
            collection as a lost dense side.

    Raises:
        StorageError: The collection is manifest-bound and holds documents,
            but its vector store is empty.
    """
    if manifest is None:
        return
    sources = await store.get_document_sources()
    if not sources:
        return

    # A unit vector rather than zeros: cosine similarity is undefined
    # against a zero-magnitude query.
    probe = [1.0] + [0.0] * (dimensions - 1)
    if await vector_store.search(probe, top_k=1):
        return

    raise StorageError(
        f"Dense side is empty for a collection bound to an embedding identity with "
        f"{len(sources)} document(s) still in SQLite. The vectors were lost while the "
        "metadata store survived (an ephemeral vector store paired with a persisted "
        "SQLite store, or a deleted/moved LanceDB directory). Refusing to continue: "
        "the incremental skip gate would report every document unchanged, re-embed "
        "nothing, and return nothing from the dense side, silently. Rebuild the "
        "collection, or restore the vector store alongside its SQLite file."
    )

Metadata store

The durable truth for documents and chunks (ADR-0002): one SQLite connection per collection, foreign_keys=ON, journal_mode=WAL, every call serialized through a single asyncio.Lock held across the awaited call — the lock, not the flag, is what makes concurrent awaits safe against sqlite3's own thread-safety limits. Also owns the ADR-0004 collection manifest, which binds a collection to the embedding (provider, model_name, dimensions) triple it was built with; see Configuration for the identity model itself.

groundkit.index.metadata

Document/chunk metadata store backed by SQLite, behind an interface (Phase 1).

Implements :class:~groundkit.index.protocols.MetadataStoreProtocol. SQLite is the durable truth for documents and chunks (ADR-0002); the BM25 index is rebuilt from this store at process start rather than persisted itself.

Concurrency model (ADR-0002): a single sqlite3.Connection opened with check_same_thread=False, with every call serialized through one asyncio.Lock. sqlite3 connections are not safe to use concurrently from multiple threads, and asyncio.to_thread hands each call to a fresh worker thread — the lock, held for the duration of the coroutine (including the awaited to_thread call), guarantees only one thread ever touches the connection at a time. This was chosen over per-call connections to avoid repeated file-handle churn and "database is locked" contention under WAL for groundkit's expected single-process local deployment.

Also implements the ADR-0004 collection manifest: a single-row collection_manifest table binding a collection to the embedding (provider, model_name, dimensions) triple it was built with, plus the PRAGMA application_id/user_version stamp that lets :meth:open recognize a store created before that manifest existed and refuse it for dense work rather than trust an identity it never recorded.

SQLiteMetadataStore

SQLiteMetadataStore(
    connection: Connection,
    db_path: Path,
    *,
    schema_current: bool,
)

Durable store for documents, chunks, and ingest state, backed by SQLite.

Construct via :meth:open, not directly — the constructor takes an already-configured connection and its precomputed schema-currency verdict.

Attributes:

Name	Type	Description
db_path		Path to the collection's SQLite file, kept for diagnostics.

Source code in src/groundkit/index/metadata.py

def __init__(
    self, connection: sqlite3.Connection, db_path: Path, *, schema_current: bool
) -> None:
    self._conn = connection
    self._lock = asyncio.Lock()
    self.db_path = db_path
    #: True when this store's PRAGMA application_id/user_version match
    #: groundkit's current stamp exactly (set by :meth:`open`). False for
    #: a store written against an older schema, one that predates the
    #: ADR-0004 stamp entirely, or one groundkit never created.
    #:
    #: Named for the schema rather than for the manifest because it now
    #: gates two unrelated capabilities: :meth:`write_manifest` and
    #: :meth:`verify_manifest` refuse a non-current store (ADR-0004), and
    #: :meth:`get_generation` reports "freshness unanswerable" for one
    #: (ADR-0013). Such a store still works for BM25-only reads and
    #: writes; it is uncacheable and closed to dense work, not broken.
    self._schema_current = schema_current

add_chunks async

add_chunks(chunks: list[Chunk], source: str) -> None

Persist chunks for a document.

Parameters:

Name	Type	Description	Default
chunks	list[Chunk]	Chunks to persist. A no-op if empty.	required
source	str	The source identifier the document was upserted under — used to resolve the expected document_id so a caller cannot silently attach chunks to the wrong document.	required

Raises:

Type	Description
StorageError	source has no upserted document, a chunk's document_id does not match the document registered for source, or a backend failure occurs (including a duplicate chunk_id).

Source code in src/groundkit/index/metadata.py

async def add_chunks(self, chunks: list[Chunk], source: str) -> None:
    """Persist chunks for a document.

    Args:
        chunks: Chunks to persist. A no-op if empty.
        source: The source identifier the document was upserted under —
            used to resolve the expected ``document_id`` so a caller
            cannot silently attach chunks to the wrong document.

    Raises:
        StorageError: ``source`` has no upserted document, a chunk's
            ``document_id`` does not match the document registered for
            ``source``, or a backend failure occurs (including a
            duplicate ``chunk_id``).
    """
    if not chunks:
        return

    def _op() -> None:
        cur = self._conn.execute(
            "SELECT document_id FROM documents WHERE source = ?", (source,)
        )
        row = cur.fetchone()
        if row is None:
            raise StorageError(f"cannot add chunks: no document upserted for source {source!r}")
        expected_document_id = row[0]
        for chunk in chunks:
            if chunk.document_id != expected_document_id:
                raise StorageError(
                    f"chunk.document_id {chunk.document_id!r} does not match "
                    f"document_id {expected_document_id!r} registered for source {source!r}"
                )
        self._insert_chunks(chunks)
        self._bump_generation()
        self._conn.commit()

    await self._run(_op)

close async

close() -> None

Close the underlying connection.

Routed through :meth:_run rather than a bare async with self._lock: await asyncio.to_thread(self._conn.close), for the same cancellation reason every other method on this class goes through it: cancelling that plain async with block does not stop the worker thread mid-close() -- it only unwinds the awaiting coroutine, and __aexit__ would release the lock right then, while the worker was still inside sqlite3.Connection.close() on a second thread. A coroutine that then acquired the lock could touch the same check_same_thread=False connection concurrently -- exactly the undefined behaviour the lock exists to rule out (see this module's header docstring: "the lock, not the flag, is what makes concurrent awaits safe"). _run's add_done_callback-gated release ties the lock to the worker's actual completion instead, so that race cannot open here either.

_run's rollback-on-exception wrapper is safe to run for a close: the only way it fires is if self._conn.close() itself raises, and if that leaves the connection already unusable, the follow-up rollback() attempt is caught and logged rather than allowed to replace the original exception (see _run's _guarded) -- the same guard that already covers "operation racing store teardown" for every other method.

Source code in src/groundkit/index/metadata.py

async def close(self) -> None:
    """Close the underlying connection.

    Routed through :meth:`_run` rather than a bare
    ``async with self._lock: await asyncio.to_thread(self._conn.close)``,
    for the same cancellation reason every other method on this class
    goes through it: cancelling that plain ``async with`` block does not
    stop the worker thread mid-``close()`` -- it only unwinds the
    *awaiting* coroutine, and ``__aexit__`` would release the lock right
    then, while the worker was still inside ``sqlite3.Connection.close()``
    on a second thread. A coroutine that then acquired the lock could
    touch the same ``check_same_thread=False`` connection concurrently --
    exactly the undefined behaviour the lock exists to rule out (see this
    module's header docstring: "the lock, not the flag, is what makes
    concurrent awaits safe"). ``_run``'s ``add_done_callback``-gated
    release ties the lock to the worker's actual completion instead, so
    that race cannot open here either.

    ``_run``'s rollback-on-exception wrapper is safe to run for a close:
    the only way it fires is if ``self._conn.close()`` itself raises, and
    if that leaves the connection already unusable, the follow-up
    ``rollback()`` attempt is caught and logged rather than allowed to
    replace the original exception (see ``_run``'s ``_guarded``) -- the
    same guard that already covers "operation racing store teardown"
    for every other method.
    """
    await self._run(self._conn.close)
    logger.debug("closed metadata store")

count_chunks async

count_chunks() -> int

Return how many chunks are persisted, without reading any of them.

Deliberately not a :class:~groundkit.index.protocols.MetadataStoreProtocol member. That protocol is held to exact signature parity by tests/test_protocol_conformance.py and implemented by hand-built doubles across the suite; widening it for a reporting convenience is the trade ADR-0012 decision 3 already refused for model_name. It is instead declared on the optional :class:~groundkit.index.protocols.DocumentRecordStoreProtocol (GK-019), which only a store that can genuinely answer cheaply implements — so the capability is now stated in the type system rather than reached for by holding the concrete class.

What it replaces: len(await store.get_chunks()), which materialized every chunk's full text — the entire corpus, in memory, as :class:~groundkit.contracts.Chunk models, each re-validated on construction — to produce one integer. That is the read behind index_status, the cheapest-looking call on an unauthenticated service surface, so its cost was both the largest on that surface and the least apparent from its name.

Returns:

Type	Description
int	The number of rows in chunks.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def count_chunks(self) -> int:
    """Return how many chunks are persisted, without reading any of them.

    Deliberately **not** a :class:`~groundkit.index.protocols.MetadataStoreProtocol`
    member. That protocol is held to exact signature parity by
    ``tests/test_protocol_conformance.py`` and implemented by hand-built
    doubles across the suite; widening it for a reporting
    convenience is the trade ADR-0012 decision 3 already refused for
    ``model_name``. It is instead declared on the *optional*
    :class:`~groundkit.index.protocols.DocumentRecordStoreProtocol`
    (GK-019), which only a store that can genuinely answer cheaply
    implements — so the capability is now stated in the type system
    rather than reached for by holding the concrete class.

    What it replaces: ``len(await store.get_chunks())``, which
    materialized every chunk's full text — the entire corpus, in memory,
    as :class:`~groundkit.contracts.Chunk` models, each re-validated on
    construction — to produce one integer. That is the read behind
    ``index_status``, the cheapest-looking call on an unauthenticated
    service surface, so its cost was both the largest on that surface
    and the least apparent from its name.

    Returns:
        The number of rows in ``chunks``.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> int:
        cur = self._conn.execute("SELECT COUNT(*) FROM chunks")
        return int(cur.fetchone()[0])

    return await self._run(_op)

count_documents async

count_documents() -> int

Return how many documents are persisted, without reading their sources.

The sibling of :meth:count_chunks, and off :class:~groundkit.index.protocols.MetadataStoreProtocol for the same reason. It replaces len(await store.get_document_sources()), which built a full {document_id: source} dict so that index_status could take its length — the same materialize-to-measure shape as the chunk half, one line above it in service/tools.py and missed when that half was fixed.

Declared on the optional :class:~groundkit.index.protocols.DocumentRecordStoreProtocol since GK-019, alongside the keyed document read: both describe the same property of a store — that it can answer a bounded question without materializing a table — and neither belongs on the protocol every store must satisfy.

Returns:

Type	Description
int	The number of rows in documents.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def count_documents(self) -> int:
    """Return how many documents are persisted, without reading their sources.

    The sibling of :meth:`count_chunks`, and off
    :class:`~groundkit.index.protocols.MetadataStoreProtocol` for the same
    reason. It replaces ``len(await store.get_document_sources())``, which
    built a full ``{document_id: source}`` dict so that ``index_status``
    could take its length — the same materialize-to-measure shape as the
    chunk half, one line above it in ``service/tools.py`` and missed when
    that half was fixed.

    Declared on the optional
    :class:`~groundkit.index.protocols.DocumentRecordStoreProtocol` since
    GK-019, alongside the keyed document read: both describe the same
    property of a store — that it can answer a bounded question without
    materializing a table — and neither belongs on the protocol every
    store must satisfy.

    Returns:
        The number of rows in ``documents``.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> int:
        cur = self._conn.execute("SELECT COUNT(*) FROM documents")
        return int(cur.fetchone()[0])

    return await self._run(_op)

delete_document async

delete_document(document_id: str) -> int

Delete a document and its chunks.

Chunks are deleted explicitly rather than relying solely on the ON DELETE CASCADE foreign key, so the deleted-chunk count is known without a second query.

Parameters:

Name	Type	Description	Default
document_id	str	The document's unique identifier.	required

Returns:

Type	Description
int	The number of chunks deleted (0 if the document did not exist).

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def delete_document(self, document_id: str) -> int:
    """Delete a document and its chunks.

    Chunks are deleted explicitly rather than relying solely on the
    ``ON DELETE CASCADE`` foreign key, so the deleted-chunk count is
    known without a second query.

    Args:
        document_id: The document's unique identifier.

    Returns:
        The number of chunks deleted (0 if the document did not exist).

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> int:
        cur = self._conn.execute(
            "SELECT COUNT(*) FROM chunks WHERE document_id = ?", (document_id,)
        )
        count = int(cur.fetchone()[0])
        self._conn.execute("DELETE FROM chunks WHERE document_id = ?", (document_id,))
        self._conn.execute("DELETE FROM documents WHERE document_id = ?", (document_id,))
        self._bump_generation()
        self._conn.commit()
        return count

    return await self._run(_op)

get_chunk async

get_chunk(chunk_id: str) -> Chunk | None

Return one chunk by ID, or None.

Parameters:

Name	Type	Description	Default
chunk_id	str	The chunk's unique identifier.	required

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_chunk(self, chunk_id: str) -> Chunk | None:
    """Return one chunk by ID, or None.

    Args:
        chunk_id: The chunk's unique identifier.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> Chunk | None:
        cur = self._conn.execute(
            "SELECT chunk_id, document_id, chunk_index, content, start_offset, "
            "end_offset, metadata FROM chunks WHERE chunk_id = ?",
            (chunk_id,),
        )
        row = cur.fetchone()
        return _row_to_chunk(row) if row is not None else None

    return await self._run(_op)

get_chunks async

get_chunks() -> list[Chunk]

Return all persisted chunks in the collection.

Returns:

Type	Description
list[Chunk]	Chunks ordered by document then position, for deterministic
list[Chunk]	downstream indexing (e.g. BM25 rebuild).

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_chunks(self) -> list[Chunk]:
    """Return all persisted chunks in the collection.

    Returns:
        Chunks ordered by document then position, for deterministic
        downstream indexing (e.g. BM25 rebuild).

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> list[Chunk]:
        cur = self._conn.execute(
            "SELECT chunk_id, document_id, chunk_index, content, start_offset, "
            "end_offset, metadata FROM chunks ORDER BY document_id, chunk_index"
        )
        return [_row_to_chunk(row) for row in cur.fetchall()]

    return await self._run(_op)

get_document_hash async

get_document_hash(source: str) -> str | None

Return the stored content hash for source, or None if unseen.

Parameters:

Name	Type	Description	Default
source	str	File path, URL, or other source identifier.	required

Returns:

Type	Description
str | None	The stored content hash, or None if source has never
str | None	been ingested.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_document_hash(self, source: str) -> str | None:
    """Return the stored content hash for ``source``, or None if unseen.

    Args:
        source: File path, URL, or other source identifier.

    Returns:
        The stored content hash, or ``None`` if ``source`` has never
        been ingested.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> str | None:
        cur = self._conn.execute(
            "SELECT content_hash FROM documents WHERE source = ?", (source,)
        )
        row = cur.fetchone()
        return str(row[0]) if row is not None else None

    return await self._run(_op)

get_document_id async

get_document_id(source: str) -> str | None

Return the stored document ID for source, or None if unseen.

Parameters:

Name	Type	Description	Default
source	str	File path, URL, or other source identifier.	required

Returns:

Type	Description
str | None	The stored document ID, or None if source has never
str | None	been ingested.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_document_id(self, source: str) -> str | None:
    """Return the stored document ID for ``source``, or None if unseen.

    Args:
        source: File path, URL, or other source identifier.

    Returns:
        The stored document ID, or ``None`` if ``source`` has never
        been ingested.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> str | None:
        cur = self._conn.execute(
            "SELECT document_id FROM documents WHERE source = ?", (source,)
        )
        row = cur.fetchone()
        return str(row[0]) if row is not None else None

    return await self._run(_op)

get_document_record async

get_document_record(
    document_id: str,
) -> DocumentRecord | None

Return one document's :class:DocumentRecord, or None if there is no such row.

The keyed form of :meth:get_document_records, added by GK-019. document_id is the documents primary key, so this is an index seek: what it replaces is a full table scan that built a validated model per stored row so that its two callers — :meth:~groundkit.retrieval.search.Retriever.search, on every query, and service.tools.handle_fetch_chunk, for a single ID — could look up at most top_k keys in the result. Cost scaled with the corpus while the question scaled with top_k.

Returning None for an unknown ID rather than raising is what lets those callers keep their fail-closed branch: a chunk whose document row is gone must be refused, and refusing is their decision to make with their own error vocabulary, not this store's.

Parameters:

Name	Type	Description	Default
document_id	str	The document's unique identifier.	required

Raises:

Type	Description
StorageError	This store predates the source_class/ extractor columns (schema v3, ADR-0016), or a backend failure occurs.

Source code in src/groundkit/index/metadata.py

async def get_document_record(self, document_id: str) -> DocumentRecord | None:
    """Return one document's :class:`DocumentRecord`, or ``None`` if there is no such row.

    The keyed form of :meth:`get_document_records`, added by GK-019.
    ``document_id`` is the ``documents`` primary key, so this is an index
    seek: what it replaces is a full table scan that built a validated
    model per stored row so that its two callers —
    :meth:`~groundkit.retrieval.search.Retriever.search`, on **every**
    query, and ``service.tools.handle_fetch_chunk``, for a single ID —
    could look up at most ``top_k`` keys in the result. Cost scaled with
    the corpus while the question scaled with ``top_k``.

    Returning ``None`` for an unknown ID rather than raising is what lets
    those callers keep their fail-closed branch: a chunk whose document
    row is gone must be refused, and refusing is their decision to make
    with their own error vocabulary, not this store's.

    Args:
        document_id: The document's unique identifier.

    Raises:
        StorageError: This store predates the ``source_class``/
            ``extractor`` columns (schema v3, ADR-0016), or a backend
            failure occurs.
    """
    self._require_source_class_capable("read a document record")

    def _op() -> DocumentRecord | None:
        cur = self._conn.execute(
            "SELECT source, source_class, extractor FROM documents WHERE document_id = ?",
            (document_id,),
        )
        row = cur.fetchone()
        if row is None:
            return None
        return DocumentRecord(source=str(row[0]), source_class=row[1], extractor=row[2])

    return await self._run(_op)

get_document_records async

get_document_records() -> dict[str, DocumentRecord]

Return a document_id -> DocumentRecord map for every stored document.

A sibling of :meth:get_document_sources, not a widening of it (ADR-0016). Every one of that method's five existing call sites reads only the source string or the key set, so changing its return type would touch five call sites for a field none of them need; adding this method instead confined the change to the two call sites that actually need the richer read.

Whole-table by construction. Since GK-019 the two callers that wanted a bounded set of keys use :meth:get_document_record instead, and this remains for callers that genuinely want every row: a diagnostic or export, and Retriever's fallback branch for a store that does not implement :class:~groundkit.index.protocols.DocumentRecordStoreProtocol at all.

Raises:

Type	Description
StorageError	This store predates the source_class/ extractor columns (schema v3, ADR-0016), or a backend failure occurs.

Source code in src/groundkit/index/metadata.py

async def get_document_records(self) -> dict[str, DocumentRecord]:
    """Return a ``document_id -> DocumentRecord`` map for every stored document.

    A sibling of :meth:`get_document_sources`, not a widening of it
    (ADR-0016). Every one of that method's five existing call sites reads
    only the source string or the key set, so changing its return type
    would touch five call sites for a field none of them need; adding
    this method instead confined the change to the two call sites that
    actually need the richer read.

    Whole-table by construction. Since GK-019 the two callers that wanted
    a bounded set of keys use :meth:`get_document_record` instead, and
    this remains for callers that genuinely want every row: a diagnostic
    or export, and ``Retriever``'s fallback branch for a store that does
    not implement :class:`~groundkit.index.protocols.DocumentRecordStoreProtocol`
    at all.

    Raises:
        StorageError: This store predates the ``source_class``/
            ``extractor`` columns (schema v3, ADR-0016), or a backend
            failure occurs.
    """
    self._require_source_class_capable("read document records")

    def _op() -> dict[str, DocumentRecord]:
        cur = self._conn.execute(
            "SELECT document_id, source, source_class, extractor FROM documents"
        )
        return {
            str(row[0]): DocumentRecord(
                source=str(row[1]), source_class=row[2], extractor=row[3]
            )
            for row in cur.fetchall()
        }

    return await self._run(_op)

get_document_sources async

get_document_sources() -> dict[str, str]

Return a document_id -> source map for every stored document.

Deliberately narrower than :meth:get_document_records: this query never selects source_class/extractor, so it keeps working unchanged against a v1, v2, or v3 store — none of this store's five pre-ADR-0016 callers (Indexer's two prune paths, the dense-side presence check, and two evals/runner.py reads) need the richer record, and widening this method's return type would force every one of them to change for a field they never use. See :meth:get_document_records's docstring for the fuller read.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_document_sources(self) -> dict[str, str]:
    """Return a ``document_id -> source`` map for every stored document.

    Deliberately narrower than :meth:`get_document_records`: this query
    never selects ``source_class``/``extractor``, so it keeps working
    unchanged against a v1, v2, *or* v3 store — none of this store's five
    pre-ADR-0016 callers (``Indexer``'s two prune paths, the dense-side
    presence check, and two ``evals/runner.py`` reads) need the richer
    record, and widening this method's return type would force every one
    of them to change for a field they never use. See
    :meth:`get_document_records`'s docstring for the fuller read.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> dict[str, str]:
        cur = self._conn.execute("SELECT document_id, source FROM documents")
        return {str(row[0]): str(row[1]) for row in cur.fetchall()}

    return await self._run(_op)

get_generation async

get_generation() -> int | None

Return the collection's staleness marker, or None if unanswerable (ADR-0013).

The marker advances on every commit that changes durable state, and the only operation defined on it is equality against a previously observed value. Nothing compares generations for ordering or distance; a caller that has one may ask "is this still the state I built against", and nothing else.

None means freshness cannot be asserted, never unchanged. It is returned for a store that is not schema-current — one predating ADR-0013, or predating ADR-0004's stamp entirely — and for a v2 store whose row is somehow absent. A caller must treat None as "assume it changed" and rebuild: answering a freshness question that cannot be answered is how a cache ends up serving an index it has no basis to vouch for. This deliberately degrades to reopen-per-request rather than refusing, because nothing is wrong on a legacy store — only uncacheable — and SPEC.md §2's fail-closed rule governs wrong answers, not slow ones.

Returns:

Type	Description
int | None	The current generation, or None when freshness is
int | None	unanswerable.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_generation(self) -> int | None:
    """Return the collection's staleness marker, or ``None`` if unanswerable (ADR-0013).

    The marker advances on every commit that changes durable state, and
    the only operation defined on it is **equality against a previously
    observed value**. Nothing compares generations for ordering or
    distance; a caller that has one may ask "is this still the state I
    built against", and nothing else.

    ``None`` means *freshness cannot be asserted*, never *unchanged*. It
    is returned for a store that is not schema-current — one predating
    ADR-0013, or predating ADR-0004's stamp entirely — and for a v2 store
    whose row is somehow absent. A caller must treat ``None`` as "assume
    it changed" and rebuild: answering a freshness question that cannot
    be answered is how a cache ends up serving an index it has no basis
    to vouch for. This deliberately degrades to reopen-per-request rather
    than refusing, because nothing is *wrong* on a legacy store — only
    uncacheable — and SPEC.md §2's fail-closed rule governs wrong
    answers, not slow ones.

    Returns:
        The current generation, or ``None`` when freshness is
        unanswerable.

    Raises:
        StorageError: On a backend failure.
    """
    if not self._schema_current:
        return None

    def _op() -> int | None:
        cur = self._conn.execute("SELECT generation FROM collection_state WHERE id = 1")
        row = cur.fetchone()
        return None if row is None else int(row[0])

    return await self._run(_op)

get_manifest async

get_manifest() -> CollectionManifest | None

Return the collection's embedding-identity manifest, or None if unset.

Unlike :meth:write_manifest and :meth:verify_manifest, this is a plain read and does not require the store to be manifest-capable: a legacy pre-ADR-0004 store never has a manifest row, so it reports None here rather than raising — useful for diagnostics (e.g. a future index_status tool) without needing to catch :class:~groundkit.errors.IndexIdentityError just to check.

Raises:

Type	Description
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def get_manifest(self) -> CollectionManifest | None:
    """Return the collection's embedding-identity manifest, or None if unset.

    Unlike :meth:`write_manifest` and :meth:`verify_manifest`, this is a
    plain read and does not require the store to be manifest-capable: a
    legacy pre-ADR-0004 store never has a manifest row, so it reports
    ``None`` here rather than raising — useful for diagnostics (e.g. a
    future ``index_status`` tool) without needing to catch
    :class:`~groundkit.errors.IndexIdentityError` just to check.

    Raises:
        StorageError: On a backend failure.
    """

    def _op() -> CollectionManifest | None:
        return self._select_manifest()

    return await self._run(_op)

open async classmethod

open(
    index_dir: Path, collection: str
) -> SQLiteMetadataStore

Open (creating if absent) the SQLite store for a collection.

Also determines whether this store predates the ADR-0004 collection manifest, by comparing its PRAGMA application_id/user_version against :data:APPLICATION_ID/:data:SCHEMA_VERSION. A freshly created file is stamped with both, inside the same transaction that applies the schema, as the last statements before commit. This never blocks opening the store — BM25-only collections must keep working unchanged — it only gates :meth:write_manifest and :meth:verify_manifest.

Parameters:

Name	Type	Description	Default
index_dir	Path	Directory holding the collection's persisted state. Created (with parents) if it does not exist.	required
collection	str	Collection name; backs the file <collection>.sqlite3. Validated against _COLLECTION_NAME_PATTERN and, once resolved, checked to stay contained within index_dir — an unchecked value could otherwise create or open a database anywhere on disk (e.g. collection="../outside" or an absolute path).	required

Returns:

Type	Description
SQLiteMetadataStore	A ready-to-use store with the schema applied.

Raises:

Type	Description
ConfigurationError	collection is empty or whitespace-only, contains a null byte or a path separator, is . or .., contains characters outside the allowed set, or otherwise resolves to a database path outside index_dir.
StorageError	The directory cannot be created, the database file cannot be opened, or the schema cannot be applied (e.g. a corrupted file or a path occupied by something other than a regular SQLite file).

Source code in src/groundkit/index/metadata.py

@classmethod
async def open(cls, index_dir: Path, collection: str) -> SQLiteMetadataStore:
    """Open (creating if absent) the SQLite store for a collection.

    Also determines whether this store predates the ADR-0004 collection
    manifest, by comparing its ``PRAGMA application_id``/``user_version``
    against :data:`APPLICATION_ID`/:data:`SCHEMA_VERSION`. A freshly
    created file is stamped with both, inside the same transaction that
    applies the schema, as the last statements before commit. This
    never blocks opening the store — BM25-only collections must keep
    working unchanged — it only gates :meth:`write_manifest` and
    :meth:`verify_manifest`.

    Args:
        index_dir: Directory holding the collection's persisted state.
            Created (with parents) if it does not exist.
        collection: Collection name; backs the file ``<collection>.sqlite3``.
            Validated against ``_COLLECTION_NAME_PATTERN`` and, once
            resolved, checked to stay contained within ``index_dir`` —
            an unchecked value could otherwise create or open a database
            anywhere on disk (e.g. ``collection="../outside"`` or an
            absolute path).

    Returns:
        A ready-to-use store with the schema applied.

    Raises:
        ConfigurationError: ``collection`` is empty or whitespace-only,
            contains a null byte or a path separator, is ``.`` or
            ``..``, contains characters outside the allowed set, or
            otherwise resolves to a database path outside ``index_dir``.
        StorageError: The directory cannot be created, the database file
            cannot be opened, or the schema cannot be applied (e.g. a
            corrupted file or a path occupied by something other than a
            regular SQLite file).
    """
    validate_collection_name(collection)
    db_path = index_dir / f"{collection}.sqlite3"

    def _connect() -> tuple[sqlite3.Connection, bool]:
        index_dir.mkdir(parents=True, exist_ok=True)
        _chmod_best_effort(index_dir, 0o700)
        try:
            ensure_within_base(db_path, index_dir)
        except ValueError as exc:
            raise ConfigurationError(
                f"collection {collection!r} resolves outside index_dir {index_dir}"
            ) from exc
        # Captured before connect() — sqlite3 creates the file on first
        # connection, so this is the only point at which "did this file
        # already exist" is still observable.
        is_new_file = not db_path.exists()
        # Refuse a stranger's database BEFORE the unconditional
        # `executescript(_SCHEMA)` below, which would otherwise create
        # groundkit's four tables inside it. The probe is read-only, so
        # the guard cannot itself perform the write it prevents.
        if not is_new_file and not is_groundkit_store(db_path):
            raise StorageError(
                f"{db_path} is not a groundkit collection: its "
                "PRAGMA application_id does not match, and it holds tables that are "
                "not groundkit's. Refusing to open it, because opening applies "
                "groundkit's schema and would add tables to a database that is not "
                "ours. Move or rename the file, or choose another index directory."
            )
        conn = sqlite3.connect(str(db_path), check_same_thread=False)
        try:
            # First, before any statement that can contend: journal_mode
            # and the schema application below are both writes, and a
            # busy_timeout set after them would not have covered them.
            # PRAGMA takes no bound parameters; the value is a module
            # constant, never externally supplied.
            conn.execute(f"PRAGMA busy_timeout = {BUSY_TIMEOUT_MS}")
            conn.execute("PRAGMA foreign_keys = ON")
            conn.execute("PRAGMA journal_mode = WAL")
            conn.executescript(_SCHEMA)
            if is_new_file:
                # Seed the ADR-0013 generation marker at 0, in the same
                # transaction as the schema and the stamps below. A v2
                # store therefore always has a row to read: a *missing*
                # row is then indistinguishable from a legacy store to
                # get_generation(), and both correctly answer "freshness
                # unanswerable" rather than a fabricated 0 that would let
                # a cache engage over an index it cannot vouch for.
                conn.execute(
                    "INSERT INTO collection_state (id, generation, updated_at) "
                    "VALUES (1, 0, ?)",
                    (_now_iso(),),
                )
                # Stamp groundkit's identity as the last statements
                # before commit (ADR-0004 decision 4): both
                # PRAGMA writes land in the same transaction as the
                # schema application above, so a failure anywhere in
                # this block (caught below) leaves nothing committed —
                # never a store claiming a version it does not have.
                # PRAGMA does not accept bound parameters; the
                # interpolated values are fixed module constants, never
                # externally supplied data.
                conn.execute(f"PRAGMA application_id = {APPLICATION_ID}")
                conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
            conn.commit()
        except Exception:
            # Schema application failed (e.g. a corrupted pre-existing
            # file) — close explicitly rather than leaving the handle to
            # CPython refcounting, which is not a guarantee.
            conn.close()
            raise
        _chmod_best_effort(db_path, 0o600)
        app_id = int(conn.execute("PRAGMA application_id").fetchone()[0])
        version = int(conn.execute("PRAGMA user_version").fetchone()[0])
        schema_current = app_id == APPLICATION_ID and version == SCHEMA_VERSION
        return conn, schema_current

    try:
        connection, schema_current = await asyncio.to_thread(_connect)
    except (OSError, sqlite3.Error) as exc:
        raise StorageError(f"failed to open metadata store at {db_path}") from exc

    logger.debug(
        "opened metadata store (collection=%s, schema_current=%s)",
        collection,
        schema_current,
    )
    return cls(connection, db_path, schema_current=schema_current)

replace_document async

replace_document(
    source: str,
    document_id: str,
    content_hash: str,
    chunks: list[Chunk],
    *,
    source_class: SourceClass = "text",
    extractor: str | None = None,
) -> None

Atomically replace a document's row and its chunks in one transaction.

Equivalent to calling :meth:upsert_document followed by :meth:add_chunks, but as a single commit. That matters because the two-call sequence has a durable-partial-write hazard: a crash (or any exception) between the two calls leaves the document row committed with its new content hash but zero chunks, and the incremental re-index skip (get_document_hash matching) then treats that document as already up to date forever. This method closes that gap by doing the delete-old-row, insert-document, and insert-chunks steps inside one :meth:_run call, so any failure — including a chunk with a mismatched document_id or non-JSON-serializable metadata — rolls back the whole thing and leaves the prior state (if any) untouched. Callers that need document and chunks written together (the indexer) should use this instead of the two-call sequence.

Parameters:

Name	Type	Description	Default
source	str	File path, URL, or other source identifier. Unique. If source already has a document row, that row and all of its chunks are deleted before the new ones are inserted.	required
document_id	str	Identifier to store the document under.	required
content_hash	str	Hash of the document's current content.	required
chunks	list[Chunk]	Chunks to persist for this document. May be empty.	required
source_class	SourceClass	How source maps to stored content (ADR-0016). Keyword-only, defaulting to "text" so every caller that predates this parameter keeps compiling unchanged. This is the field the ADR-0016 fail-open defect dropped silently: :class:~groundkit.indexer.Indexer now passes document.source_class through here on every write.	'text'
extractor	str | None	Extractor identity for an extracted source; None for every other class.	None

Raises:

Type	Description
StorageError	This store predates the source_class/ extractor columns (schema v3, ADR-0016); a chunk's document_id does not match document_id; a chunk has non-JSON-serializable metadata; or a backend failure occurs.

Source code in src/groundkit/index/metadata.py

async def replace_document(
    self,
    source: str,
    document_id: str,
    content_hash: str,
    chunks: list[Chunk],
    *,
    source_class: SourceClass = "text",
    extractor: str | None = None,
) -> None:
    """Atomically replace a document's row and its chunks in one transaction.

    Equivalent to calling :meth:`upsert_document` followed by
    :meth:`add_chunks`, but as a single commit. That matters because the
    two-call sequence has a durable-partial-write hazard: a crash (or any
    exception) between the two calls leaves the document row committed
    with its new content hash but zero chunks, and the incremental
    re-index skip (``get_document_hash`` matching) then treats that
    document as already up to date forever. This method closes that gap
    by doing the delete-old-row, insert-document, and insert-chunks steps
    inside one :meth:`_run` call, so any failure — including a chunk with
    a mismatched ``document_id`` or non-JSON-serializable metadata —
    rolls back the whole thing and leaves the prior state (if any)
    untouched. Callers that need document and chunks written together
    (the indexer) should use this instead of the two-call sequence.

    Args:
        source: File path, URL, or other source identifier. Unique. If
            ``source`` already has a document row, that row and all of
            its chunks are deleted before the new ones are inserted.
        document_id: Identifier to store the document under.
        content_hash: Hash of the document's current content.
        chunks: Chunks to persist for this document. May be empty.
        source_class: How ``source`` maps to stored content (ADR-0016).
            Keyword-only, defaulting to ``"text"`` so every caller that
            predates this parameter keeps compiling unchanged. This is
            the field the ADR-0016 fail-open defect dropped silently:
            :class:`~groundkit.indexer.Indexer` now passes
            ``document.source_class`` through here on every write.
        extractor: Extractor identity for an ``extracted`` source;
            ``None`` for every other class.

    Raises:
        StorageError: This store predates the ``source_class``/
            ``extractor`` columns (schema v3, ADR-0016); a chunk's
            ``document_id`` does not match ``document_id``; a chunk has
            non-JSON-serializable metadata; or a backend failure occurs.
    """
    self._require_source_class_capable("replace a document")
    ingested_at = _now_iso()

    def _op() -> None:
        self._delete_existing_source(source)
        self._conn.execute(
            "INSERT INTO documents "
            "(document_id, source, content_hash, ingested_at, source_class, extractor) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (document_id, source, content_hash, ingested_at, source_class, extractor),
        )
        for chunk in chunks:
            if chunk.document_id != document_id:
                raise StorageError(
                    f"chunk.document_id {chunk.document_id!r} does not match "
                    f"document_id {document_id!r} for source {source!r}"
                )
        self._insert_chunks(chunks)
        self._bump_generation()
        self._conn.commit()

    await self._run(_op)

upsert_document async

upsert_document(
    source: str,
    document_id: str,
    content_hash: str,
    *,
    source_class: SourceClass = "text",
    extractor: str | None = None,
) -> None

Record (or replace) a document's ingest state.

If source already has a document row (possibly under a different document_id — e.g. re-ingestion that regenerates IDs), that row and all of its chunks are explicitly deleted before the new row is inserted, so no stale chunks from a prior version of the document survive.

Parameters:

Name	Type	Description	Default
source	str	File path, URL, or other source identifier. Unique.	required
document_id	str	Identifier to store the document under.	required
content_hash	str	Hash of the document's current content.	required
source_class	SourceClass	How source maps to stored content (ADR-0016). Keyword-only, defaulting to "text" so every caller that predates this parameter keeps compiling unchanged.	'text'
extractor	str | None	Extractor identity for an extracted source; None for every other class.	None

Raises:

Type	Description
StorageError	This store predates the source_class/ extractor columns (schema v3, ADR-0016), or a backend failure occurs.

Source code in src/groundkit/index/metadata.py

async def upsert_document(
    self,
    source: str,
    document_id: str,
    content_hash: str,
    *,
    source_class: SourceClass = "text",
    extractor: str | None = None,
) -> None:
    """Record (or replace) a document's ingest state.

    If ``source`` already has a document row (possibly under a different
    ``document_id`` — e.g. re-ingestion that regenerates IDs), that row
    and all of its chunks are explicitly deleted before the new row is
    inserted, so no stale chunks from a prior version of the document
    survive.

    Args:
        source: File path, URL, or other source identifier. Unique.
        document_id: Identifier to store the document under.
        content_hash: Hash of the document's current content.
        source_class: How ``source`` maps to stored content (ADR-0016).
            Keyword-only, defaulting to ``"text"`` so every caller that
            predates this parameter keeps compiling unchanged.
        extractor: Extractor identity for an ``extracted`` source;
            ``None`` for every other class.

    Raises:
        StorageError: This store predates the ``source_class``/
            ``extractor`` columns (schema v3, ADR-0016), or a backend
            failure occurs.
    """
    self._require_source_class_capable("upsert a document")
    ingested_at = _now_iso()

    def _op() -> None:
        self._delete_existing_source(source)
        self._conn.execute(
            "INSERT INTO documents "
            "(document_id, source, content_hash, ingested_at, source_class, extractor) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (document_id, source, content_hash, ingested_at, source_class, extractor),
        )
        self._bump_generation()
        self._conn.commit()

    await self._run(_op)

verify_manifest async

verify_manifest(
    identity: EmbeddingIdentity,
) -> CollectionManifest | None

Verify identity matches the collection's stored identity manifest.

A collection with no manifest yet — no dense write has ever happened — has nothing to conflict with, so verification passes trivially; :meth:write_manifest is what establishes the manifest, not this method. Never a re-embed, never a fallback, never a warn-and-continue (SPEC.md §2): a real mismatch always raises.

Parameters:

Name	Type	Description	Default
identity	EmbeddingIdentity	The active embedding identity to check.	required

Returns:

Type	Description
CollectionManifest | None	The manifest this call verified identity against, or
CollectionManifest | None	None when the collection has none. The verdict and the
CollectionManifest | None	manifest come out of the same read, inside one :meth:_run,
CollectionManifest | None	which is what lets a caller decide "is this collection
CollectionManifest | None	dense-bound" without a second read that could observe a
CollectionManifest | None	different state — see Retriever.open for the concrete race
CollectionManifest | None	this closes.

Raises:

Type	Description
IndexIdentityError	This store predates the embedding-identity manifest (ADR-0004) and cannot be used for dense work, or the stored manifest's identity triple does not match identity.
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def verify_manifest(self, identity: EmbeddingIdentity) -> CollectionManifest | None:
    """Verify ``identity`` matches the collection's stored identity manifest.

    A collection with no manifest yet — no dense write has ever
    happened — has nothing to conflict with, so verification passes
    trivially; :meth:`write_manifest` is what establishes the manifest,
    not this method. Never a re-embed, never a fallback, never a
    warn-and-continue (SPEC.md §2): a real mismatch always raises.

    Args:
        identity: The active embedding identity to check.

    Returns:
        The manifest this call verified ``identity`` against, or
        ``None`` when the collection has none. The verdict and the
        manifest come out of the *same* read, inside one :meth:`_run`,
        which is what lets a caller decide "is this collection
        dense-bound" without a second read that could observe a
        different state — see ``Retriever.open`` for the concrete race
        this closes.

    Raises:
        IndexIdentityError: This store predates the embedding-identity
            manifest (ADR-0004) and cannot be used for dense work, or
            the stored manifest's identity triple does not match
            ``identity``.
        StorageError: On a backend failure.
    """
    self._require_manifest_capable("verify")

    def _op() -> CollectionManifest | None:
        existing = self._select_manifest()
        if existing is None or _manifest_matches(existing, identity):
            return existing
        raise IndexIdentityError(_identity_mismatch_message(existing, identity))

    return await self._run(_op)

write_manifest async

write_manifest(identity: EmbeddingIdentity) -> None

Write the collection's embedding-identity manifest, once (ADR-0004).

Called on the collection's first dense write. The manifest is immutable for the collection's lifetime thereafter: a later call with the same (provider, model_name, dimensions) triple as the stored manifest is a no-op — re-ingesting into an already-bound collection must keep working — but a call with a different triple is refused outright. Never a silent overwrite, never a re-embed.

Parameters:

Name	Type	Description	Default
identity	EmbeddingIdentity	The embedding identity establishing (or, on a later call, being checked against) this collection's identity.	required

Raises:

Type	Description
IndexIdentityError	This store predates the embedding-identity manifest (ADR-0004) and cannot be used for dense work, or a manifest already exists with a different identity triple.
StorageError	On a backend failure.

Source code in src/groundkit/index/metadata.py

async def write_manifest(self, identity: EmbeddingIdentity) -> None:
    """Write the collection's embedding-identity manifest, once (ADR-0004).

    Called on the collection's first dense write. The manifest is
    immutable for the collection's lifetime thereafter: a later call
    with the *same* ``(provider, model_name, dimensions)`` triple as the
    stored manifest is a no-op — re-ingesting into an already-bound
    collection must keep working — but a call with a *different*
    triple is refused outright. Never a silent overwrite, never a
    re-embed.

    Args:
        identity: The embedding identity establishing (or, on a
            later call, being checked against) this collection's
            identity.

    Raises:
        IndexIdentityError: This store predates the embedding-identity
            manifest (ADR-0004) and cannot be used for dense work, or a
            manifest already exists with a different identity triple.
        StorageError: On a backend failure.
    """
    self._require_manifest_capable("write")
    created_at = _now_iso()

    def _op() -> None:
        existing = self._select_manifest()
        if existing is None:
            self._conn.execute(
                "INSERT INTO collection_manifest "
                "(id, provider, model_name, dimensions, created_at) "
                "VALUES (1, ?, ?, ?, ?)",
                (identity.provider, identity.model_name, identity.dimensions, created_at),
            )
            self._bump_generation()
            self._conn.commit()
            return
        if _manifest_matches(existing, identity):
            return  # same identity: re-ingesting a bound collection is a no-op
        raise IndexIdentityError(_identity_mismatch_message(existing, identity))

    await self._run(_op)

is_groundkit_store

is_groundkit_store(db_path: Path) -> bool

True when db_path is a groundkit collection, checked without writing.

Exists because :meth:SQLiteMetadataStore.open applies _SCHEMA unconditionally: pointed at an unrelated SQLite file it would create documents, chunks, collection_manifest and collection_state inside it. That made two read-only operations mutate a stranger's database — a *.sqlite3 a user happened to leave in the index directory would be advertised by list_collections and then written to by index_status.

The check opens the file with mode=ro&immutable=1, and both flags are load-bearing. mode=ro alone is not write-free against a WAL database: SQLite needs the -shm shared-memory file to read one, and creates -shm/-wal beside the store if they are absent — which is a write, and exactly the write this function exists to avoid. tests/test_service_tools.py::test_list_collections_opens_nothing catches it. immutable=1 promises the file cannot change, so SQLite skips the WAL machinery entirely and touches nothing.

The cost of immutable=1 is that any uncheckpointed WAL content is invisible. That does not matter here: application_id lives in the database header, and the sqlite_master read below is only reached on the unstamped path, where "no tables visible" and "tables visible" both answer True.

Recognized as groundkit:

• PRAGMA application_id equal to :data:APPLICATION_ID — the normal case, whatever user_version says. Keying on the stamp alone is deliberate: a pre-v3 store is still ours and must stay readable (ADR-0016 refuses its writes, not its reads), so folding user_version into this test would lock out the very stores that guard was written to keep working.
• application_id of 0 with both :data:_LEGACY_MARKER_TABLES present — a groundkit store predating the stamp.
• application_id of 0 with no tables at all — an empty file, which :meth:open may legitimately adopt.

Anything else — including an unreadable or corrupt file — is not ours. Returning False rather than raising keeps the caller's choice open: listing skips it, opening refuses it.

Parameters:

Name	Type	Description	Default
db_path	Path	The candidate file. Must already exist; a missing path is not this function's question and returns False.	required

Returns:

Type	Description
bool	Whether the file may be treated as a groundkit collection.

Source code in src/groundkit/index/metadata.py

def is_groundkit_store(db_path: Path) -> bool:
    """True when ``db_path`` is a groundkit collection, checked without writing.

    Exists because :meth:`SQLiteMetadataStore.open` applies ``_SCHEMA``
    unconditionally: pointed at an unrelated SQLite file it would *create*
    `documents`, `chunks`, `collection_manifest` and `collection_state` inside
    it. That made two read-only operations mutate a stranger's database — a
    `*.sqlite3` a user happened to leave in the index directory would be
    advertised by `list_collections` and then written to by `index_status`.

    The check opens the file with ``mode=ro&immutable=1``, and **both flags are
    load-bearing.** ``mode=ro`` alone is not write-free against a WAL database:
    SQLite needs the ``-shm`` shared-memory file to read one, and creates
    ``-shm``/``-wal`` beside the store if they are absent — which is a write,
    and exactly the write this function exists to avoid.
    ``tests/test_service_tools.py::test_list_collections_opens_nothing``
    catches it. ``immutable=1`` promises the file cannot change, so SQLite
    skips the WAL machinery entirely and touches nothing.

    The cost of ``immutable=1`` is that any uncheckpointed WAL content is
    invisible. That does not matter here: ``application_id`` lives in the
    database header, and the ``sqlite_master`` read below is only reached on
    the unstamped path, where "no tables visible" and "tables visible" both
    answer ``True``.

    Recognized as groundkit:

    - ``PRAGMA application_id`` equal to :data:`APPLICATION_ID` — the normal
      case, whatever ``user_version`` says. Keying on the stamp alone is
      deliberate: a pre-v3 store is still *ours* and must stay readable
      (ADR-0016 refuses its writes, not its reads), so folding
      ``user_version`` into this test would lock out the very stores that
      guard was written to keep working.
    - ``application_id`` of ``0`` with both :data:`_LEGACY_MARKER_TABLES`
      present — a groundkit store predating the stamp.
    - ``application_id`` of ``0`` with no tables at all — an empty file, which
      :meth:`open` may legitimately adopt.

    Anything else — including an unreadable or corrupt file — is not ours.
    Returning ``False`` rather than raising keeps the caller's choice open:
    listing skips it, opening refuses it.

    Args:
        db_path: The candidate file. Must already exist; a missing path is
            not this function's question and returns ``False``.

    Returns:
        Whether the file may be treated as a groundkit collection.
    """
    if not db_path.is_file():
        return False
    try:
        conn = sqlite3.connect(f"{db_path.as_uri()}?mode=ro&immutable=1", uri=True)
    except sqlite3.Error:
        return False
    try:
        app_id = int(conn.execute("PRAGMA application_id").fetchone()[0])
        if app_id == APPLICATION_ID:
            return True
        if app_id != 0:
            return False
        tables = {
            str(row[0])
            for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")
        }
    except sqlite3.Error:
        return False
    finally:
        conn.close()
    return not tables or _LEGACY_MARKER_TABLES.issubset(tables)

validate_collection_name

validate_collection_name(collection: str) -> None

Validate that collection is safe to interpolate into a file path.

This is the first of two containment layers for :meth:SQLiteMetadataStore.open (the second is the resolved-path check via :func:~groundkit.utils.path_safety.ensure_within_base): it states the contract on the name itself, up front, so a bad value fails with a specific message rather than relying solely on the later path check.

Parameters:

Name	Type	Description	Default
collection	str	Candidate collection name.	required

Raises:

Type	Description
ConfigurationError	collection is empty or whitespace-only, contains a null byte, is . or .., or contains characters outside _COLLECTION_NAME_PATTERN (which itself excludes path separators, drive/UNC prefixes, and absolute paths — all of those require a character the pattern disallows).

Source code in src/groundkit/index/metadata.py

def validate_collection_name(collection: str) -> None:
    """Validate that ``collection`` is safe to interpolate into a file path.

    This is the first of two containment layers for :meth:`SQLiteMetadataStore.open`
    (the second is the resolved-path check via
    :func:`~groundkit.utils.path_safety.ensure_within_base`): it states the
    contract on the name itself, up front, so a bad value fails with a
    specific message rather than relying solely on the later path check.

    Args:
        collection: Candidate collection name.

    Raises:
        ConfigurationError: ``collection`` is empty or whitespace-only,
            contains a null byte, is ``.`` or ``..``, or contains characters
            outside ``_COLLECTION_NAME_PATTERN`` (which itself excludes path
            separators, drive/UNC prefixes, and absolute paths — all of
            those require a character the pattern disallows).
    """
    if "\0" in collection:
        raise ConfigurationError("collection name must not contain a null byte")
    if not collection.strip():
        raise ConfigurationError("collection name must not be empty or whitespace-only")
    if collection in (".", ".."):
        raise ConfigurationError(f"collection name must not be {collection!r}")
    if not _COLLECTION_NAME_PATTERN.fullmatch(collection):
        raise ConfigurationError(
            f"collection name {collection!r} contains characters outside the allowed "
            "set (letters, digits, '-', '_', '.')"
        )