Skip to content

Stores

varagity.stores.records

Chunk-level data model — the canonical metadata record (spec §8.1).

Every chunk carries a complete, validated metadata record. Chunks live in both stores (pgvector and Elasticsearch) and are joinable by the shared identity (doc_id, original_index).

Identity derivation (plan decision #6): doc_id hashes the path relative to DOCS_PATH — not the absolute path spec §8.1 sketched — because absolute paths differ between host and container and across machines, which would break idempotency and make golden eval sets non-portable. The absolute path is still recorded in source.

ChunkRecord

Bases: BaseModel

One chunk's full metadata record (every spec §8.1 field).

Attributes:

Name Type Description
doc_id str

Stable id per source document (see :func:derive_doc_id).
chunk_id str

f"{doc_id}::{chunk_index}" — the pgvector primary key.
original_index int

Global monotonic chunk index across the corpus (the hybrid-fusion identity key, spec §8).
chunk_index int

Chunk position within its document.
source str

Absolute file path (host- or container-local; provenance only).
file_name str

Basename of the source file.
file_type str

pdf / txt / md.
page int | None

Page number (PDF; None otherwise).
content str

Original chunk text.
context str | None

LLM-generated situating blurb (None when CONTEXTUALIZE is off — the non-contextual baseline).
contextualized_content str

The text actually embedded and BM25-indexed; identical to content while context is None.
chunk_size int

Chunk size parameter used, in characters (provenance).
chunk_overlap int

Chunk overlap parameter used, in characters.
chunking_strategy str

Registry name of the chunker used.
embedding_model str

Served embedding model name.
n_tokens int

Approximate token count of content (plan decision #8).
content_hash str

The parent document's :func:content_hash.
created_at datetime

Ingestion timestamp (UTC) — when the chunk was made, not a property of the source file (see file_modified_at).
file_created_at datetime | None

Filesystem birth time of the source file (UTC). Best-effort: birth time is filesystem/platform-dependent, and a copy or download resets it — None when unavailable.
file_modified_at datetime | None

Filesystem mtime of the source file (UTC) — when the document's bytes last changed, independent of when they were ingested. None only on rows written before the field existed (or if the file vanished mid-ingest).
extraction str

How text was extracted: "text" (default), "ocr_fallback" (set by the PDF OCR fallback pass), or "ocr" (image parser — OCR is that format's only path) — retrieval-quality provenance beyond spec §8.1.
heading_path str | None

The chunk's markdown heading breadcrumb (e.g. "Operations > Dredging"), set by the heading-aware chunkers (spec_v2 §7); None for strategies without structure.
Source code in varagity/stores/records.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class ChunkRecord(BaseModel):
    """One chunk's full metadata record (every spec §8.1 field).

    Attributes:
        doc_id: Stable id per source document (see :func:`derive_doc_id`).
        chunk_id: ``f"{doc_id}::{chunk_index}"`` — the pgvector primary key.
        original_index: Global monotonic chunk index across the corpus (the
            hybrid-fusion identity key, spec §8).
        chunk_index: Chunk position within its document.
        source: Absolute file path (host- or container-local; provenance only).
        file_name: Basename of the source file.
        file_type: ``pdf`` / ``txt`` / ``md``.
        page: Page number (PDF; ``None`` otherwise).
        content: Original chunk text.
        context: LLM-generated situating blurb (``None`` when
            ``CONTEXTUALIZE`` is off — the non-contextual baseline).
        contextualized_content: The text actually embedded and BM25-indexed;
            identical to ``content`` while ``context`` is ``None``.
        chunk_size: Chunk size parameter used, in characters (provenance).
        chunk_overlap: Chunk overlap parameter used, in characters.
        chunking_strategy: Registry name of the chunker used.
        embedding_model: Served embedding model name.
        n_tokens: Approximate token count of ``content`` (plan decision #8).
        content_hash: The parent document's :func:`content_hash`.
        created_at: Ingestion timestamp (UTC) — when the chunk was made,
            not a property of the source file (see ``file_modified_at``).
        file_created_at: Filesystem birth time of the source file (UTC).
            Best-effort: birth time is filesystem/platform-dependent, and a
            copy or download resets it — ``None`` when unavailable.
        file_modified_at: Filesystem mtime of the source file (UTC) — when
            the document's bytes last changed, independent of when they
            were ingested. ``None`` only on rows written before the field
            existed (or if the file vanished mid-ingest).
        extraction: How text was extracted: ``"text"`` (default),
            ``"ocr_fallback"`` (set by the PDF OCR fallback pass), or
            ``"ocr"`` (image parser — OCR is that format's only path) —
            retrieval-quality provenance beyond spec §8.1.
        heading_path: The chunk's markdown heading breadcrumb (e.g.
            ``"Operations > Dredging"``), set by the heading-aware chunkers
            (spec_v2 §7); ``None`` for strategies without structure.
    """

    doc_id: str
    chunk_id: str
    original_index: int
    chunk_index: int
    source: str
    file_name: str
    file_type: str
    page: int | None = None
    content: str
    context: str | None = None
    contextualized_content: str
    chunk_size: int
    chunk_overlap: int
    chunking_strategy: str
    embedding_model: str
    n_tokens: int
    content_hash: str
    created_at: datetime
    file_created_at: datetime | None = None
    file_modified_at: datetime | None = None
    extraction: str = "text"
    heading_path: str | None = None

    @classmethod
    def create(
        cls,
        *,
        doc_id: str,
        original_index: int,
        chunk_index: int,
        source: str,
        file_name: str,
        file_type: str,
        page: int | None,
        content: str,
        context: str | None,
        chunk_size: int,
        chunk_overlap: int,
        chunking_strategy: str,
        embedding_model: str,
        content_hash: str,
        file_created_at: datetime | None = None,
        file_modified_at: datetime | None = None,
        extraction: str = "text",
        heading_path: str | None = None,
    ) -> "ChunkRecord":
        r"""Build a record, deriving the dependent fields.

        Derives ``chunk_id``, ``n_tokens``, ``created_at``, and the
        ``contextualized_content`` composition rule (spec §9.4): with a
        ``context`` blurb it is ``context + "\n\n" + content``; without one
        (``CONTEXTUALIZE`` off, plan decision #2) it is ``content``
        unchanged.

        Args:
            doc_id: Stable id of the parent document.
            original_index: Global monotonic chunk index across the corpus.
            chunk_index: Chunk position within its document.
            source: Absolute file path.
            file_name: Basename of the source file.
            file_type: ``pdf`` / ``txt`` / ``md``.
            page: Page number (PDF; ``None`` otherwise).
            content: Original chunk text.
            context: LLM situating blurb, or ``None`` (``CONTEXTUALIZE`` off).
            chunk_size: Chunk size parameter used (characters).
            chunk_overlap: Chunk overlap parameter used (characters).
            chunking_strategy: Registry name of the chunker used.
            embedding_model: Served embedding model name.
            content_hash: The parent document's content hash.
            file_created_at: Filesystem birth time of the source file, or
                ``None`` when the platform/filesystem doesn't expose one.
            file_modified_at: Filesystem mtime of the source file, or
                ``None`` when it couldn't be read.
            extraction: Extraction provenance (``"text"``, ``"ocr"``, or
                ``"ocr_fallback"``).
            heading_path: Markdown heading breadcrumb, or ``None`` for
                strategies without structure.

        Returns:
            The fully-populated record.
        """
        contextualized = f"{context}\n\n{content}" if context else content
        return cls(
            doc_id=doc_id,
            chunk_id=f"{doc_id}::{chunk_index}",
            original_index=original_index,
            chunk_index=chunk_index,
            source=source,
            file_name=file_name,
            file_type=file_type,
            page=page,
            content=content,
            context=context,
            contextualized_content=contextualized,
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            chunking_strategy=chunking_strategy,
            embedding_model=embedding_model,
            n_tokens=count_tokens(content),
            content_hash=content_hash,
            created_at=datetime.now(UTC),
            file_created_at=file_created_at,
            file_modified_at=file_modified_at,
            extraction=extraction,
            heading_path=heading_path,
        )

create(*, doc_id, original_index, chunk_index, source, file_name, file_type, page, content, context, chunk_size, chunk_overlap, chunking_strategy, embedding_model, content_hash, file_created_at=None, file_modified_at=None, extraction='text', heading_path=None) classmethod

Build a record, deriving the dependent fields.

Derives chunk_id, n_tokens, created_at, and the contextualized_content composition rule (spec §9.4): with a context blurb it is context + "\n\n" + content; without one (CONTEXTUALIZE off, plan decision #2) it is content unchanged.

Parameters:

Name Type Description Default
doc_id str

Stable id of the parent document.

 required
original_index int

Global monotonic chunk index across the corpus.

 required
chunk_index int

Chunk position within its document.

 required
source str

Absolute file path.

 required
file_name str

Basename of the source file.

 required
file_type str

pdf / txt / md.

 required
page int | None

Page number (PDF; None otherwise).

 required
content str

Original chunk text.

 required
context str | None

LLM situating blurb, or None (CONTEXTUALIZE off).

 required
chunk_size int

Chunk size parameter used (characters).

 required
chunk_overlap int

Chunk overlap parameter used (characters).

 required
chunking_strategy str

Registry name of the chunker used.

 required
embedding_model str

Served embedding model name.

 required
content_hash str

The parent document's content hash.

 required
file_created_at datetime | None

Filesystem birth time of the source file, or None when the platform/filesystem doesn't expose one.

 None
file_modified_at datetime | None

Filesystem mtime of the source file, or None when it couldn't be read.

 None
extraction str

Extraction provenance ("text", "ocr", or "ocr_fallback").

 'text'
heading_path str | None

Markdown heading breadcrumb, or None for strategies without structure.

 None

Returns:

Type Description
ChunkRecord

The fully-populated record.
Source code in varagity/stores/records.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@classmethod
def create(
    cls,
    *,
    doc_id: str,
    original_index: int,
    chunk_index: int,
    source: str,
    file_name: str,
    file_type: str,
    page: int | None,
    content: str,
    context: str | None,
    chunk_size: int,
    chunk_overlap: int,
    chunking_strategy: str,
    embedding_model: str,
    content_hash: str,
    file_created_at: datetime | None = None,
    file_modified_at: datetime | None = None,
    extraction: str = "text",
    heading_path: str | None = None,
) -> "ChunkRecord":
    r"""Build a record, deriving the dependent fields.

    Derives ``chunk_id``, ``n_tokens``, ``created_at``, and the
    ``contextualized_content`` composition rule (spec §9.4): with a
    ``context`` blurb it is ``context + "\n\n" + content``; without one
    (``CONTEXTUALIZE`` off, plan decision #2) it is ``content``
    unchanged.

    Args:
        doc_id: Stable id of the parent document.
        original_index: Global monotonic chunk index across the corpus.
        chunk_index: Chunk position within its document.
        source: Absolute file path.
        file_name: Basename of the source file.
        file_type: ``pdf`` / ``txt`` / ``md``.
        page: Page number (PDF; ``None`` otherwise).
        content: Original chunk text.
        context: LLM situating blurb, or ``None`` (``CONTEXTUALIZE`` off).
        chunk_size: Chunk size parameter used (characters).
        chunk_overlap: Chunk overlap parameter used (characters).
        chunking_strategy: Registry name of the chunker used.
        embedding_model: Served embedding model name.
        content_hash: The parent document's content hash.
        file_created_at: Filesystem birth time of the source file, or
            ``None`` when the platform/filesystem doesn't expose one.
        file_modified_at: Filesystem mtime of the source file, or
            ``None`` when it couldn't be read.
        extraction: Extraction provenance (``"text"``, ``"ocr"``, or
            ``"ocr_fallback"``).
        heading_path: Markdown heading breadcrumb, or ``None`` for
            strategies without structure.

    Returns:
        The fully-populated record.
    """
    contextualized = f"{context}\n\n{content}" if context else content
    return cls(
        doc_id=doc_id,
        chunk_id=f"{doc_id}::{chunk_index}",
        original_index=original_index,
        chunk_index=chunk_index,
        source=source,
        file_name=file_name,
        file_type=file_type,
        page=page,
        content=content,
        context=context,
        contextualized_content=contextualized,
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        chunking_strategy=chunking_strategy,
        embedding_model=embedding_model,
        n_tokens=count_tokens(content),
        content_hash=content_hash,
        created_at=datetime.now(UTC),
        file_created_at=file_created_at,
        file_modified_at=file_modified_at,
        extraction=extraction,
        heading_path=heading_path,
    )

DocumentInfo

Bases: BaseModel

One ingested document, as listed by GET /api/documents (spec_v2 §4.2).

A documents-table row joined with its chunks' extraction mix — the corpus-management view (file, type, chunk count, ingested-at, how much of it came through OCR).

Attributes:

Name Type Description
doc_id str

The document's stable id.
source str

Absolute file path recorded at ingest time.
file_type str

File extension without the dot (pdf, docx, …).
content_hash str

sha256 of the source file's bytes at ingest time — the preview path re-hashes the on-disk file against it, so an edited-but-not-reingested document degrades honestly (file_changed) instead of previewing the wrong bytes.
n_chunks int

Chunks ingested (0 = no extractable text).
ingested_at datetime

When the document (last) landed in the stores.
extraction_mix dict[str, int]

Chunk count per extraction method (text / ocr / ocr_fallback); empty for a 0-chunk document.
Source code in varagity/stores/records.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
class DocumentInfo(BaseModel):
    """One ingested document, as listed by ``GET /api/documents`` (spec_v2 §4.2).

    A ``documents``-table row joined with its chunks' extraction mix — the
    corpus-management view (file, type, chunk count, ingested-at, how much
    of it came through OCR).

    Attributes:
        doc_id: The document's stable id.
        source: Absolute file path recorded at ingest time.
        file_type: File extension without the dot (``pdf``, ``docx``, …).
        content_hash: sha256 of the source file's bytes at ingest time —
            the preview path re-hashes the on-disk file against it, so an
            edited-but-not-reingested document degrades honestly
            (``file_changed``) instead of previewing the wrong bytes.
        n_chunks: Chunks ingested (``0`` = no extractable text).
        ingested_at: When the document (last) landed in the stores.
        extraction_mix: Chunk count per extraction method (``text`` /
            ``ocr`` / ``ocr_fallback``); empty for a 0-chunk document.
    """

    doc_id: str
    source: str
    file_type: str
    content_hash: str
    n_chunks: int
    ingested_at: datetime
    extraction_mix: dict[str, int]

GraphSourceSnapshot

Bases: BaseModel

One cited transcript day of a graph-targeted turn (spec_graphrag §4.2).

The graph twin of :class:RetrievedChunk for persistence only: a graph answer's evidence is document-grain (ADR-017's priced regret — the engines cannot honestly attribute a single message), so what gets snapshotted into message_sources is the transcript day the answer cited, keyed by its doc_key.

A plain record rather than :class:varagity.graph.records.TranscriptExcerpt on purpose: the store layer must not import :mod:varagity.graph (the same rule that keeps recent_turns returning primitive pairs instead of chat turns), and the snapshot is a frozen shape — it has to keep explaining a conversation after the graph it came from was rebuilt.

Attributes:

Name Type Description
doc_key str

The transcript document key, a soft reference exactly like chunk_id (no FK: a rebuild re-derives keys).
thread_name str

Human-facing thread label.
span str

The document's day span (YYYY-MM-DD or first..last).
excerpt str

The retrieved passage, capped for storage by the caller.
message_guids list[str]

Per-message provenance for the day, or [] when the graph's manifest could not resolve the key.
Source code in varagity/stores/records.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class GraphSourceSnapshot(BaseModel):
    """One cited transcript day of a graph-targeted turn (spec_graphrag §4.2).

    The graph twin of :class:`RetrievedChunk` *for persistence only*: a graph
    answer's evidence is document-grain (ADR-017's priced regret — the
    engines cannot honestly attribute a single message), so what gets
    snapshotted into ``message_sources`` is the transcript day the answer
    cited, keyed by its ``doc_key``.

    A plain record rather than
    :class:`varagity.graph.records.TranscriptExcerpt` on purpose: the store
    layer must not import :mod:`varagity.graph` (the same rule that keeps
    ``recent_turns`` returning primitive pairs instead of chat turns), and
    the snapshot is a *frozen* shape — it has to keep explaining a
    conversation after the graph it came from was rebuilt.

    Attributes:
        doc_key: The transcript document key, a soft reference exactly like
            ``chunk_id`` (no FK: a rebuild re-derives keys).
        thread_name: Human-facing thread label.
        span: The document's day span (``YYYY-MM-DD`` or ``first..last``).
        excerpt: The retrieved passage, capped for storage by the caller.
        message_guids: Per-message provenance for the day, or ``[]`` when the
            graph's manifest could not resolve the key.
    """

    doc_key: str
    thread_name: str
    span: str
    excerpt: str
    message_guids: list[str] = []

RetrievalTrace

Bases: BaseModel

Why one retrieved chunk ranked where it did (spec_v2 §9.2).

Query-time only — computed per answer by the retrievers, never persisted on the immutable chunk rows. Ranks are 1-based (display-ready); an arm's rank/score are None when that arm's ranked list never surfaced the chunk. Single-arm retrievers report their arm's score/rank as the fused values (there is nothing to fuse).

Attributes:

Name Type Description
semantic_rank int | None

Rank in the semantic (pgvector) arm.
semantic_score float | None

Cosine similarity in the semantic arm.
bm25_rank int | None

Rank in the BM25 (Elasticsearch) arm.
bm25_score float | None

BM25 relevance in that arm.
fused_score float

Weighted reciprocal-rank fusion score (spec §11.4).
fused_rank int

Rank after fusion.
rerank_score float | None

Cross-encoder relevance (None when reranking is off the path).
rerank_delta int | None

Positions moved by reranking, pre − post (+ moved up / − moved down; None when reranking is off the path).
final_rank int

The rank actually returned to the caller.
Source code in varagity/stores/records.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class RetrievalTrace(BaseModel):
    """Why one retrieved chunk ranked where it did (spec_v2 §9.2).

    Query-time only — computed per answer by the retrievers, never persisted
    on the immutable chunk rows. Ranks are 1-based (display-ready); an arm's
    rank/score are ``None`` when that arm's ranked list never surfaced the
    chunk. Single-arm retrievers report their arm's score/rank as the fused
    values (there is nothing to fuse).

    Attributes:
        semantic_rank: Rank in the semantic (pgvector) arm.
        semantic_score: Cosine similarity in the semantic arm.
        bm25_rank: Rank in the BM25 (Elasticsearch) arm.
        bm25_score: BM25 relevance in that arm.
        fused_score: Weighted reciprocal-rank fusion score (spec §11.4).
        fused_rank: Rank after fusion.
        rerank_score: Cross-encoder relevance (``None`` when reranking is
            off the path).
        rerank_delta: Positions moved by reranking, ``pre − post`` (+ moved
            up / − moved down; ``None`` when reranking is off the path).
        final_rank: The rank actually returned to the caller.
    """

    semantic_rank: int | None = None
    semantic_score: float | None = None
    bm25_rank: int | None = None
    bm25_score: float | None = None
    fused_score: float
    fused_rank: int
    rerank_score: float | None = None
    rerank_delta: int | None = None
    final_rank: int

RetrievedChunk

Bases: BaseModel

A chunk returned by a store search, with its relevance score.

Field set mirrors the spec §11.2 SELECT list; metadata holds the full :class:ChunkRecord dump persisted at ingest time.

Attributes:

Name Type Description
chunk_id str

The chunk's primary key.
doc_id str

Parent document id.
original_index int

Global chunk index (fusion identity).
content str

Original chunk text.
context str | None

LLM situating blurb (None when ingested with CONTEXTUALIZE off).
metadata dict[str, Any]

Full persisted metadata record.
score float

Similarity score — cosine similarity 1 - distance for the vector store (higher is better).
trace RetrievalTrace | None

Rank provenance attached by the retrievers (spec_v2 §9.2); None on raw store results, so pre-trace callers are unaffected.
Source code in varagity/stores/records.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
class RetrievedChunk(BaseModel):
    """A chunk returned by a store search, with its relevance score.

    Field set mirrors the spec §11.2 ``SELECT`` list; ``metadata`` holds the
    full :class:`ChunkRecord` dump persisted at ingest time.

    Attributes:
        chunk_id: The chunk's primary key.
        doc_id: Parent document id.
        original_index: Global chunk index (fusion identity).
        content: Original chunk text.
        context: LLM situating blurb (``None`` when ingested with
            ``CONTEXTUALIZE`` off).
        metadata: Full persisted metadata record.
        score: Similarity score — cosine similarity ``1 - distance`` for the
            vector store (higher is better).
        trace: Rank provenance attached by the retrievers (spec_v2 §9.2);
            ``None`` on raw store results, so pre-trace callers are
            unaffected.
    """

    chunk_id: str
    doc_id: str
    original_index: int
    content: str
    context: str | None
    metadata: dict[str, Any]
    score: float
    trace: RetrievalTrace | None = None

content_hash(data)

Hash raw file bytes for idempotency / dedup (spec §8.2).

Hashing the bytes (rather than parsed text) lets re-runs skip unchanged files before paying the parse cost.

Parameters:

Name Type Description Default
data bytes

The file's raw bytes.

 required

Returns:

Type Description
str

The sha256 hex digest.
Source code in varagity/stores/records.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def content_hash(data: bytes) -> str:
    """Hash raw file bytes for idempotency / dedup (spec §8.2).

    Hashing the bytes (rather than parsed text) lets re-runs skip unchanged
    files *before* paying the parse cost.

    Args:
        data: The file's raw bytes.

    Returns:
        The sha256 hex digest.
    """
    return hashlib.sha256(data).hexdigest()

derive_doc_id(relative_path, file_hash)

Derive the stable per-document id (plan decision #6).

Parameters:

Name Type Description Default
relative_path str

POSIX-style path of the document relative to DOCS_PATH (portable across host/container/machines).

 required
file_hash str

The document's :func:content_hash.

 required

Returns:

Type Description
str

sha256(relative_path + ":" + file_hash) truncated to 16 hex chars.
Source code in varagity/stores/records.py 
38
39
40
41
42
43
44
45
46
47
48
49
def derive_doc_id(relative_path: str, file_hash: str) -> str:
    """Derive the stable per-document id (plan decision #6).

    Args:
        relative_path: POSIX-style path of the document **relative to**
            ``DOCS_PATH`` (portable across host/container/machines).
        file_hash: The document's :func:`content_hash`.

    Returns:
        ``sha256(relative_path + ":" + file_hash)`` truncated to 16 hex chars.
    """
    return hashlib.sha256(f"{relative_path}:{file_hash}".encode()).hexdigest()[:16]

varagity.stores.vector_store

ContextualVectorDB — the pgvector-backed dense vector store.

The Anthropic cookbook's in-memory ContextualVectorDB responsibilities (store vectors + metadata, cosine search), re-backed with PostgreSQL/pgvector (spec §11.2) so the index is durable, concurrent, and inspectable via SQL. The schema itself lives in schema.sql (applied by the postgres container on first boot).

It also owns the RAG corpus's advisory lock (ADR-019): one fixed key in the database's advisory-lock namespace that every ingest holds shared for its whole run and a backup restore holds exclusive from before it empties the tables until the BM25 index is rebuilt. The API's in-process corpus gate cannot see a CLI ingest (main.py ingest / chat run in their own process); this lock can, because it lives in the one place every writer connects to. Ingests still overlap each other exactly as before, and a session-level lock dies with its connection, so a crashed process never strands the corpus.

ContextualVectorDB

Bases: ClosingContextMixin

Dense vector store over PostgreSQL + pgvector.

Owns one connection (autocommit; per-document writes group into an explicit transaction via :meth:store_document). Use as a context manager or call :meth:close when done.

Source code in varagity/stores/vector_store.py 
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
class ContextualVectorDB(ClosingContextMixin):
    """Dense vector store over PostgreSQL + pgvector.

    Owns one connection (autocommit; per-document writes group into an
    explicit transaction via :meth:`store_document`). Use as a context
    manager or call :meth:`close` when done.
    """

    def __init__(self, conninfo: str | None = None) -> None:
        """Connect and register the pgvector type adapters.

        Args:
            conninfo: libpq connection string; defaults to the
                ``POSTGRES_*`` settings.

        Raises:
            psycopg.OperationalError: If the database is unreachable.
        """
        self._conn = psycopg.connect(conninfo or default_conninfo(), autocommit=True)
        register_vector(self._conn)

    def close(self) -> None:
        """Close the underlying connection (idempotent)."""
        if not self._conn.closed:
            self._conn.close()

    def document_exists(self, doc_id: str, content_hash: str) -> bool:
        """Check whether a document was already ingested (idempotency, spec §8.2).

        Args:
            doc_id: The document's stable id.
            content_hash: The document's content hash (``doc_id`` already
                encodes it — checking both is belt-and-braces).

        Returns:
            ``True`` if a matching ``documents`` row exists.
        """
        row = self._conn.execute(
            "SELECT 1 FROM documents WHERE doc_id = %s AND content_hash = %s",
            (doc_id, content_hash),
        ).fetchone()
        return row is not None

    def document_n_chunks(self, doc_id: str) -> int | None:
        """Fetch a document's recorded chunk count.

        Lets the loader re-warn about known-empty documents (``n_chunks = 0``)
        without re-parsing them.

        Args:
            doc_id: The document's stable id.

        Returns:
            The ``n_chunks`` value, or ``None`` if the document is unknown.
        """
        row = self._conn.execute(
            "SELECT n_chunks FROM documents WHERE doc_id = %s", (doc_id,)
        ).fetchone()
        return None if row is None else int(row[0])

    def document_count(self) -> int:
        """Count the ingested documents.

        Backs the settings route's corpus-stale check (spec_v2 §4.7): an
        empty corpus has nothing to go stale, so a reingest-affecting
        override change only raises the flag when this is positive.

        Returns:
            Number of ``documents`` rows.
        """
        row = self._conn.execute("SELECT count(*) FROM documents").fetchone()
        return 0 if row is None else int(row[0])

    def chunk_count(self) -> int:
        """Count the stored chunks.

        Backs the ``varagity_corpus_chunks`` gauge (spec_v3 §6.1a): the
        store is the truthful source for "how big is my corpus", unlike the
        per-process ingest counters that reset with the API.

        Returns:
            Number of ``chunks`` rows.
        """
        row = self._conn.execute("SELECT count(*) FROM chunks").fetchone()
        return 0 if row is None else int(row[0])

    def document_count_by_type(self) -> dict[str, int]:
        """Count the ingested documents per file type.

        Backs the ``varagity_corpus_documents_by_type`` gauge (spec_v3
        §6.1a) — the store-derived replacement for the Ingestion
        dashboard's ``increase()``-over-a-flat-counter panel.

        Returns:
            ``file_type`` → document count; empty on an empty corpus.
        """
        rows = self._conn.execute(_DOCUMENTS_BY_TYPE_SQL).fetchall()
        return {str(file_type): int(count) for file_type, count in rows}

    def chunk_count_by_strategy(self) -> dict[str, int]:
        """Count the stored chunks per chunking strategy.

        Backs the ``varagity_corpus_chunks_by_strategy`` gauge (spec_v3
        §6.1a). The strategy is read from each chunk's ``metadata`` JSONB
        (it is not a column); chunks lacking the field count as
        ``unknown``.

        Returns:
            ``chunking_strategy`` → chunk count; empty on an empty corpus.
        """
        rows = self._conn.execute(_CHUNKS_BY_STRATEGY_SQL).fetchall()
        return {str(strategy): int(count) for strategy, count in rows}

    def list_documents(self) -> list[DocumentInfo]:
        """List every ingested document with its extraction mix.

        Backs ``GET /api/documents`` (spec_v2 §4.2): the corpus-management
        table of file, type, chunk count, ingested-at, and how many chunks
        came through OCR versus the digital text layer.

        Returns:
            One :class:`~varagity.stores.records.DocumentInfo` per
            ``documents`` row, newest ingest first (0-chunk documents
            included, with an empty mix).
        """
        rows = self._conn.execute(_LIST_DOCUMENTS_SQL).fetchall()
        return [_row_to_document(row) for row in rows]

    @contextmanager
    def iter_corpus(self) -> Iterator[CorpusSnapshot]:
        """Open one consistent snapshot of the whole corpus (a backup export).

        Everything a RAG bundle holds — every ``documents`` row, then every
        ``chunks`` row with its vector — is read inside **one** REPEATABLE
        READ, READ ONLY transaction (ADR-019), so the export is a single
        moment of the corpus even while an ingest or delete commits beside
        it. The rows stream through server-side cursors, so memory stays
        flat however large the corpus is. Readers never block writers in
        Postgres: the open snapshot holds no lock an ingest would wait on.

        Yields:
            The snapshot, valid until the ``with`` block exits (which ends
            the transaction).
        """
        with self._conn.transaction():
            self._conn.execute(_SNAPSHOT_ISOLATION_SQL)
            row = self._conn.execute(_CORPUS_COUNTS_SQL).fetchone()
            documents, chunks = (0, 0) if row is None else (int(row[0]), int(row[1]))
            yield CorpusSnapshot(self._conn, document_count=documents, chunk_count=chunks)

    @contextmanager
    def corpus_lock(self, *, exclusive: bool) -> Iterator[None]:
        """Hold the RAG corpus's advisory lock for the ``with`` block.

        Ingests hold it **shared** — they overlap each other exactly as they
        always have — and a backup restore holds it **exclusive**, so a
        restore and an ingest in *any* two processes exclude each other
        (ADR-019). The lock belongs to this store's database session: it is
        released when the block exits, and with the connection if the
        process dies first. Never blocks — a held lock is refused at once.

        Args:
            exclusive: Take it exclusively (a restore) rather than shared
                (an ingest).

        Yields:
            Nothing; the corpus lock is held until the block exits.

        Raises:
            CorpusLocked: If the lock is held in a mode that excludes this
                request.
        """
        row = self._conn.execute(_TRY_LOCK_SQL[exclusive], (CORPUS_LOCK_KEY,)).fetchone()
        if row is None or not row[0]:
            raise CorpusLocked(exclusive=exclusive)
        try:
            yield
        finally:
            try:
                self._conn.execute(_UNLOCK_SQL[exclusive], (CORPUS_LOCK_KEY,))
            except psycopg.Error as error:
                # The session is gone (or going): its end releases the lock.
                logger.warning("could not release the corpus lock (%s)", error)

    def corpus_lock_free(self) -> bool:
        """Report whether a restore could take the corpus lock right now.

        Takes the lock exclusively and gives it straight back — so ``False``
        means an ingest, in this process or another, holds it shared. The
        restore route's pre-202 check; the restore itself still takes the
        lock for real (and fails cleanly if an ingest won the race).

        Returns:
            ``True`` when nothing holds the corpus lock.
        """
        row = self._conn.execute(_TRY_LOCK_SQL[True], (CORPUS_LOCK_KEY,)).fetchone()
        if row is None or not row[0]:
            return False
        self._conn.execute(_UNLOCK_SQL[True], (CORPUS_LOCK_KEY,))
        return True

    def replace_corpus(
        self,
        documents: Iterable[CorpusDocumentRow],
        chunks: Iterable[CorpusChunkRow],
        *,
        on_indexing: Callable[[], None] | None = None,
    ) -> tuple[int, int]:
        """Replace every ``documents`` and ``chunks`` row in one transaction.

        The backup restore's write (ADR-019), in this order: empty both
        tables, drop the HNSW index (its definition saved from
        ``pg_indexes``), binary-``COPY`` the documents and then the chunks,
        rebuild the index from that definition, ``ANALYZE``, commit. The
        commit is where the restore takes effect: any failure before it —
        a row the bundle got wrong, a stream that turns out damaged, the
        caller raising mid-iteration — rolls the whole thing back and the
        previous corpus is untouched. ``TRUNCATE`` holds both tables until
        then, so a concurrent reader waits for the restored rows rather
        than seeing half of them.

        The two streams are consumed strictly in order (documents first, as
        the foreign key demands), so a chunk stream may depend on what the
        document stream has already yielded.

        Args:
            documents: Every document row, in any order.
            chunks: Every chunk row with its vector; each ``doc_id`` must be
                among ``documents``.
            on_indexing: Called once the rows are loaded, just before the
                vector index build — the slow tail of a large restore.

        Returns:
            ``(documents, chunks)`` written.

        Raises:
            CorpusTablesBusy: If another session held the tables past the
                lock timeout (nothing was replaced).
        """
        with self._conn.transaction():
            self._conn.execute(_RESTORE_LOCK_TIMEOUT_SQL)
            try:
                self._conn.execute(_TRUNCATE_CORPUS_SQL)
            except psycopg.errors.LockNotAvailable as error:
                raise CorpusTablesBusy(
                    "another database session kept the documents/chunks tables open — "
                    "nothing was replaced; try again"
                ) from error
            row = self._conn.execute(_INDEX_DEFINITION_SQL).fetchone()
            index_sql = _DEFAULT_HNSW_SQL if row is None else str(row[0])
            if row is not None:
                self._conn.execute(_DROP_HNSW_SQL)
            documents_written = self._copy_rows(
                _COPY_DOCUMENTS_SQL,
                _DOCUMENT_COPY_TYPES,
                (
                    (
                        document.doc_id,
                        document.source,
                        document.file_type,
                        document.content_hash,
                        document.n_chunks,
                        document.ingested_at,
                    )
                    for document in documents
                ),
            )
            chunks_written = self._copy_rows(
                _COPY_CHUNKS_SQL,
                _CHUNK_COPY_TYPES,
                (
                    (
                        chunk.chunk_id,
                        chunk.doc_id,
                        chunk.original_index,
                        chunk.chunk_index,
                        chunk.content,
                        chunk.context,
                        chunk.contextualized_content,
                        chunk.embedding,
                        chunk.metadata,
                        chunk.created_at,
                    )
                    for chunk in chunks
                ),
            )
            if on_indexing is not None:
                on_indexing()
            self._conn.execute(index_sql)
            for statement in _ANALYZE_SQL:
                self._conn.execute(statement)
        logger.info(
            "replaced the corpus: %d document(s), %d chunk(s); %s rebuilt",
            documents_written,
            chunks_written,
            _HNSW_INDEX_NAME,
        )
        return documents_written, chunks_written

    def _copy_rows(self, sql: str, types: list[str], rows: Iterable[tuple[Any, ...]]) -> int:
        """Stream rows into one binary ``COPY … FROM STDIN``.

        Args:
            sql: The ``COPY`` statement.
            types: The Postgres type of each column, in order (binary COPY
                carries no type information of its own).
            rows: The column values, row by row.

        Returns:
            The rows written.
        """
        written = 0
        with self._conn.cursor() as cursor, cursor.copy(sql) as copy:
            copy.set_types(types)
            for values in rows:
                copy.write_row(values)
                written += 1
        return written

    def iter_index_rows(self) -> Generator[CorpusIndexRow, None, None]:
        """Stream what the BM25 index mirrors of every chunk.

        Backs the restore's BM25 rebuild (ADR-019): the five fields of each
        stored chunk, through a server-side cursor so the corpus streams in
        bounded memory. The transaction stays open until the iterator is
        exhausted or closed — close it (``contextlib.closing``) when
        abandoning it early.

        Yields:
            One row per chunk, ``original_index`` order.
        """
        with (
            self._conn.transaction(),
            self._conn.cursor(name="varagity_index_rows") as cursor,
        ):
            cursor.itersize = _SNAPSHOT_BATCH_ROWS
            cursor.execute(_INDEX_ROWS_SQL)
            for row in cursor:
                yield CorpusIndexRow(*row)

    def get_document(self, doc_id: str) -> DocumentInfo | None:
        """Fetch one ingested document by id.

        Backs the preview routes (ADR-010): resolving ``doc_id`` →
        ``source``/``content_hash`` happens once per locate/render request,
        so it gets a single-row lookup instead of a full-corpus list.

        Args:
            doc_id: The document's stable id.

        Returns:
            The document row with its extraction mix, or ``None`` when the
            id is unknown.
        """
        row = self._conn.execute(_GET_DOCUMENT_SQL, {"doc_id": doc_id}).fetchone()
        return None if row is None else _row_to_document(row)

    def delete_document(self, doc_id: str) -> int:
        """Delete a document row and, via FK cascade, all its chunks.

        Backs ``ingest --reingest``: pipeline-setting changes (e.g. toggling
        ``CONTEXTUALIZE``) don't change content hashes, so re-processing an
        unchanged file requires deleting its previous ingest first.

        Args:
            doc_id: The document's stable id (deleting an unknown id is a
                no-op).

        Returns:
            The number of ``documents`` rows deleted (0 or 1).
        """
        return self.delete_documents([doc_id])

    def delete_documents(self, doc_ids: Sequence[str]) -> int:
        """Delete several document rows and, via FK cascade, their chunks.

        Backs the corpus table's multi-select delete (spec_v2 §4.2): one
        statement for the whole set, so the marker rows the delete route
        writes last all fall in the same round trip.

        Args:
            doc_ids: The documents' stable ids (unknown ids are no-ops; an
                empty sequence skips the statement entirely).

        Returns:
            The number of ``documents`` rows deleted.
        """
        if not doc_ids:
            return 0
        cursor = self._conn.execute(
            "DELETE FROM documents WHERE doc_id = ANY(%s)", (list(doc_ids),)
        )
        return cursor.rowcount

    def next_original_index(self) -> int:
        """Allocate the next global chunk index (called once per ingest run).

        Returns:
            ``max(original_index) + 1`` across the corpus, or ``0`` when
            empty; the loader then increments monotonically in-process.
        """
        row = self._conn.execute(
            "SELECT COALESCE(MAX(original_index), -1) + 1 FROM chunks"
        ).fetchone()
        return 0 if row is None else int(row[0])

    def upsert_document(
        self, *, doc_id: str, source: str, file_type: str, content_hash: str, n_chunks: int
    ) -> None:
        """Insert or update one ``documents`` row.

        Args:
            doc_id: The document's stable id.
            source: Absolute file path (provenance).
            file_type: ``pdf`` / ``txt`` / ``md``.
            content_hash: The document's content hash.
            n_chunks: Number of chunks ingested (``0`` records a document
                with no extractable text).
        """
        self._conn.execute(
            _UPSERT_DOCUMENT_SQL,
            {
                "doc_id": doc_id,
                "source": source,
                "file_type": file_type,
                "content_hash": content_hash,
                "n_chunks": n_chunks,
            },
        )

    def upsert_chunks(self, records: list[ChunkRecord], embeddings: list[list[float]]) -> None:
        """Insert or update one ``chunks`` row per record.

        Args:
            records: The chunk metadata records.
            embeddings: One vector per record (same order).

        Raises:
            ValueError: If ``records`` and ``embeddings`` lengths differ.
            psycopg.errors.UniqueViolation: If an insert collides on the
                ``(doc_id, original_index)`` identity (an ingest bug — the
                unique index exists to catch it early).
        """
        if len(records) != len(embeddings):
            raise ValueError(
                f"got {len(records)} records but {len(embeddings)} embeddings — "
                "every chunk needs exactly one vector"
            )
        params = [
            {
                "chunk_id": record.chunk_id,
                "doc_id": record.doc_id,
                "original_index": record.original_index,
                "chunk_index": record.chunk_index,
                "content": record.content,
                "context": record.context,
                "contextualized_content": record.contextualized_content,
                "embedding": Vector(embedding),
                "metadata": Json(record.model_dump(mode="json")),
                "created_at": record.created_at,
            }
            for record, embedding in zip(records, embeddings, strict=True)
        ]
        with self._conn.cursor() as cur:
            cur.executemany(_UPSERT_CHUNK_SQL, params)

    def store_document(
        self,
        *,
        doc_id: str,
        source: str,
        file_type: str,
        content_hash: str,
        records: list[ChunkRecord],
        embeddings: list[list[float]],
    ) -> None:
        """Atomically upsert a document row and all its chunks.

        One transaction per document: a partial failure rolls back the
        ``documents`` row too, so a re-run's idempotency check re-attempts
        the file instead of skipping a half-ingested one.

        Args:
            doc_id: The document's stable id.
            source: Absolute file path (provenance).
            file_type: ``pdf`` / ``txt`` / ``md``.
            content_hash: The document's content hash.
            records: The chunk metadata records.
            embeddings: One vector per record (same order).
        """
        with self._conn.transaction():
            self.upsert_document(
                doc_id=doc_id,
                source=source,
                file_type=file_type,
                content_hash=content_hash,
                n_chunks=len(records),
            )
            self.upsert_chunks(records, embeddings)

    def search(
        self, query_vector: list[float], k: int, verbose: int | None = None
    ) -> list[RetrievedChunk]:
        """Cosine top-k search (spec §11.2).

        Args:
            query_vector: The query embedding (e5 **query mode** — see
                :meth:`varagity.models.embeddings.EmbeddingsClient.embed_query`).
            k: Number of chunks to return.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The top-k chunks by cosine similarity, best first, with
            ``score = 1 - cosine_distance``.

        Raises:
            ValueError: If ``verbose`` is invalid.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        rows = self._conn.execute(_SEARCH_SQL, {"qvec": Vector(query_vector), "k": k}).fetchall()
        return [_row_to_retrieved(row) for row in rows]

    def document_chunks(self, doc_id: str) -> list[RetrievedChunk]:
        """Fetch one document's chunks in reading order.

        Backs the eval harness's fact-anchored golden resolution (spec_v2
        §7.4): a chunker sweep must locate which strategy-true chunk holds a
        planted fact, so it scans the document's stored contents rather than
        trusting index-anchored refs authored against other boundaries.

        Args:
            doc_id: The parent document id.

        Returns:
            The document's chunks ordered by ``chunk_index`` (empty when the
            document is unknown or has no extractable text). Chunks carry
            ``score = 0.0`` — there is no query here.
        """
        rows = self._conn.execute(_DOCUMENT_CHUNKS_SQL, {"doc_id": doc_id}).fetchall()
        return [_row_to_retrieved((*row, 0.0)) for row in rows]

    def fetch_by_identity(
        self, keys: list[tuple[str, int]]
    ) -> dict[tuple[str, int], RetrievedChunk]:
        """Fetch full chunk rows by their fusion identity (spec §11.4 hydrate).

        Backs the bm25/hybrid retrievers: Elasticsearch only stores the
        identity and text fields, so complete records (source, page, context,
        full metadata) are hydrated from here. The returned chunks carry
        ``score = 0.0`` — relevance belongs to the caller, which attaches its
        own (BM25 or fused) score.

        Args:
            keys: ``(doc_id, original_index)`` identity tuples.

        Returns:
            A mapping of identity tuple → chunk for every key that exists;
            unknown keys are simply absent (the caller decides whether that
            is an inconsistency worth logging).
        """
        if not keys:
            return {}
        rows = self._conn.execute(
            _FETCH_BY_IDENTITY_SQL,
            {
                "doc_ids": [doc_id for doc_id, _ in keys],
                "original_indexes": [original_index for _, original_index in keys],
            },
        ).fetchall()
        found: dict[tuple[str, int], RetrievedChunk] = {}
        for row in rows:
            chunk = _row_to_retrieved((*row, 0.0))
            found[(chunk.doc_id, chunk.original_index)] = chunk
        return found

__init__(conninfo=None)

Connect and register the pgvector type adapters.

Parameters:

Name Type Description Default
conninfo str | None

libpq connection string; defaults to the POSTGRES_* settings.

 None

Raises:

Type Description
OperationalError

If the database is unreachable.
Source code in varagity/stores/vector_store.py 
456
457
458
459
460
461
462
463
464
465
466
467
def __init__(self, conninfo: str | None = None) -> None:
    """Connect and register the pgvector type adapters.

    Args:
        conninfo: libpq connection string; defaults to the
            ``POSTGRES_*`` settings.

    Raises:
        psycopg.OperationalError: If the database is unreachable.
    """
    self._conn = psycopg.connect(conninfo or default_conninfo(), autocommit=True)
    register_vector(self._conn)

chunk_count()

Count the stored chunks.

Backs the varagity_corpus_chunks gauge (spec_v3 §6.1a): the store is the truthful source for "how big is my corpus", unlike the per-process ingest counters that reset with the API.

Returns:

Type Description
int

Number of chunks rows.
Source code in varagity/stores/vector_store.py 
521
522
523
524
525
526
527
528
529
530
531
532
def chunk_count(self) -> int:
    """Count the stored chunks.

    Backs the ``varagity_corpus_chunks`` gauge (spec_v3 §6.1a): the
    store is the truthful source for "how big is my corpus", unlike the
    per-process ingest counters that reset with the API.

    Returns:
        Number of ``chunks`` rows.
    """
    row = self._conn.execute("SELECT count(*) FROM chunks").fetchone()
    return 0 if row is None else int(row[0])

chunk_count_by_strategy()

Count the stored chunks per chunking strategy.

Backs the varagity_corpus_chunks_by_strategy gauge (spec_v3 §6.1a). The strategy is read from each chunk's metadata JSONB (it is not a column); chunks lacking the field count as unknown.

Returns:

Type Description
dict[str, int]

chunking_strategy → chunk count; empty on an empty corpus.
Source code in varagity/stores/vector_store.py 
547
548
549
550
551
552
553
554
555
556
557
558
559
def chunk_count_by_strategy(self) -> dict[str, int]:
    """Count the stored chunks per chunking strategy.

    Backs the ``varagity_corpus_chunks_by_strategy`` gauge (spec_v3
    §6.1a). The strategy is read from each chunk's ``metadata`` JSONB
    (it is not a column); chunks lacking the field count as
    ``unknown``.

    Returns:
        ``chunking_strategy`` → chunk count; empty on an empty corpus.
    """
    rows = self._conn.execute(_CHUNKS_BY_STRATEGY_SQL).fetchall()
    return {str(strategy): int(count) for strategy, count in rows}

close()

Close the underlying connection (idempotent).

Source code in varagity/stores/vector_store.py 
469
470
471
472
def close(self) -> None:
    """Close the underlying connection (idempotent)."""
    if not self._conn.closed:
        self._conn.close()

corpus_lock(*, exclusive)

Hold the RAG corpus's advisory lock for the with block.

Ingests hold it shared — they overlap each other exactly as they always have — and a backup restore holds it exclusive, so a restore and an ingest in any two processes exclude each other (ADR-019). The lock belongs to this store's database session: it is released when the block exits, and with the connection if the process dies first. Never blocks — a held lock is refused at once.

Parameters:

Name Type Description Default
exclusive bool

Take it exclusively (a restore) rather than shared (an ingest).

 required

Yields:

Type Description
None

Nothing; the corpus lock is held until the block exits.

Raises:

Type Description
CorpusLocked

If the lock is held in a mode that excludes this request.
Source code in varagity/stores/vector_store.py 
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
@contextmanager
def corpus_lock(self, *, exclusive: bool) -> Iterator[None]:
    """Hold the RAG corpus's advisory lock for the ``with`` block.

    Ingests hold it **shared** — they overlap each other exactly as they
    always have — and a backup restore holds it **exclusive**, so a
    restore and an ingest in *any* two processes exclude each other
    (ADR-019). The lock belongs to this store's database session: it is
    released when the block exits, and with the connection if the
    process dies first. Never blocks — a held lock is refused at once.

    Args:
        exclusive: Take it exclusively (a restore) rather than shared
            (an ingest).

    Yields:
        Nothing; the corpus lock is held until the block exits.

    Raises:
        CorpusLocked: If the lock is held in a mode that excludes this
            request.
    """
    row = self._conn.execute(_TRY_LOCK_SQL[exclusive], (CORPUS_LOCK_KEY,)).fetchone()
    if row is None or not row[0]:
        raise CorpusLocked(exclusive=exclusive)
    try:
        yield
    finally:
        try:
            self._conn.execute(_UNLOCK_SQL[exclusive], (CORPUS_LOCK_KEY,))
        except psycopg.Error as error:
            # The session is gone (or going): its end releases the lock.
            logger.warning("could not release the corpus lock (%s)", error)

corpus_lock_free()

Report whether a restore could take the corpus lock right now.

Takes the lock exclusively and gives it straight back — so False means an ingest, in this process or another, holds it shared. The restore route's pre-202 check; the restore itself still takes the lock for real (and fails cleanly if an ingest won the race).

Returns:

Type Description
bool

True when nothing holds the corpus lock.
Source code in varagity/stores/vector_store.py 
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
def corpus_lock_free(self) -> bool:
    """Report whether a restore could take the corpus lock right now.

    Takes the lock exclusively and gives it straight back — so ``False``
    means an ingest, in this process or another, holds it shared. The
    restore route's pre-202 check; the restore itself still takes the
    lock for real (and fails cleanly if an ingest won the race).

    Returns:
        ``True`` when nothing holds the corpus lock.
    """
    row = self._conn.execute(_TRY_LOCK_SQL[True], (CORPUS_LOCK_KEY,)).fetchone()
    if row is None or not row[0]:
        return False
    self._conn.execute(_UNLOCK_SQL[True], (CORPUS_LOCK_KEY,))
    return True

delete_document(doc_id)

Delete a document row and, via FK cascade, all its chunks.

Backs ingest --reingest: pipeline-setting changes (e.g. toggling CONTEXTUALIZE) don't change content hashes, so re-processing an unchanged file requires deleting its previous ingest first.

Parameters:

Name Type Description Default
doc_id str

The document's stable id (deleting an unknown id is a no-op).

 required

Returns:

Type Description
int

The number of documents rows deleted (0 or 1).
Source code in varagity/stores/vector_store.py 
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
def delete_document(self, doc_id: str) -> int:
    """Delete a document row and, via FK cascade, all its chunks.

    Backs ``ingest --reingest``: pipeline-setting changes (e.g. toggling
    ``CONTEXTUALIZE``) don't change content hashes, so re-processing an
    unchanged file requires deleting its previous ingest first.

    Args:
        doc_id: The document's stable id (deleting an unknown id is a
            no-op).

    Returns:
        The number of ``documents`` rows deleted (0 or 1).
    """
    return self.delete_documents([doc_id])

delete_documents(doc_ids)

Delete several document rows and, via FK cascade, their chunks.

Backs the corpus table's multi-select delete (spec_v2 §4.2): one statement for the whole set, so the marker rows the delete route writes last all fall in the same round trip.

Parameters:

Name Type Description Default
doc_ids Sequence[str]

The documents' stable ids (unknown ids are no-ops; an empty sequence skips the statement entirely).

 required

Returns:

Type Description
int

The number of documents rows deleted.
Source code in varagity/stores/vector_store.py 
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
def delete_documents(self, doc_ids: Sequence[str]) -> int:
    """Delete several document rows and, via FK cascade, their chunks.

    Backs the corpus table's multi-select delete (spec_v2 §4.2): one
    statement for the whole set, so the marker rows the delete route
    writes last all fall in the same round trip.

    Args:
        doc_ids: The documents' stable ids (unknown ids are no-ops; an
            empty sequence skips the statement entirely).

    Returns:
        The number of ``documents`` rows deleted.
    """
    if not doc_ids:
        return 0
    cursor = self._conn.execute(
        "DELETE FROM documents WHERE doc_id = ANY(%s)", (list(doc_ids),)
    )
    return cursor.rowcount

document_chunks(doc_id)

Fetch one document's chunks in reading order.

Backs the eval harness's fact-anchored golden resolution (spec_v2 §7.4): a chunker sweep must locate which strategy-true chunk holds a planted fact, so it scans the document's stored contents rather than trusting index-anchored refs authored against other boundaries.

Parameters:

Name Type Description Default
doc_id str

The parent document id.

 required

Returns:

Type Description
list[RetrievedChunk]

The document's chunks ordered by chunk_index (empty when the
list[RetrievedChunk]

document is unknown or has no extractable text). Chunks carry
list[RetrievedChunk]

score = 0.0 — there is no query here.
Source code in varagity/stores/vector_store.py 
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
def document_chunks(self, doc_id: str) -> list[RetrievedChunk]:
    """Fetch one document's chunks in reading order.

    Backs the eval harness's fact-anchored golden resolution (spec_v2
    §7.4): a chunker sweep must locate which strategy-true chunk holds a
    planted fact, so it scans the document's stored contents rather than
    trusting index-anchored refs authored against other boundaries.

    Args:
        doc_id: The parent document id.

    Returns:
        The document's chunks ordered by ``chunk_index`` (empty when the
        document is unknown or has no extractable text). Chunks carry
        ``score = 0.0`` — there is no query here.
    """
    rows = self._conn.execute(_DOCUMENT_CHUNKS_SQL, {"doc_id": doc_id}).fetchall()
    return [_row_to_retrieved((*row, 0.0)) for row in rows]

document_count()

Count the ingested documents.

Backs the settings route's corpus-stale check (spec_v2 §4.7): an empty corpus has nothing to go stale, so a reingest-affecting override change only raises the flag when this is positive.

Returns:

Type Description
int

Number of documents rows.
Source code in varagity/stores/vector_store.py 
508
509
510
511
512
513
514
515
516
517
518
519
def document_count(self) -> int:
    """Count the ingested documents.

    Backs the settings route's corpus-stale check (spec_v2 §4.7): an
    empty corpus has nothing to go stale, so a reingest-affecting
    override change only raises the flag when this is positive.

    Returns:
        Number of ``documents`` rows.
    """
    row = self._conn.execute("SELECT count(*) FROM documents").fetchone()
    return 0 if row is None else int(row[0])

document_count_by_type()

Count the ingested documents per file type.

Backs the varagity_corpus_documents_by_type gauge (spec_v3 §6.1a) — the store-derived replacement for the Ingestion dashboard's increase()-over-a-flat-counter panel.

Returns:

Type Description
dict[str, int]

file_type → document count; empty on an empty corpus.
Source code in varagity/stores/vector_store.py 
534
535
536
537
538
539
540
541
542
543
544
545
def document_count_by_type(self) -> dict[str, int]:
    """Count the ingested documents per file type.

    Backs the ``varagity_corpus_documents_by_type`` gauge (spec_v3
    §6.1a) — the store-derived replacement for the Ingestion
    dashboard's ``increase()``-over-a-flat-counter panel.

    Returns:
        ``file_type`` → document count; empty on an empty corpus.
    """
    rows = self._conn.execute(_DOCUMENTS_BY_TYPE_SQL).fetchall()
    return {str(file_type): int(count) for file_type, count in rows}

document_exists(doc_id, content_hash)

Check whether a document was already ingested (idempotency, spec §8.2).

Parameters:

Name Type Description Default
doc_id str

The document's stable id.

 required
content_hash str

The document's content hash (doc_id already encodes it — checking both is belt-and-braces).

 required

Returns:

Type Description
bool

True if a matching documents row exists.
Source code in varagity/stores/vector_store.py 
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
def document_exists(self, doc_id: str, content_hash: str) -> bool:
    """Check whether a document was already ingested (idempotency, spec §8.2).

    Args:
        doc_id: The document's stable id.
        content_hash: The document's content hash (``doc_id`` already
            encodes it — checking both is belt-and-braces).

    Returns:
        ``True`` if a matching ``documents`` row exists.
    """
    row = self._conn.execute(
        "SELECT 1 FROM documents WHERE doc_id = %s AND content_hash = %s",
        (doc_id, content_hash),
    ).fetchone()
    return row is not None

document_n_chunks(doc_id)

Fetch a document's recorded chunk count.

Lets the loader re-warn about known-empty documents (n_chunks = 0) without re-parsing them.

Parameters:

Name Type Description Default
doc_id str

The document's stable id.

 required

Returns:

Type Description
int | None

The n_chunks value, or None if the document is unknown.
Source code in varagity/stores/vector_store.py 
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
def document_n_chunks(self, doc_id: str) -> int | None:
    """Fetch a document's recorded chunk count.

    Lets the loader re-warn about known-empty documents (``n_chunks = 0``)
    without re-parsing them.

    Args:
        doc_id: The document's stable id.

    Returns:
        The ``n_chunks`` value, or ``None`` if the document is unknown.
    """
    row = self._conn.execute(
        "SELECT n_chunks FROM documents WHERE doc_id = %s", (doc_id,)
    ).fetchone()
    return None if row is None else int(row[0])

fetch_by_identity(keys)

Fetch full chunk rows by their fusion identity (spec §11.4 hydrate).

Backs the bm25/hybrid retrievers: Elasticsearch only stores the identity and text fields, so complete records (source, page, context, full metadata) are hydrated from here. The returned chunks carry score = 0.0 — relevance belongs to the caller, which attaches its own (BM25 or fused) score.

Parameters:

Name Type Description Default
keys list[tuple[str, int]]

(doc_id, original_index) identity tuples.

 required

Returns:

Type Description
dict[tuple[str, int], RetrievedChunk]

A mapping of identity tuple → chunk for every key that exists;
dict[tuple[str, int], RetrievedChunk]

unknown keys are simply absent (the caller decides whether that
dict[tuple[str, int], RetrievedChunk]

is an inconsistency worth logging).
Source code in varagity/stores/vector_store.py 
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
def fetch_by_identity(
    self, keys: list[tuple[str, int]]
) -> dict[tuple[str, int], RetrievedChunk]:
    """Fetch full chunk rows by their fusion identity (spec §11.4 hydrate).

    Backs the bm25/hybrid retrievers: Elasticsearch only stores the
    identity and text fields, so complete records (source, page, context,
    full metadata) are hydrated from here. The returned chunks carry
    ``score = 0.0`` — relevance belongs to the caller, which attaches its
    own (BM25 or fused) score.

    Args:
        keys: ``(doc_id, original_index)`` identity tuples.

    Returns:
        A mapping of identity tuple → chunk for every key that exists;
        unknown keys are simply absent (the caller decides whether that
        is an inconsistency worth logging).
    """
    if not keys:
        return {}
    rows = self._conn.execute(
        _FETCH_BY_IDENTITY_SQL,
        {
            "doc_ids": [doc_id for doc_id, _ in keys],
            "original_indexes": [original_index for _, original_index in keys],
        },
    ).fetchall()
    found: dict[tuple[str, int], RetrievedChunk] = {}
    for row in rows:
        chunk = _row_to_retrieved((*row, 0.0))
        found[(chunk.doc_id, chunk.original_index)] = chunk
    return found

get_document(doc_id)

Fetch one ingested document by id.

Backs the preview routes (ADR-010): resolving doc_id → source/content_hash happens once per locate/render request, so it gets a single-row lookup instead of a full-corpus list.

Parameters:

Name Type Description Default
doc_id str

The document's stable id.

 required

Returns:

Type Description
DocumentInfo | None

The document row with its extraction mix, or None when the
DocumentInfo | None

id is unknown.
Source code in varagity/stores/vector_store.py 
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
def get_document(self, doc_id: str) -> DocumentInfo | None:
    """Fetch one ingested document by id.

    Backs the preview routes (ADR-010): resolving ``doc_id`` →
    ``source``/``content_hash`` happens once per locate/render request,
    so it gets a single-row lookup instead of a full-corpus list.

    Args:
        doc_id: The document's stable id.

    Returns:
        The document row with its extraction mix, or ``None`` when the
        id is unknown.
    """
    row = self._conn.execute(_GET_DOCUMENT_SQL, {"doc_id": doc_id}).fetchone()
    return None if row is None else _row_to_document(row)

iter_corpus()

Open one consistent snapshot of the whole corpus (a backup export).

Everything a RAG bundle holds — every documents row, then every chunks row with its vector — is read inside one REPEATABLE READ, READ ONLY transaction (ADR-019), so the export is a single moment of the corpus even while an ingest or delete commits beside it. The rows stream through server-side cursors, so memory stays flat however large the corpus is. Readers never block writers in Postgres: the open snapshot holds no lock an ingest would wait on.

Yields:

Type Description
CorpusSnapshot

The snapshot, valid until the with block exits (which ends
CorpusSnapshot

the transaction).
Source code in varagity/stores/vector_store.py 
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
@contextmanager
def iter_corpus(self) -> Iterator[CorpusSnapshot]:
    """Open one consistent snapshot of the whole corpus (a backup export).

    Everything a RAG bundle holds — every ``documents`` row, then every
    ``chunks`` row with its vector — is read inside **one** REPEATABLE
    READ, READ ONLY transaction (ADR-019), so the export is a single
    moment of the corpus even while an ingest or delete commits beside
    it. The rows stream through server-side cursors, so memory stays
    flat however large the corpus is. Readers never block writers in
    Postgres: the open snapshot holds no lock an ingest would wait on.

    Yields:
        The snapshot, valid until the ``with`` block exits (which ends
        the transaction).
    """
    with self._conn.transaction():
        self._conn.execute(_SNAPSHOT_ISOLATION_SQL)
        row = self._conn.execute(_CORPUS_COUNTS_SQL).fetchone()
        documents, chunks = (0, 0) if row is None else (int(row[0]), int(row[1]))
        yield CorpusSnapshot(self._conn, document_count=documents, chunk_count=chunks)

iter_index_rows()

Stream what the BM25 index mirrors of every chunk.

Backs the restore's BM25 rebuild (ADR-019): the five fields of each stored chunk, through a server-side cursor so the corpus streams in bounded memory. The transaction stays open until the iterator is exhausted or closed — close it (contextlib.closing) when abandoning it early.

Yields:

Type Description
CorpusIndexRow

One row per chunk, original_index order.
Source code in varagity/stores/vector_store.py 
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
def iter_index_rows(self) -> Generator[CorpusIndexRow, None, None]:
    """Stream what the BM25 index mirrors of every chunk.

    Backs the restore's BM25 rebuild (ADR-019): the five fields of each
    stored chunk, through a server-side cursor so the corpus streams in
    bounded memory. The transaction stays open until the iterator is
    exhausted or closed — close it (``contextlib.closing``) when
    abandoning it early.

    Yields:
        One row per chunk, ``original_index`` order.
    """
    with (
        self._conn.transaction(),
        self._conn.cursor(name="varagity_index_rows") as cursor,
    ):
        cursor.itersize = _SNAPSHOT_BATCH_ROWS
        cursor.execute(_INDEX_ROWS_SQL)
        for row in cursor:
            yield CorpusIndexRow(*row)

list_documents()

List every ingested document with its extraction mix.

Backs GET /api/documents (spec_v2 §4.2): the corpus-management table of file, type, chunk count, ingested-at, and how many chunks came through OCR versus the digital text layer.

Returns:

Name Type Description
One list[DocumentInfo]

class:~varagity.stores.records.DocumentInfo per
list[DocumentInfo]

documents row, newest ingest first (0-chunk documents
list[DocumentInfo]

included, with an empty mix).
Source code in varagity/stores/vector_store.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
def list_documents(self) -> list[DocumentInfo]:
    """List every ingested document with its extraction mix.

    Backs ``GET /api/documents`` (spec_v2 §4.2): the corpus-management
    table of file, type, chunk count, ingested-at, and how many chunks
    came through OCR versus the digital text layer.

    Returns:
        One :class:`~varagity.stores.records.DocumentInfo` per
        ``documents`` row, newest ingest first (0-chunk documents
        included, with an empty mix).
    """
    rows = self._conn.execute(_LIST_DOCUMENTS_SQL).fetchall()
    return [_row_to_document(row) for row in rows]

next_original_index()

Allocate the next global chunk index (called once per ingest run).

Returns:

Type Description
int

max(original_index) + 1 across the corpus, or 0 when
int

empty; the loader then increments monotonically in-process.
Source code in varagity/stores/vector_store.py 
842
843
844
845
846
847
848
849
850
851
852
def next_original_index(self) -> int:
    """Allocate the next global chunk index (called once per ingest run).

    Returns:
        ``max(original_index) + 1`` across the corpus, or ``0`` when
        empty; the loader then increments monotonically in-process.
    """
    row = self._conn.execute(
        "SELECT COALESCE(MAX(original_index), -1) + 1 FROM chunks"
    ).fetchone()
    return 0 if row is None else int(row[0])

replace_corpus(documents, chunks, *, on_indexing=None)

Replace every documents and chunks row in one transaction.

The backup restore's write (ADR-019), in this order: empty both tables, drop the HNSW index (its definition saved from pg_indexes), binary-COPY the documents and then the chunks, rebuild the index from that definition, ANALYZE, commit. The commit is where the restore takes effect: any failure before it — a row the bundle got wrong, a stream that turns out damaged, the caller raising mid-iteration — rolls the whole thing back and the previous corpus is untouched. TRUNCATE holds both tables until then, so a concurrent reader waits for the restored rows rather than seeing half of them.

The two streams are consumed strictly in order (documents first, as the foreign key demands), so a chunk stream may depend on what the document stream has already yielded.

Parameters:

Name Type Description Default
documents Iterable[CorpusDocumentRow]

Every document row, in any order.

 required
chunks Iterable[CorpusChunkRow]

Every chunk row with its vector; each doc_id must be among documents.

 required
on_indexing Callable[[], None] | None

Called once the rows are loaded, just before the vector index build — the slow tail of a large restore.

 None

Returns:

Type Description
tuple[int, int]

(documents, chunks) written.

Raises:

Type Description
CorpusTablesBusy

If another session held the tables past the lock timeout (nothing was replaced).
Source code in varagity/stores/vector_store.py 
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
def replace_corpus(
    self,
    documents: Iterable[CorpusDocumentRow],
    chunks: Iterable[CorpusChunkRow],
    *,
    on_indexing: Callable[[], None] | None = None,
) -> tuple[int, int]:
    """Replace every ``documents`` and ``chunks`` row in one transaction.

    The backup restore's write (ADR-019), in this order: empty both
    tables, drop the HNSW index (its definition saved from
    ``pg_indexes``), binary-``COPY`` the documents and then the chunks,
    rebuild the index from that definition, ``ANALYZE``, commit. The
    commit is where the restore takes effect: any failure before it —
    a row the bundle got wrong, a stream that turns out damaged, the
    caller raising mid-iteration — rolls the whole thing back and the
    previous corpus is untouched. ``TRUNCATE`` holds both tables until
    then, so a concurrent reader waits for the restored rows rather
    than seeing half of them.

    The two streams are consumed strictly in order (documents first, as
    the foreign key demands), so a chunk stream may depend on what the
    document stream has already yielded.

    Args:
        documents: Every document row, in any order.
        chunks: Every chunk row with its vector; each ``doc_id`` must be
            among ``documents``.
        on_indexing: Called once the rows are loaded, just before the
            vector index build — the slow tail of a large restore.

    Returns:
        ``(documents, chunks)`` written.

    Raises:
        CorpusTablesBusy: If another session held the tables past the
            lock timeout (nothing was replaced).
    """
    with self._conn.transaction():
        self._conn.execute(_RESTORE_LOCK_TIMEOUT_SQL)
        try:
            self._conn.execute(_TRUNCATE_CORPUS_SQL)
        except psycopg.errors.LockNotAvailable as error:
            raise CorpusTablesBusy(
                "another database session kept the documents/chunks tables open — "
                "nothing was replaced; try again"
            ) from error
        row = self._conn.execute(_INDEX_DEFINITION_SQL).fetchone()
        index_sql = _DEFAULT_HNSW_SQL if row is None else str(row[0])
        if row is not None:
            self._conn.execute(_DROP_HNSW_SQL)
        documents_written = self._copy_rows(
            _COPY_DOCUMENTS_SQL,
            _DOCUMENT_COPY_TYPES,
            (
                (
                    document.doc_id,
                    document.source,
                    document.file_type,
                    document.content_hash,
                    document.n_chunks,
                    document.ingested_at,
                )
                for document in documents
            ),
        )
        chunks_written = self._copy_rows(
            _COPY_CHUNKS_SQL,
            _CHUNK_COPY_TYPES,
            (
                (
                    chunk.chunk_id,
                    chunk.doc_id,
                    chunk.original_index,
                    chunk.chunk_index,
                    chunk.content,
                    chunk.context,
                    chunk.contextualized_content,
                    chunk.embedding,
                    chunk.metadata,
                    chunk.created_at,
                )
                for chunk in chunks
            ),
        )
        if on_indexing is not None:
            on_indexing()
        self._conn.execute(index_sql)
        for statement in _ANALYZE_SQL:
            self._conn.execute(statement)
    logger.info(
        "replaced the corpus: %d document(s), %d chunk(s); %s rebuilt",
        documents_written,
        chunks_written,
        _HNSW_INDEX_NAME,
    )
    return documents_written, chunks_written

search(query_vector, k, verbose=None)

Cosine top-k search (spec §11.2).

Parameters:

Name Type Description Default
query_vector list[float]

The query embedding (e5 query mode — see :meth:varagity.models.embeddings.EmbeddingsClient.embed_query).

 required
k int

Number of chunks to return.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
list[RetrievedChunk]

The top-k chunks by cosine similarity, best first, with
list[RetrievedChunk]

score = 1 - cosine_distance.

Raises:

Type Description
ValueError

If verbose is invalid.
Source code in varagity/stores/vector_store.py 
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
def search(
    self, query_vector: list[float], k: int, verbose: int | None = None
) -> list[RetrievedChunk]:
    """Cosine top-k search (spec §11.2).

    Args:
        query_vector: The query embedding (e5 **query mode** — see
            :meth:`varagity.models.embeddings.EmbeddingsClient.embed_query`).
        k: Number of chunks to return.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The top-k chunks by cosine similarity, best first, with
        ``score = 1 - cosine_distance``.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    rows = self._conn.execute(_SEARCH_SQL, {"qvec": Vector(query_vector), "k": k}).fetchall()
    return [_row_to_retrieved(row) for row in rows]

store_document(*, doc_id, source, file_type, content_hash, records, embeddings)

Atomically upsert a document row and all its chunks.

One transaction per document: a partial failure rolls back the documents row too, so a re-run's idempotency check re-attempts the file instead of skipping a half-ingested one.

Parameters:

Name Type Description Default
doc_id str

The document's stable id.

 required
source str

Absolute file path (provenance).

 required
file_type str

pdf / txt / md.

 required
content_hash str

The document's content hash.

 required
records list[ChunkRecord]

The chunk metadata records.

 required
embeddings list[list[float]]

One vector per record (same order).

 required
Source code in varagity/stores/vector_store.py 
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
def store_document(
    self,
    *,
    doc_id: str,
    source: str,
    file_type: str,
    content_hash: str,
    records: list[ChunkRecord],
    embeddings: list[list[float]],
) -> None:
    """Atomically upsert a document row and all its chunks.

    One transaction per document: a partial failure rolls back the
    ``documents`` row too, so a re-run's idempotency check re-attempts
    the file instead of skipping a half-ingested one.

    Args:
        doc_id: The document's stable id.
        source: Absolute file path (provenance).
        file_type: ``pdf`` / ``txt`` / ``md``.
        content_hash: The document's content hash.
        records: The chunk metadata records.
        embeddings: One vector per record (same order).
    """
    with self._conn.transaction():
        self.upsert_document(
            doc_id=doc_id,
            source=source,
            file_type=file_type,
            content_hash=content_hash,
            n_chunks=len(records),
        )
        self.upsert_chunks(records, embeddings)

upsert_chunks(records, embeddings)

Insert or update one chunks row per record.

Parameters:

Name Type Description Default
records list[ChunkRecord]

The chunk metadata records.

 required
embeddings list[list[float]]

One vector per record (same order).

 required

Raises:

Type Description
ValueError

If records and embeddings lengths differ.
UniqueViolation

If an insert collides on the (doc_id, original_index) identity (an ingest bug — the unique index exists to catch it early).
Source code in varagity/stores/vector_store.py 
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
def upsert_chunks(self, records: list[ChunkRecord], embeddings: list[list[float]]) -> None:
    """Insert or update one ``chunks`` row per record.

    Args:
        records: The chunk metadata records.
        embeddings: One vector per record (same order).

    Raises:
        ValueError: If ``records`` and ``embeddings`` lengths differ.
        psycopg.errors.UniqueViolation: If an insert collides on the
            ``(doc_id, original_index)`` identity (an ingest bug — the
            unique index exists to catch it early).
    """
    if len(records) != len(embeddings):
        raise ValueError(
            f"got {len(records)} records but {len(embeddings)} embeddings — "
            "every chunk needs exactly one vector"
        )
    params = [
        {
            "chunk_id": record.chunk_id,
            "doc_id": record.doc_id,
            "original_index": record.original_index,
            "chunk_index": record.chunk_index,
            "content": record.content,
            "context": record.context,
            "contextualized_content": record.contextualized_content,
            "embedding": Vector(embedding),
            "metadata": Json(record.model_dump(mode="json")),
            "created_at": record.created_at,
        }
        for record, embedding in zip(records, embeddings, strict=True)
    ]
    with self._conn.cursor() as cur:
        cur.executemany(_UPSERT_CHUNK_SQL, params)

upsert_document(*, doc_id, source, file_type, content_hash, n_chunks)

Insert or update one documents row.

Parameters:

Name Type Description Default
doc_id str

The document's stable id.

 required
source str

Absolute file path (provenance).

 required
file_type str

pdf / txt / md.

 required
content_hash str

The document's content hash.

 required
n_chunks int

Number of chunks ingested (0 records a document with no extractable text).

 required
Source code in varagity/stores/vector_store.py 
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
def upsert_document(
    self, *, doc_id: str, source: str, file_type: str, content_hash: str, n_chunks: int
) -> None:
    """Insert or update one ``documents`` row.

    Args:
        doc_id: The document's stable id.
        source: Absolute file path (provenance).
        file_type: ``pdf`` / ``txt`` / ``md``.
        content_hash: The document's content hash.
        n_chunks: Number of chunks ingested (``0`` records a document
            with no extractable text).
    """
    self._conn.execute(
        _UPSERT_DOCUMENT_SQL,
        {
            "doc_id": doc_id,
            "source": source,
            "file_type": file_type,
            "content_hash": content_hash,
            "n_chunks": n_chunks,
        },
    )

CorpusChunkRow dataclass

One chunks row, every column, as a corpus snapshot reads it.

Attributes:

Name Type Description
chunk_id str

{doc_id}::{chunk_index}.
doc_id str

The owning document.
original_index int

The corpus-global index (the cross-store join key).
chunk_index int

Position within the document.
content str

The original chunk text.
context str | None

The contextual blurb (None when contextualization was off).
contextualized_content str

The embedded and BM25-indexed text.
metadata dict[str, Any]

The full ChunkRecord JSON.
created_at datetime

The chunk's creation timestamp.
embedding Vector

The passage vector, exactly as stored (read in pgvector's binary format, so no decimal round trip touches a bit).
Source code in varagity/stores/vector_store.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
@dataclass(frozen=True, slots=True)
class CorpusChunkRow:
    """One ``chunks`` row, every column, as a corpus snapshot reads it.

    Attributes:
        chunk_id: ``{doc_id}::{chunk_index}``.
        doc_id: The owning document.
        original_index: The corpus-global index (the cross-store join key).
        chunk_index: Position within the document.
        content: The original chunk text.
        context: The contextual blurb (``None`` when contextualization was
            off).
        contextualized_content: The embedded and BM25-indexed text.
        metadata: The full ``ChunkRecord`` JSON.
        created_at: The chunk's creation timestamp.
        embedding: The passage vector, exactly as stored (read in pgvector's
            binary format, so no decimal round trip touches a bit).
    """

    chunk_id: str
    doc_id: str
    original_index: int
    chunk_index: int
    content: str
    context: str | None
    contextualized_content: str
    metadata: dict[str, Any]
    created_at: datetime
    embedding: Vector

CorpusDocumentRow dataclass

One documents row, every column, as a corpus snapshot reads it.

Attributes:

Name Type Description
doc_id str

The document's stable id.
source str

Absolute file path recorded at ingest time.
file_type str

File extension without the dot.
content_hash str

sha256 of the source file's bytes at ingest time.
n_chunks int

Chunks ingested (0 = no extractable text).
ingested_at datetime

When the document (last) landed in the stores.
Source code in varagity/stores/vector_store.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@dataclass(frozen=True, slots=True)
class CorpusDocumentRow:
    """One ``documents`` row, every column, as a corpus snapshot reads it.

    Attributes:
        doc_id: The document's stable id.
        source: Absolute file path recorded at ingest time.
        file_type: File extension without the dot.
        content_hash: sha256 of the source file's bytes at ingest time.
        n_chunks: Chunks ingested (``0`` = no extractable text).
        ingested_at: When the document (last) landed in the stores.
    """

    doc_id: str
    source: str
    file_type: str
    content_hash: str
    n_chunks: int
    ingested_at: datetime

CorpusIndexRow dataclass

What the BM25 index mirrors of one chunk — its five fields.

Attributes:

Name Type Description
chunk_id str

{doc_id}::{chunk_index} (the Elasticsearch _id).
doc_id str

The owning document.
original_index int

The corpus-global index (the fusion join key).
content str

The original chunk text.
contextualized_content str

The embedded and BM25-indexed text.
Source code in varagity/stores/vector_store.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
@dataclass(frozen=True, slots=True)
class CorpusIndexRow:
    """What the BM25 index mirrors of one chunk — its five fields.

    Attributes:
        chunk_id: ``{doc_id}::{chunk_index}`` (the Elasticsearch ``_id``).
        doc_id: The owning document.
        original_index: The corpus-global index (the fusion join key).
        content: The original chunk text.
        contextualized_content: The embedded and BM25-indexed text.
    """

    chunk_id: str
    doc_id: str
    original_index: int
    content: str
    contextualized_content: str

CorpusLocked

Bases: RuntimeError

Raised when the corpus lock is held in a mode that excludes the request.

Attributes:

Name Type Description
exclusive

What the refused request wanted — True for a restore (refused by an ingest's shared hold), False for an ingest (refused by a restore's exclusive hold).
Source code in varagity/stores/vector_store.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
class CorpusLocked(RuntimeError):
    """Raised when the corpus lock is held in a mode that excludes the request.

    Attributes:
        exclusive: What the refused request wanted — ``True`` for a restore
            (refused by an ingest's shared hold), ``False`` for an ingest
            (refused by a restore's exclusive hold).
    """

    def __init__(self, *, exclusive: bool) -> None:
        """Word the refusal for whoever was refused.

        Args:
            exclusive: Whether the refused request wanted the lock
                exclusively.
        """
        super().__init__(
            "an ingest is running in another process (it holds the corpus lock) — "
            "try again when it finishes"
            if exclusive
            else "the corpus is being restored — try again when the restore finishes"
        )
        self.exclusive = exclusive

__init__(*, exclusive)

Word the refusal for whoever was refused.

Parameters:

Name Type Description Default
exclusive bool

Whether the refused request wanted the lock exclusively.

 required
Source code in varagity/stores/vector_store.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def __init__(self, *, exclusive: bool) -> None:
    """Word the refusal for whoever was refused.

    Args:
        exclusive: Whether the refused request wanted the lock
            exclusively.
    """
    super().__init__(
        "an ingest is running in another process (it holds the corpus lock) — "
        "try again when it finishes"
        if exclusive
        else "the corpus is being restored — try again when the restore finishes"
    )
    self.exclusive = exclusive

CorpusSnapshot

One consistent, read-only view of documents and chunks.

Handed out by :meth:ContextualVectorDB.iter_corpus for the life of its with block: the counts and both row streams come from the same REPEATABLE READ transaction, so an ingest or delete committing mid-export can never tear a bundle — every row belongs to one moment, and the counts agree with the streams.

Attributes:

Name Type Description
document_count

documents rows in the snapshot.
chunk_count

chunks rows in the snapshot.
Source code in varagity/stores/vector_store.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
class CorpusSnapshot:
    """One consistent, read-only view of ``documents`` and ``chunks``.

    Handed out by :meth:`ContextualVectorDB.iter_corpus` for the life of its
    ``with`` block: the counts and both row streams come from the same
    REPEATABLE READ transaction, so an ingest or delete committing mid-export
    can never tear a bundle — every row belongs to one moment, and the
    counts agree with the streams.

    Attributes:
        document_count: ``documents`` rows in the snapshot.
        chunk_count: ``chunks`` rows in the snapshot.
    """

    def __init__(
        self, conn: psycopg.Connection[Any], *, document_count: int, chunk_count: int
    ) -> None:
        """Bind the snapshot to its open transaction.

        Args:
            conn: The connection holding the snapshot's transaction.
            document_count: ``documents`` rows in the snapshot.
            chunk_count: ``chunks`` rows in the snapshot.
        """
        self._conn = conn
        self.document_count = document_count
        self.chunk_count = chunk_count

    def documents(self) -> Iterator[CorpusDocumentRow]:
        """Stream every ``documents`` row, ``doc_id`` order.

        Yields:
            One row per document.
        """
        for row in self._stream("varagity_snapshot_documents", _CORPUS_DOCUMENTS_SQL):
            yield CorpusDocumentRow(*row)

    def chunks(self) -> Iterator[CorpusChunkRow]:
        """Stream every ``chunks`` row with its vector, ingest order.

        Yields:
            One row per chunk, ordered by ``original_index`` (``doc_id``
            breaking ties) — the order a bundle's ``chunks.jsonl`` and its
            embedding matrix share.
        """
        for row in self._stream("varagity_snapshot_chunks", _CORPUS_CHUNKS_SQL):
            yield CorpusChunkRow(*row)

    def _stream(self, name: str, sql: str) -> Iterator[tuple[Any, ...]]:
        """Run one query through a server-side cursor inside the snapshot.

        A named (server-side) cursor fetches in batches, so a corpus of any
        size streams in bounded memory; binary results hand the vectors over
        bit-exact.

        Args:
            name: The server-side cursor's name.
            sql: The query to stream.

        Yields:
            The raw result rows, in query order.
        """
        with self._conn.cursor(name=name, binary=True) as cursor:
            cursor.itersize = _SNAPSHOT_BATCH_ROWS
            cursor.execute(sql)
            yield from cursor

__init__(conn, *, document_count, chunk_count)

Bind the snapshot to its open transaction.

Parameters:

Name Type Description Default
conn Connection[Any]

The connection holding the snapshot's transaction.

 required
document_count int

documents rows in the snapshot.

 required
chunk_count int

chunks rows in the snapshot.

 required
Source code in varagity/stores/vector_store.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
def __init__(
    self, conn: psycopg.Connection[Any], *, document_count: int, chunk_count: int
) -> None:
    """Bind the snapshot to its open transaction.

    Args:
        conn: The connection holding the snapshot's transaction.
        document_count: ``documents`` rows in the snapshot.
        chunk_count: ``chunks`` rows in the snapshot.
    """
    self._conn = conn
    self.document_count = document_count
    self.chunk_count = chunk_count

chunks()

Stream every chunks row with its vector, ingest order.

Yields:

Type Description
CorpusChunkRow

One row per chunk, ordered by original_index (doc_id
CorpusChunkRow

breaking ties) — the order a bundle's chunks.jsonl and its
CorpusChunkRow

embedding matrix share.
Source code in varagity/stores/vector_store.py 
397
398
399
400
401
402
403
404
405
406
def chunks(self) -> Iterator[CorpusChunkRow]:
    """Stream every ``chunks`` row with its vector, ingest order.

    Yields:
        One row per chunk, ordered by ``original_index`` (``doc_id``
        breaking ties) — the order a bundle's ``chunks.jsonl`` and its
        embedding matrix share.
    """
    for row in self._stream("varagity_snapshot_chunks", _CORPUS_CHUNKS_SQL):
        yield CorpusChunkRow(*row)

documents()

Stream every documents row, doc_id order.

Yields:

Type Description
CorpusDocumentRow

One row per document.
Source code in varagity/stores/vector_store.py 
388
389
390
391
392
393
394
395
def documents(self) -> Iterator[CorpusDocumentRow]:
    """Stream every ``documents`` row, ``doc_id`` order.

    Yields:
        One row per document.
    """
    for row in self._stream("varagity_snapshot_documents", _CORPUS_DOCUMENTS_SQL):
        yield CorpusDocumentRow(*row)

CorpusTablesBusy

Bases: RuntimeError

Raised when a restore can't get the corpus tables to itself in time.

Some session kept documents/chunks open past the restore's lock timeout (an idle-in-transaction connection, a long dump). Nothing was replaced: the restore's transaction rolled back.

Source code in varagity/stores/vector_store.py 
280
281
282
283
284
285
286
class CorpusTablesBusy(RuntimeError):
    """Raised when a restore can't get the corpus tables to itself in time.

    Some session kept ``documents``/``chunks`` open past the restore's lock
    timeout (an idle-in-transaction connection, a long dump). Nothing was
    replaced: the restore's transaction rolled back.
    """

default_conninfo()

Build the connection string from settings.

Shared by every Postgres-backed store (this one, the conversation store) and the migration runner, so POSTGRES_* is interpreted in exactly one place.

Returns:

Type Description
str

A libpq conninfo string for the configured PostgreSQL instance.
Source code in varagity/stores/vector_store.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
def default_conninfo() -> str:
    """Build the connection string from settings.

    Shared by every Postgres-backed store (this one, the conversation
    store) and the migration runner, so ``POSTGRES_*`` is interpreted in
    exactly one place.

    Returns:
        A libpq conninfo string for the configured PostgreSQL instance.
    """
    settings = get_settings()
    return psycopg.conninfo.make_conninfo(
        host=settings.POSTGRES_HOST,
        port=settings.POSTGRES_PORT,
        dbname=settings.POSTGRES_DB,
        user=settings.POSTGRES_USER,
        password=settings.POSTGRES_PASSWORD,
    )

varagity.stores.bm25_store

ElasticsearchBM25 — the sparse (contextual BM25) store.

Ported near-verbatim from the Anthropic cookbook's ElasticsearchBM25 (spec §11.3): an index with the spec §8.3 mapping — content and contextualized_content analyzed with the built-in english analyzer under BM25 similarity, identity fields stored but not indexed — plus bulk indexing and a multi_match search over both text fields. Because chunks are contextualized before they reach this store, the BM25 index is contextual from its first document.

Identity fields are mapped "index": false but keep doc values, so term-level queries against them (delete_documents' delete_by_query) still work — just via a slower doc-values scan, which is fine at dev scale.

Every field of an indexed document is a copy of a chunks column, so the whole index can be rebuilt from the stored rows with no model call — :meth:ElasticsearchBM25.rebuild, which a backup restore runs (ADR-019).

BM25Hit

Bases: BaseModel

One BM25 search result (the spec §11.3 return fields).

Carries the fusion identity plus the indexed texts — not the full metadata record, which lives in pgvector; retrievers hydrate complete rows by (doc_id, original_index).

Attributes:

Name Type Description
doc_id str

Parent document id.
original_index int

Global chunk index (fusion identity).
content str

Original chunk text.
contextualized_content str

The situated text that was BM25-indexed.
score float

Elasticsearch BM25 relevance score (unbounded; higher is better).
Source code in varagity/stores/bm25_store.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class BM25Hit(BaseModel):
    """One BM25 search result (the spec §11.3 return fields).

    Carries the fusion identity plus the indexed texts — not the full
    metadata record, which lives in pgvector; retrievers hydrate complete
    rows by ``(doc_id, original_index)``.

    Attributes:
        doc_id: Parent document id.
        original_index: Global chunk index (fusion identity).
        content: Original chunk text.
        contextualized_content: The situated text that was BM25-indexed.
        score: Elasticsearch BM25 relevance score (unbounded; higher is
            better).
    """

    doc_id: str
    original_index: int
    content: str
    contextualized_content: str
    score: float

ElasticsearchBM25

Bases: ClosingContextMixin

Sparse keyword store over an Elasticsearch BM25 index.

Owns one client. Use as a context manager or call :meth:close when done. The index is created lazily via :meth:create_index (the loader calls it idempotently at ingest start).

Source code in varagity/stores/bm25_store.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
class ElasticsearchBM25(ClosingContextMixin):
    """Sparse keyword store over an Elasticsearch BM25 index.

    Owns one client. Use as a context manager or call :meth:`close` when
    done. The index is created lazily via :meth:`create_index` (the loader
    calls it idempotently at ingest start).
    """

    def __init__(self, *, url: str | None = None, index_name: str | None = None) -> None:
        """Create the client (no I/O until the first operation).

        Args:
            url: Elasticsearch base URL; defaults to
                ``settings.ELASTICSEARCH_URL``.
            index_name: BM25 index name; defaults to
                ``settings.BM25_INDEX_NAME``.
        """
        settings = get_settings()
        self.index_name = index_name or settings.BM25_INDEX_NAME
        self._client = Elasticsearch(
            url or settings.ELASTICSEARCH_URL,
            max_retries=0,  # tenacity owns retries (see module docstring)
            request_timeout=_REQUEST_TIMEOUT_S,
        )

    def close(self) -> None:
        """Close the underlying client (idempotent)."""
        self._client.close()

    @_es_retry
    def create_index(self) -> bool:
        """Create the BM25 index if it doesn't exist (idempotent).

        Returns:
            ``True`` if the index was created, ``False`` if it already
            existed.

        Raises:
            elastic_transport.ConnectionError: If Elasticsearch is still
                unreachable after retries.
        """
        if self._client.indices.exists(index=self.index_name):
            return False
        self._client.indices.create(
            index=self.index_name, settings=_INDEX_SETTINGS, mappings=_INDEX_MAPPINGS
        )
        logger.info("created BM25 index %r", self.index_name)
        return True

    @_es_retry
    def index_chunks(self, records: list[ChunkRecord]) -> int:
        """Bulk-index chunks, then refresh so they are searchable.

        Documents are addressed by ``chunk_id``, so re-indexing the same
        chunk overwrites instead of duplicating — the sparse-side counterpart
        of the vector store's ``ON CONFLICT (chunk_id) DO UPDATE``.

        Args:
            records: The chunk metadata records to index.

        Returns:
            The number of successfully indexed documents.

        Raises:
            elasticsearch.helpers.BulkIndexError: If any document is
                rejected (permanent — not retried).
            elastic_transport.ConnectionError: If Elasticsearch is still
                unreachable after retries.
        """
        if not records:
            return 0
        actions = [self._action(record) for record in records]
        success, _ = helpers.bulk(self._client, actions)
        self._client.indices.refresh(index=self.index_name)
        return success

    def _action(self, chunk: IndexableChunk) -> dict[str, Any]:
        """Build one chunk's bulk ``index`` action.

        Args:
            chunk: The chunk to index.

        Returns:
            The action, addressed by ``chunk_id``.
        """
        return {
            "_index": self.index_name,
            "_id": chunk.chunk_id,
            "_source": {
                "content": chunk.content,
                "contextualized_content": chunk.contextualized_content,
                "doc_id": chunk.doc_id,
                "chunk_id": chunk.chunk_id,
                "original_index": chunk.original_index,
            },
        }

    def rebuild(
        self,
        chunks: Iterable[IndexableChunk],
        *,
        on_indexed: Callable[[int], None] | None = None,
    ) -> int:
        """Recreate the index from scratch and stream every chunk into it.

        A backup restore's second half (ADR-019): the index is deleted and
        created anew with the spec mapping — so nothing of the previous
        corpus survives, and no auto-created default mapping can either —
        then the chunks stream in through ``helpers.streaming_bulk`` (bounded
        memory however large the corpus) and a final refresh makes them
        searchable. Index recreation is retried like every other call; a
        failure mid-stream raises, and running the rebuild again repairs the
        index.

        Args:
            chunks: Every chunk the index should hold.
            on_indexed: Called with the running count after each indexed
                chunk (the restore's progress ticks).

        Returns:
            The number of chunks indexed.

        Raises:
            elasticsearch.helpers.BulkIndexError: If any chunk is rejected.
            elastic_transport.ConnectionError: If Elasticsearch is
                unreachable (after retries, for the index recreation).
        """
        self._recreate_index()
        indexed = 0
        for ok, _ in helpers.streaming_bulk(
            self._client,
            (self._action(chunk) for chunk in chunks),
            chunk_size=_REBUILD_CHUNK_SIZE,
            max_retries=_REBUILD_MAX_RETRIES,
        ):
            if ok:
                indexed += 1
                if on_indexed is not None:
                    on_indexed(indexed)
        self._refresh()
        logger.info("rebuilt BM25 index %r with %d chunk(s)", self.index_name, indexed)
        return indexed

    @_es_retry
    def _recreate_index(self) -> None:
        """Delete the index (if any) and create it again, empty.

        Idempotent as a pair, which is what makes retrying it safe: a create
        that timed out client-side after succeeding is simply redone.
        """
        self._client.indices.delete(index=self.index_name, ignore_unavailable=True)
        self._client.indices.create(
            index=self.index_name, settings=_INDEX_SETTINGS, mappings=_INDEX_MAPPINGS
        )

    @_es_retry
    def _refresh(self) -> None:
        """Make everything indexed so far searchable."""
        self._client.indices.refresh(index=self.index_name)

    def delete_document(self, doc_id: str) -> int:
        """Delete all of a document's chunks (``--reingest`` consistency).

        Keeps re-ingestion consistent across both stores: the vector store
        deletes via FK cascade, this store via ``delete_by_query``.

        Args:
            doc_id: The document's stable id (deleting an unknown id is a
                no-op).

        Returns:
            The number of chunks deleted.

        Raises:
            elastic_transport.ConnectionError: If Elasticsearch is still
                unreachable after retries.
        """
        return self.delete_documents([doc_id])

    @_es_retry
    def delete_documents(self, doc_ids: Sequence[str]) -> int:
        """Delete every chunk of several documents in one pass (bulk GC).

        One ``terms`` ``delete_by_query`` rather than a query per document:
        the whole set costs a single round trip and a single forced index
        refresh, which is what makes the corpus table's multi-select delete
        (spec_v2 §4.2) cheap. The retry lives here, so the single-document
        wrapper inherits it exactly once.

        Args:
            doc_ids: The documents' stable ids (unknown ids are no-ops; an
                empty sequence skips the round trip entirely).

        Returns:
            The number of chunks deleted across all of them.

        Raises:
            elastic_transport.ConnectionError: If Elasticsearch is still
                unreachable after retries.
        """
        if not doc_ids:
            return 0
        response = self._client.delete_by_query(
            index=self.index_name,
            query={"terms": {"doc_id": list(doc_ids)}},
            refresh=True,
        )
        return int(response["deleted"])

    @_es_retry
    def search(self, query: str, k: int, verbose: int | None = None) -> list[BM25Hit]:
        """BM25 top-k search via ``multi_match`` over both text fields.

        Args:
            query: The user's query, unformatted (Elasticsearch's ``english``
                analyzer does its own normalization).
            k: Number of hits to return.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The top-k hits, best first.

        Raises:
            ValueError: If ``verbose`` is invalid.
            elastic_transport.ConnectionError: If Elasticsearch is still
                unreachable after retries.
        """
        check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
        response = self._client.search(
            index=self.index_name,
            query={
                "multi_match": {
                    "query": query,
                    "fields": ["content", "contextualized_content"],
                }
            },
            size=k,
        )
        return [
            BM25Hit(
                doc_id=hit["_source"]["doc_id"],
                original_index=hit["_source"]["original_index"],
                content=hit["_source"]["content"],
                contextualized_content=hit["_source"]["contextualized_content"],
                score=hit["_score"],
            )
            for hit in response["hits"]["hits"]
        ]

__init__(*, url=None, index_name=None)

Create the client (no I/O until the first operation).

Parameters:

Name Type Description Default
url str | None

Elasticsearch base URL; defaults to settings.ELASTICSEARCH_URL.

 None
index_name str | None

BM25 index name; defaults to settings.BM25_INDEX_NAME.

 None
Source code in varagity/stores/bm25_store.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def __init__(self, *, url: str | None = None, index_name: str | None = None) -> None:
    """Create the client (no I/O until the first operation).

    Args:
        url: Elasticsearch base URL; defaults to
            ``settings.ELASTICSEARCH_URL``.
        index_name: BM25 index name; defaults to
            ``settings.BM25_INDEX_NAME``.
    """
    settings = get_settings()
    self.index_name = index_name or settings.BM25_INDEX_NAME
    self._client = Elasticsearch(
        url or settings.ELASTICSEARCH_URL,
        max_retries=0,  # tenacity owns retries (see module docstring)
        request_timeout=_REQUEST_TIMEOUT_S,
    )

close()

Close the underlying client (idempotent).

Source code in varagity/stores/bm25_store.py 
179
180
181
def close(self) -> None:
    """Close the underlying client (idempotent)."""
    self._client.close()

create_index()

Create the BM25 index if it doesn't exist (idempotent).

Returns:

Type Description
bool

True if the index was created, False if it already
bool

existed.

Raises:

Type Description
ConnectionError

If Elasticsearch is still unreachable after retries.
Source code in varagity/stores/bm25_store.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
@_es_retry
def create_index(self) -> bool:
    """Create the BM25 index if it doesn't exist (idempotent).

    Returns:
        ``True`` if the index was created, ``False`` if it already
        existed.

    Raises:
        elastic_transport.ConnectionError: If Elasticsearch is still
            unreachable after retries.
    """
    if self._client.indices.exists(index=self.index_name):
        return False
    self._client.indices.create(
        index=self.index_name, settings=_INDEX_SETTINGS, mappings=_INDEX_MAPPINGS
    )
    logger.info("created BM25 index %r", self.index_name)
    return True

delete_document(doc_id)

Delete all of a document's chunks (--reingest consistency).

Keeps re-ingestion consistent across both stores: the vector store deletes via FK cascade, this store via delete_by_query.

Parameters:

Name Type Description Default
doc_id str

The document's stable id (deleting an unknown id is a no-op).

 required

Returns:

Type Description
int

The number of chunks deleted.

Raises:

Type Description
ConnectionError

If Elasticsearch is still unreachable after retries.
Source code in varagity/stores/bm25_store.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def delete_document(self, doc_id: str) -> int:
    """Delete all of a document's chunks (``--reingest`` consistency).

    Keeps re-ingestion consistent across both stores: the vector store
    deletes via FK cascade, this store via ``delete_by_query``.

    Args:
        doc_id: The document's stable id (deleting an unknown id is a
            no-op).

    Returns:
        The number of chunks deleted.

    Raises:
        elastic_transport.ConnectionError: If Elasticsearch is still
            unreachable after retries.
    """
    return self.delete_documents([doc_id])

delete_documents(doc_ids)

Delete every chunk of several documents in one pass (bulk GC).

One terms delete_by_query rather than a query per document: the whole set costs a single round trip and a single forced index refresh, which is what makes the corpus table's multi-select delete (spec_v2 §4.2) cheap. The retry lives here, so the single-document wrapper inherits it exactly once.

Parameters:

Name Type Description Default
doc_ids Sequence[str]

The documents' stable ids (unknown ids are no-ops; an empty sequence skips the round trip entirely).

 required

Returns:

Type Description
int

The number of chunks deleted across all of them.

Raises:

Type Description
ConnectionError

If Elasticsearch is still unreachable after retries.
Source code in varagity/stores/bm25_store.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
@_es_retry
def delete_documents(self, doc_ids: Sequence[str]) -> int:
    """Delete every chunk of several documents in one pass (bulk GC).

    One ``terms`` ``delete_by_query`` rather than a query per document:
    the whole set costs a single round trip and a single forced index
    refresh, which is what makes the corpus table's multi-select delete
    (spec_v2 §4.2) cheap. The retry lives here, so the single-document
    wrapper inherits it exactly once.

    Args:
        doc_ids: The documents' stable ids (unknown ids are no-ops; an
            empty sequence skips the round trip entirely).

    Returns:
        The number of chunks deleted across all of them.

    Raises:
        elastic_transport.ConnectionError: If Elasticsearch is still
            unreachable after retries.
    """
    if not doc_ids:
        return 0
    response = self._client.delete_by_query(
        index=self.index_name,
        query={"terms": {"doc_id": list(doc_ids)}},
        refresh=True,
    )
    return int(response["deleted"])

index_chunks(records)

Bulk-index chunks, then refresh so they are searchable.

Documents are addressed by chunk_id, so re-indexing the same chunk overwrites instead of duplicating — the sparse-side counterpart of the vector store's ON CONFLICT (chunk_id) DO UPDATE.

Parameters:

Name Type Description Default
records list[ChunkRecord]

The chunk metadata records to index.

 required

Returns:

Type Description
int

The number of successfully indexed documents.

Raises:

Type Description
BulkIndexError

If any document is rejected (permanent — not retried).
ConnectionError

If Elasticsearch is still unreachable after retries.
Source code in varagity/stores/bm25_store.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
@_es_retry
def index_chunks(self, records: list[ChunkRecord]) -> int:
    """Bulk-index chunks, then refresh so they are searchable.

    Documents are addressed by ``chunk_id``, so re-indexing the same
    chunk overwrites instead of duplicating — the sparse-side counterpart
    of the vector store's ``ON CONFLICT (chunk_id) DO UPDATE``.

    Args:
        records: The chunk metadata records to index.

    Returns:
        The number of successfully indexed documents.

    Raises:
        elasticsearch.helpers.BulkIndexError: If any document is
            rejected (permanent — not retried).
        elastic_transport.ConnectionError: If Elasticsearch is still
            unreachable after retries.
    """
    if not records:
        return 0
    actions = [self._action(record) for record in records]
    success, _ = helpers.bulk(self._client, actions)
    self._client.indices.refresh(index=self.index_name)
    return success

rebuild(chunks, *, on_indexed=None)

Recreate the index from scratch and stream every chunk into it.

A backup restore's second half (ADR-019): the index is deleted and created anew with the spec mapping — so nothing of the previous corpus survives, and no auto-created default mapping can either — then the chunks stream in through helpers.streaming_bulk (bounded memory however large the corpus) and a final refresh makes them searchable. Index recreation is retried like every other call; a failure mid-stream raises, and running the rebuild again repairs the index.

Parameters:

Name Type Description Default
chunks Iterable[IndexableChunk]

Every chunk the index should hold.

 required
on_indexed Callable[[int], None] | None

Called with the running count after each indexed chunk (the restore's progress ticks).

 None

Returns:

Type Description
int

The number of chunks indexed.

Raises:

Type Description
BulkIndexError

If any chunk is rejected.
ConnectionError

If Elasticsearch is unreachable (after retries, for the index recreation).
Source code in varagity/stores/bm25_store.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def rebuild(
    self,
    chunks: Iterable[IndexableChunk],
    *,
    on_indexed: Callable[[int], None] | None = None,
) -> int:
    """Recreate the index from scratch and stream every chunk into it.

    A backup restore's second half (ADR-019): the index is deleted and
    created anew with the spec mapping — so nothing of the previous
    corpus survives, and no auto-created default mapping can either —
    then the chunks stream in through ``helpers.streaming_bulk`` (bounded
    memory however large the corpus) and a final refresh makes them
    searchable. Index recreation is retried like every other call; a
    failure mid-stream raises, and running the rebuild again repairs the
    index.

    Args:
        chunks: Every chunk the index should hold.
        on_indexed: Called with the running count after each indexed
            chunk (the restore's progress ticks).

    Returns:
        The number of chunks indexed.

    Raises:
        elasticsearch.helpers.BulkIndexError: If any chunk is rejected.
        elastic_transport.ConnectionError: If Elasticsearch is
            unreachable (after retries, for the index recreation).
    """
    self._recreate_index()
    indexed = 0
    for ok, _ in helpers.streaming_bulk(
        self._client,
        (self._action(chunk) for chunk in chunks),
        chunk_size=_REBUILD_CHUNK_SIZE,
        max_retries=_REBUILD_MAX_RETRIES,
    ):
        if ok:
            indexed += 1
            if on_indexed is not None:
                on_indexed(indexed)
    self._refresh()
    logger.info("rebuilt BM25 index %r with %d chunk(s)", self.index_name, indexed)
    return indexed

search(query, k, verbose=None)

BM25 top-k search via multi_match over both text fields.

Parameters:

Name Type Description Default
query str

The user's query, unformatted (Elasticsearch's english analyzer does its own normalization).

 required
k int

Number of hits to return.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
list[BM25Hit]

The top-k hits, best first.

Raises:

Type Description
ValueError

If verbose is invalid.
ConnectionError

If Elasticsearch is still unreachable after retries.
Source code in varagity/stores/bm25_store.py 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
@_es_retry
def search(self, query: str, k: int, verbose: int | None = None) -> list[BM25Hit]:
    """BM25 top-k search via ``multi_match`` over both text fields.

    Args:
        query: The user's query, unformatted (Elasticsearch's ``english``
            analyzer does its own normalization).
        k: Number of hits to return.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The top-k hits, best first.

    Raises:
        ValueError: If ``verbose`` is invalid.
        elastic_transport.ConnectionError: If Elasticsearch is still
            unreachable after retries.
    """
    check_verbose(get_settings().DEFAULT_VERBOSE if verbose is None else verbose)
    response = self._client.search(
        index=self.index_name,
        query={
            "multi_match": {
                "query": query,
                "fields": ["content", "contextualized_content"],
            }
        },
        size=k,
    )
    return [
        BM25Hit(
            doc_id=hit["_source"]["doc_id"],
            original_index=hit["_source"]["original_index"],
            content=hit["_source"]["content"],
            contextualized_content=hit["_source"]["contextualized_content"],
            score=hit["_score"],
        )
        for hit in response["hits"]["hits"]
    ]

IndexableChunk

Bases: Protocol

What the index needs of one chunk: the five fields it mirrors.

Satisfied by :class:~varagity.stores.records.ChunkRecord (ingest) and by the vector store's CorpusIndexRow (a restore's rebuild).

Source code in varagity/stores/bm25_store.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class IndexableChunk(Protocol):
    """What the index needs of one chunk: the five fields it mirrors.

    Satisfied by :class:`~varagity.stores.records.ChunkRecord` (ingest) and
    by the vector store's ``CorpusIndexRow`` (a restore's rebuild).
    """

    @property
    def chunk_id(self) -> str:
        """The chunk's id — the document's ``_id``."""
        ...

    @property
    def doc_id(self) -> str:
        """The owning document's id."""
        ...

    @property
    def original_index(self) -> int:
        """The corpus-global index (the fusion join key)."""
        ...

    @property
    def content(self) -> str:
        """The original chunk text."""
        ...

    @property
    def contextualized_content(self) -> str:
        """The contextualized text that is BM25-indexed."""
        ...

chunk_id property

The chunk's id — the document's _id.

content property

The original chunk text.

contextualized_content property

The contextualized text that is BM25-indexed.

doc_id property

The owning document's id.

original_index property

The corpus-global index (the fusion join key).

varagity.stores.conversation_store

Conversation persistence over PostgreSQL (spec_v2 §4.4, §9.1).

Single-user chat history lives in the same Postgres as the chunks (one datastore to operate) but in independent tables: an assistant turn's evidence is snapshotted into message_sources.trace — content, context, source provenance, and the :class:~varagity.stores.records.RetrievalTrace — with chunk_id kept only as a soft reference, so a historical conversation still explains itself after a reingest replaces the chunk rows. A graph-targeted turn (spec_graphrag §4.2) rides the same table under a kind discriminator: its cited transcript days are snapshotted with chunk_id holding the transcript doc_key, and the same reasoning applies — a graph rebuild re-derives keys, and the answer must still explain itself afterwards.

The tables are created by the migration runner (:mod:varagity.stores.migrate), not schema.sql.

DEFAULT_TITLE = 'New conversation' module-attribute

Title given at creation; :meth:ConversationStore.auto_title replaces it.

ConversationDetail

Bases: BaseModel

A full transcript: the conversation row plus its messages.

Attributes:

Name Type Description
conversation_id str

The app-generated id.
title str

Current title.
created_at datetime

Creation timestamp.
updated_at datetime

Last-turn timestamp.
messages list[MessageRecord]

All messages, oldest first, each with its sources.
Source code in varagity/stores/conversation_store.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class ConversationDetail(BaseModel):
    """A full transcript: the conversation row plus its messages.

    Attributes:
        conversation_id: The app-generated id.
        title: Current title.
        created_at: Creation timestamp.
        updated_at: Last-turn timestamp.
        messages: All messages, oldest first, each with its sources.
    """

    conversation_id: str
    title: str
    created_at: datetime
    updated_at: datetime
    messages: list[MessageRecord]

ConversationGroup

Bases: BaseModel

One sidebar conversation group (a user-created folder).

Attributes:

Name Type Description
group_id str

The app-generated id.
name str

Display name (not unique — ids are the identity).
created_at datetime

Creation timestamp.
Source code in varagity/stores/conversation_store.py 
113
114
115
116
117
118
119
120
121
122
123
124
class ConversationGroup(BaseModel):
    """One sidebar conversation group (a user-created folder).

    Attributes:
        group_id: The app-generated id.
        name: Display name (not unique — ids are the identity).
        created_at: Creation timestamp.
    """

    group_id: str
    name: str
    created_at: datetime

ConversationStore

Bases: ClosingContextMixin

Conversation CRUD over PostgreSQL.

Owns one autocommit connection (per-turn writes group into explicit transactions), mirroring :class:~varagity.stores.vector_store.ContextualVectorDB. Use as a context manager or call :meth:close when done.

Source code in varagity/stores/conversation_store.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
class ConversationStore(ClosingContextMixin):
    """Conversation CRUD over PostgreSQL.

    Owns one autocommit connection (per-turn writes group into explicit
    transactions), mirroring
    :class:`~varagity.stores.vector_store.ContextualVectorDB`. Use as a
    context manager or call :meth:`close` when done.
    """

    def __init__(self, conninfo: str | None = None) -> None:
        """Connect to the configured PostgreSQL instance.

        Args:
            conninfo: libpq connection string; defaults to the
                ``POSTGRES_*`` settings.

        Raises:
            psycopg.OperationalError: If the database is unreachable.
        """
        self._conn = psycopg.connect(conninfo or default_conninfo(), autocommit=True)

    def close(self) -> None:
        """Close the underlying connection (idempotent)."""
        if not self._conn.closed:
            self._conn.close()

    def create_conversation(self, title: str | None = None) -> ConversationSummary:
        """Insert a new conversation.

        Args:
            title: Initial title; defaults to :data:`DEFAULT_TITLE` (the
                first chat turn replaces it via :meth:`auto_title`).

        Returns:
            The created conversation's summary (``message_count = 0``).
        """
        conversation_id = uuid.uuid4().hex
        row = self._conn.execute(
            "INSERT INTO conversations (conversation_id, title) VALUES (%s, %s) "
            "RETURNING created_at, updated_at",
            (conversation_id, title or DEFAULT_TITLE),
        ).fetchone()
        assert row is not None  # RETURNING on a successful INSERT
        return ConversationSummary(
            conversation_id=conversation_id,
            title=title or DEFAULT_TITLE,
            created_at=row[0],
            updated_at=row[1],
            message_count=0,
        )

    def conversation_exists(self, conversation_id: str) -> bool:
        """Check whether a conversation id is known.

        Args:
            conversation_id: The id to look up.

        Returns:
            ``True`` if a matching row exists.
        """
        row = self._conn.execute(
            "SELECT 1 FROM conversations WHERE conversation_id = %s", (conversation_id,)
        ).fetchone()
        return row is not None

    def list_conversations(self) -> list[ConversationSummary]:
        """List every conversation, most recently updated first.

        Returns:
            Summaries with message counts (spec_v2 §4.2).
        """
        rows = self._conn.execute(_LIST_CONVERSATIONS_SQL).fetchall()
        return [
            ConversationSummary(
                conversation_id=row[0],
                title=row[1],
                created_at=row[2],
                updated_at=row[3],
                group_id=row[4],
                message_count=int(row[5]),
            )
            for row in rows
        ]

    def get_conversation(self, conversation_id: str) -> ConversationDetail | None:
        """Fetch a full transcript: messages plus snapshotted sources.

        Args:
            conversation_id: The conversation to fetch.

        Returns:
            The transcript, or ``None`` for an unknown id.
        """
        conversation_row = self._conn.execute(
            "SELECT title, created_at, updated_at FROM conversations WHERE conversation_id = %s",
            (conversation_id,),
        ).fetchone()
        if conversation_row is None:
            return None
        sources_by_message: dict[str, list[StoredSource]] = {}
        for message_id, rank, chunk_id, trace in self._conn.execute(
            _SELECT_SOURCES_SQL, (conversation_id,)
        ):
            sources_by_message.setdefault(message_id, []).append(
                StoredSource(rank=rank, chunk_id=chunk_id, trace=trace)
            )
        messages = [
            MessageRecord(
                message_id=row[0],
                role=row[1],
                content=row[2],
                created_at=row[3],
                retrieval_method=row[4],
                latency_ms=row[5],
                reasoning=row[6],
                condensed_query=row[7],
                chat_engine=row[8],
                target_corpus=row[9],
                graph_evidence=row[10],
                sources=sources_by_message.get(row[0], []),
            )
            for row in self._conn.execute(_SELECT_MESSAGES_SQL, (conversation_id,))
        ]
        return ConversationDetail(
            conversation_id=conversation_id,
            title=conversation_row[0],
            created_at=conversation_row[1],
            updated_at=conversation_row[2],
            messages=messages,
        )

    def recent_turns(self, conversation_id: str, limit: int) -> list[tuple[str, str]]:
        """Fetch a conversation's newest turns, re-ordered oldest first.

        The chat engine's history feed (spec_v3 §4.4) — deliberately not
        :meth:`get_conversation`, which hydrates every message's snapshotted
        sources and traces: evidence the condenser must not see, on the
        pre-first-token hot path, growing with conversation length. Here the
        bound is applied **in SQL** (newest ``limit`` rows) and only
        ``role``/``content`` are read. Primitive pairs keep
        ``varagity/stores/`` from importing ``varagity/chat/`` — callers map
        them to their own turn type.

        Args:
            conversation_id: The conversation to read (an unknown id yields
                the empty list — existence is the caller's check).
            limit: Maximum turns returned; non-positive yields the empty
                list without touching the database.

        Returns:
            Up to ``limit`` newest ``(role, content)`` pairs, oldest first.
        """
        if limit <= 0:
            return []
        rows = self._conn.execute(_RECENT_TURNS_SQL, (conversation_id, limit)).fetchall()
        return [(row[0], row[1]) for row in reversed(rows)]

    def delete_conversation(self, conversation_id: str) -> int:
        """Delete a conversation; messages and sources cascade (spec_v2 §9.1).

        Args:
            conversation_id: The conversation to delete (unknown id is a
                no-op).

        Returns:
            The number of ``conversations`` rows deleted (0 or 1).
        """
        cursor = self._conn.execute(
            "DELETE FROM conversations WHERE conversation_id = %s", (conversation_id,)
        )
        return cursor.rowcount

    def list_groups(self) -> list[ConversationGroup]:
        """List every conversation group, name order (case-folded).

        Returns:
            All groups, including empty ones — an empty folder is still a
            drop target in the sidebar.
        """
        rows = self._conn.execute(_LIST_GROUPS_SQL).fetchall()
        return [ConversationGroup(group_id=row[0], name=row[1], created_at=row[2]) for row in rows]

    def create_group(self, name: str) -> ConversationGroup:
        """Insert a new conversation group.

        Args:
            name: Display name (uniqueness is not enforced — ids are the
                identity).

        Returns:
            The created group.
        """
        group_id = uuid.uuid4().hex
        row = self._conn.execute(
            "INSERT INTO conversation_groups (group_id, name) VALUES (%s, %s) RETURNING created_at",
            (group_id, name),
        ).fetchone()
        assert row is not None  # RETURNING on a successful INSERT
        return ConversationGroup(group_id=group_id, name=name, created_at=row[0])

    def group_exists(self, group_id: str) -> bool:
        """Check whether a group id is known.

        Args:
            group_id: The id to look up.

        Returns:
            ``True`` if a matching row exists.
        """
        row = self._conn.execute(
            "SELECT 1 FROM conversation_groups WHERE group_id = %s", (group_id,)
        ).fetchone()
        return row is not None

    def delete_group(self, group_id: str) -> int:
        """Delete a group; its conversations detach (``ON DELETE SET NULL``).

        Args:
            group_id: The group to delete (unknown id is a no-op).

        Returns:
            The number of ``conversation_groups`` rows deleted (0 or 1).
        """
        cursor = self._conn.execute(
            "DELETE FROM conversation_groups WHERE group_id = %s", (group_id,)
        )
        return cursor.rowcount

    def set_conversation_group(self, conversation_id: str, group_id: str | None) -> int:
        """File a conversation under a group, or ungroup it.

        Deliberately no ``updated_at`` bump: moving a conversation must not
        re-order the recency-sorted sidebar list.

        Args:
            conversation_id: The conversation to move (unknown id is a
                no-op).
            group_id: Target group id, or ``None`` to ungroup.

        Returns:
            The number of ``conversations`` rows updated (0 or 1).

        Raises:
            psycopg.errors.ForeignKeyViolation: If ``group_id`` is unknown
                (callers validate first; the FK is the backstop).
        """
        cursor = self._conn.execute(
            "UPDATE conversations SET group_id = %s WHERE conversation_id = %s",
            (group_id, conversation_id),
        )
        return cursor.rowcount

    def append_message(
        self,
        conversation_id: str,
        role: str,
        content: str,
        *,
        retrieval_method: str | None = None,
        latency_ms: dict[str, Any] | None = None,
        reasoning: str | None = None,
        condensed_query: str | None = None,
        chat_engine: str | None = None,
        sources: Sequence[RetrievedChunk] = (),
        target_corpus: str | None = None,
        graph_evidence: dict[str, Any] | None = None,
        graph_sources: Sequence[GraphSourceSnapshot] = (),
    ) -> str:
        """Persist one turn, snapshotting an assistant turn's evidence.

        One transaction covers the message, its ``message_sources`` rows,
        and the conversation's ``updated_at`` bump — a partial failure
        leaves no dangling turn.

        Args:
            conversation_id: The conversation the turn belongs to.
            role: ``"user"`` or ``"assistant"``.
            content: The question or the generated answer.
            retrieval_method: Retrieval method used (assistant turns) — the
                registry name for a chunk turn, the engine query mode for a
                graph one (``target_corpus`` says which vocabulary applies).
            latency_ms: Per-stage timings (assistant turns).
            reasoning: Captured ``<think>`` stream, if any.
            condensed_query: The standalone search query that drove the
                turn's retrieval (assistant turns; ``None`` = the user's
                words were searched verbatim — spec_v3 §8).
            chat_engine: Registry name of the chat engine that produced
                the turn (assistant turns).
            sources: The answer's retrieved chunks, best first; each is
                snapshotted via the spec_v2 §9.1 trace blob. Empty on a
                graph turn, which cites transcript days instead.
            target_corpus: Which corpus the turn asked for (``"rag"`` |
                ``"graph"``), recorded even when a graph turn degraded to a
                chunk answer (ADR-017's honest record).
            graph_evidence: The graph retrieval's entities/relations and
                mode; ``None`` for chunk turns *and* for degraded graph
                turns — there was no graph retrieval to record.
            graph_sources: The cited transcript days, best first; each
                becomes a ``message_sources`` row keyed by its ``doc_key``,
                continuing after ``sources`` in rank order (in practice one
                list or the other is empty).

        Returns:
            The new message's id.

        Raises:
            psycopg.errors.ForeignKeyViolation: If ``conversation_id`` is
                unknown (callers validate first; the FK is the backstop).
        """
        message_id = uuid.uuid4().hex
        with self._conn.transaction():
            self._conn.execute(
                "INSERT INTO messages (message_id, conversation_id, role, content, "
                "retrieval_method, latency_ms, reasoning, condensed_query, chat_engine, "
                "target_corpus, graph_evidence) "
                "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
                (
                    message_id,
                    conversation_id,
                    role,
                    content,
                    retrieval_method,
                    None if latency_ms is None else Json(latency_ms),
                    reasoning,
                    condensed_query,
                    chat_engine,
                    target_corpus,
                    None if graph_evidence is None else Json(graph_evidence),
                ),
            )
            rank = 0
            for rank, chunk in enumerate(sources, start=1):
                self._conn.execute(
                    "INSERT INTO message_sources (message_id, rank, chunk_id, trace) "
                    "VALUES (%s, %s, %s, %s)",
                    (message_id, rank, chunk.chunk_id, Json(_source_snapshot(chunk))),
                )
            for offset, source in enumerate(graph_sources, start=1):
                self._conn.execute(
                    "INSERT INTO message_sources (message_id, rank, chunk_id, trace) "
                    "VALUES (%s, %s, %s, %s)",
                    (
                        message_id,
                        rank + offset,
                        source.doc_key,
                        Json(_graph_source_snapshot(source)),
                    ),
                )
            self._conn.execute(
                "UPDATE conversations SET updated_at = now() WHERE conversation_id = %s",
                (conversation_id,),
            )
        return message_id

    def auto_title(
        self, conversation_id: str, question: str, *, llm: LLMClient | None = None
    ) -> str | None:
        """Title a still-default-titled conversation from its first question.

        A cheap LLM one-liner (spec_v2 §4.2); on any generation failure the
        title falls back to the truncated question — titling must never
        break a chat turn. Conversations already titled (auto or manually)
        are left alone.

        Args:
            conversation_id: The conversation to title.
            question: The first user question.
            llm: Chat client; resolved via the model registry when omitted.

        Returns:
            The new title, or ``None`` when the conversation is unknown or
            no longer carries :data:`DEFAULT_TITLE`.
        """
        row = self._conn.execute(
            "SELECT title FROM conversations WHERE conversation_id = %s", (conversation_id,)
        ).fetchone()
        if row is None or row[0] != DEFAULT_TITLE:
            return None
        title = _generate_title(question, llm=llm)
        self._conn.execute(
            "UPDATE conversations SET title = %s WHERE conversation_id = %s",
            (title, conversation_id),
        )
        return title

__init__(conninfo=None)

Connect to the configured PostgreSQL instance.

Parameters:

Name Type Description Default
conninfo str | None

libpq connection string; defaults to the POSTGRES_* settings.

 None

Raises:

Type Description
OperationalError

If the database is unreachable.
Source code in varagity/stores/conversation_store.py 
271
272
273
274
275
276
277
278
279
280
281
def __init__(self, conninfo: str | None = None) -> None:
    """Connect to the configured PostgreSQL instance.

    Args:
        conninfo: libpq connection string; defaults to the
            ``POSTGRES_*`` settings.

    Raises:
        psycopg.OperationalError: If the database is unreachable.
    """
    self._conn = psycopg.connect(conninfo or default_conninfo(), autocommit=True)

append_message(conversation_id, role, content, *, retrieval_method=None, latency_ms=None, reasoning=None, condensed_query=None, chat_engine=None, sources=(), target_corpus=None, graph_evidence=None, graph_sources=())

Persist one turn, snapshotting an assistant turn's evidence.

One transaction covers the message, its message_sources rows, and the conversation's updated_at bump — a partial failure leaves no dangling turn.

Parameters:

Name Type Description Default
conversation_id str

The conversation the turn belongs to.

 required
role str

"user" or "assistant".

 required
content str

The question or the generated answer.

 required
retrieval_method str | None

Retrieval method used (assistant turns) — the registry name for a chunk turn, the engine query mode for a graph one (target_corpus says which vocabulary applies).

 None
latency_ms dict[str, Any] | None

Per-stage timings (assistant turns).

 None
reasoning str | None

Captured <think> stream, if any.

 None
condensed_query str | None

The standalone search query that drove the turn's retrieval (assistant turns; None = the user's words were searched verbatim — spec_v3 §8).

 None
chat_engine str | None

Registry name of the chat engine that produced the turn (assistant turns).

 None
sources Sequence[RetrievedChunk]

The answer's retrieved chunks, best first; each is snapshotted via the spec_v2 §9.1 trace blob. Empty on a graph turn, which cites transcript days instead.

 ()
target_corpus str | None

Which corpus the turn asked for ("rag" | "graph"), recorded even when a graph turn degraded to a chunk answer (ADR-017's honest record).

 None
graph_evidence dict[str, Any] | None

The graph retrieval's entities/relations and mode; None for chunk turns and for degraded graph turns — there was no graph retrieval to record.

 None
graph_sources Sequence[GraphSourceSnapshot]

The cited transcript days, best first; each becomes a message_sources row keyed by its doc_key, continuing after sources in rank order (in practice one list or the other is empty).

 ()

Returns:

Type Description
str

The new message's id.

Raises:

Type Description
ForeignKeyViolation

If conversation_id is unknown (callers validate first; the FK is the backstop).
Source code in varagity/stores/conversation_store.py 
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
def append_message(
    self,
    conversation_id: str,
    role: str,
    content: str,
    *,
    retrieval_method: str | None = None,
    latency_ms: dict[str, Any] | None = None,
    reasoning: str | None = None,
    condensed_query: str | None = None,
    chat_engine: str | None = None,
    sources: Sequence[RetrievedChunk] = (),
    target_corpus: str | None = None,
    graph_evidence: dict[str, Any] | None = None,
    graph_sources: Sequence[GraphSourceSnapshot] = (),
) -> str:
    """Persist one turn, snapshotting an assistant turn's evidence.

    One transaction covers the message, its ``message_sources`` rows,
    and the conversation's ``updated_at`` bump — a partial failure
    leaves no dangling turn.

    Args:
        conversation_id: The conversation the turn belongs to.
        role: ``"user"`` or ``"assistant"``.
        content: The question or the generated answer.
        retrieval_method: Retrieval method used (assistant turns) — the
            registry name for a chunk turn, the engine query mode for a
            graph one (``target_corpus`` says which vocabulary applies).
        latency_ms: Per-stage timings (assistant turns).
        reasoning: Captured ``<think>`` stream, if any.
        condensed_query: The standalone search query that drove the
            turn's retrieval (assistant turns; ``None`` = the user's
            words were searched verbatim — spec_v3 §8).
        chat_engine: Registry name of the chat engine that produced
            the turn (assistant turns).
        sources: The answer's retrieved chunks, best first; each is
            snapshotted via the spec_v2 §9.1 trace blob. Empty on a
            graph turn, which cites transcript days instead.
        target_corpus: Which corpus the turn asked for (``"rag"`` |
            ``"graph"``), recorded even when a graph turn degraded to a
            chunk answer (ADR-017's honest record).
        graph_evidence: The graph retrieval's entities/relations and
            mode; ``None`` for chunk turns *and* for degraded graph
            turns — there was no graph retrieval to record.
        graph_sources: The cited transcript days, best first; each
            becomes a ``message_sources`` row keyed by its ``doc_key``,
            continuing after ``sources`` in rank order (in practice one
            list or the other is empty).

    Returns:
        The new message's id.

    Raises:
        psycopg.errors.ForeignKeyViolation: If ``conversation_id`` is
            unknown (callers validate first; the FK is the backstop).
    """
    message_id = uuid.uuid4().hex
    with self._conn.transaction():
        self._conn.execute(
            "INSERT INTO messages (message_id, conversation_id, role, content, "
            "retrieval_method, latency_ms, reasoning, condensed_query, chat_engine, "
            "target_corpus, graph_evidence) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
            (
                message_id,
                conversation_id,
                role,
                content,
                retrieval_method,
                None if latency_ms is None else Json(latency_ms),
                reasoning,
                condensed_query,
                chat_engine,
                target_corpus,
                None if graph_evidence is None else Json(graph_evidence),
            ),
        )
        rank = 0
        for rank, chunk in enumerate(sources, start=1):
            self._conn.execute(
                "INSERT INTO message_sources (message_id, rank, chunk_id, trace) "
                "VALUES (%s, %s, %s, %s)",
                (message_id, rank, chunk.chunk_id, Json(_source_snapshot(chunk))),
            )
        for offset, source in enumerate(graph_sources, start=1):
            self._conn.execute(
                "INSERT INTO message_sources (message_id, rank, chunk_id, trace) "
                "VALUES (%s, %s, %s, %s)",
                (
                    message_id,
                    rank + offset,
                    source.doc_key,
                    Json(_graph_source_snapshot(source)),
                ),
            )
        self._conn.execute(
            "UPDATE conversations SET updated_at = now() WHERE conversation_id = %s",
            (conversation_id,),
        )
    return message_id

auto_title(conversation_id, question, *, llm=None)

Title a still-default-titled conversation from its first question.

A cheap LLM one-liner (spec_v2 §4.2); on any generation failure the title falls back to the truncated question — titling must never break a chat turn. Conversations already titled (auto or manually) are left alone.

Parameters:

Name Type Description Default
conversation_id str

The conversation to title.

 required
question str

The first user question.

 required
llm LLMClient | None

Chat client; resolved via the model registry when omitted.

 None

Returns:

Type Description
str | None

The new title, or None when the conversation is unknown or
str | None

no longer carries :data:DEFAULT_TITLE.
Source code in varagity/stores/conversation_store.py 
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
def auto_title(
    self, conversation_id: str, question: str, *, llm: LLMClient | None = None
) -> str | None:
    """Title a still-default-titled conversation from its first question.

    A cheap LLM one-liner (spec_v2 §4.2); on any generation failure the
    title falls back to the truncated question — titling must never
    break a chat turn. Conversations already titled (auto or manually)
    are left alone.

    Args:
        conversation_id: The conversation to title.
        question: The first user question.
        llm: Chat client; resolved via the model registry when omitted.

    Returns:
        The new title, or ``None`` when the conversation is unknown or
        no longer carries :data:`DEFAULT_TITLE`.
    """
    row = self._conn.execute(
        "SELECT title FROM conversations WHERE conversation_id = %s", (conversation_id,)
    ).fetchone()
    if row is None or row[0] != DEFAULT_TITLE:
        return None
    title = _generate_title(question, llm=llm)
    self._conn.execute(
        "UPDATE conversations SET title = %s WHERE conversation_id = %s",
        (title, conversation_id),
    )
    return title

close()

Close the underlying connection (idempotent).

Source code in varagity/stores/conversation_store.py 
283
284
285
286
def close(self) -> None:
    """Close the underlying connection (idempotent)."""
    if not self._conn.closed:
        self._conn.close()

conversation_exists(conversation_id)

Check whether a conversation id is known.

Parameters:

Name Type Description Default
conversation_id str

The id to look up.

 required

Returns:

Type Description
bool

True if a matching row exists.
Source code in varagity/stores/conversation_store.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
def conversation_exists(self, conversation_id: str) -> bool:
    """Check whether a conversation id is known.

    Args:
        conversation_id: The id to look up.

    Returns:
        ``True`` if a matching row exists.
    """
    row = self._conn.execute(
        "SELECT 1 FROM conversations WHERE conversation_id = %s", (conversation_id,)
    ).fetchone()
    return row is not None

create_conversation(title=None)

Insert a new conversation.

Parameters:

Name Type Description Default
title str | None

Initial title; defaults to :data:DEFAULT_TITLE (the first chat turn replaces it via :meth:auto_title).

 None

Returns:

Type Description
ConversationSummary

The created conversation's summary (message_count = 0).
Source code in varagity/stores/conversation_store.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def create_conversation(self, title: str | None = None) -> ConversationSummary:
    """Insert a new conversation.

    Args:
        title: Initial title; defaults to :data:`DEFAULT_TITLE` (the
            first chat turn replaces it via :meth:`auto_title`).

    Returns:
        The created conversation's summary (``message_count = 0``).
    """
    conversation_id = uuid.uuid4().hex
    row = self._conn.execute(
        "INSERT INTO conversations (conversation_id, title) VALUES (%s, %s) "
        "RETURNING created_at, updated_at",
        (conversation_id, title or DEFAULT_TITLE),
    ).fetchone()
    assert row is not None  # RETURNING on a successful INSERT
    return ConversationSummary(
        conversation_id=conversation_id,
        title=title or DEFAULT_TITLE,
        created_at=row[0],
        updated_at=row[1],
        message_count=0,
    )

create_group(name)

Insert a new conversation group.

Parameters:

Name Type Description Default
name str

Display name (uniqueness is not enforced — ids are the identity).

 required

Returns:

Type Description
ConversationGroup

The created group.
Source code in varagity/stores/conversation_store.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
def create_group(self, name: str) -> ConversationGroup:
    """Insert a new conversation group.

    Args:
        name: Display name (uniqueness is not enforced — ids are the
            identity).

    Returns:
        The created group.
    """
    group_id = uuid.uuid4().hex
    row = self._conn.execute(
        "INSERT INTO conversation_groups (group_id, name) VALUES (%s, %s) RETURNING created_at",
        (group_id, name),
    ).fetchone()
    assert row is not None  # RETURNING on a successful INSERT
    return ConversationGroup(group_id=group_id, name=name, created_at=row[0])

delete_conversation(conversation_id)

Delete a conversation; messages and sources cascade (spec_v2 §9.1).

Parameters:

Name Type Description Default
conversation_id str

The conversation to delete (unknown id is a no-op).

 required

Returns:

Type Description
int

The number of conversations rows deleted (0 or 1).
Source code in varagity/stores/conversation_store.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def delete_conversation(self, conversation_id: str) -> int:
    """Delete a conversation; messages and sources cascade (spec_v2 §9.1).

    Args:
        conversation_id: The conversation to delete (unknown id is a
            no-op).

    Returns:
        The number of ``conversations`` rows deleted (0 or 1).
    """
    cursor = self._conn.execute(
        "DELETE FROM conversations WHERE conversation_id = %s", (conversation_id,)
    )
    return cursor.rowcount

delete_group(group_id)

Delete a group; its conversations detach (ON DELETE SET NULL).

Parameters:

Name Type Description Default
group_id str

The group to delete (unknown id is a no-op).

 required

Returns:

Type Description
int

The number of conversation_groups rows deleted (0 or 1).
Source code in varagity/stores/conversation_store.py 
476
477
478
479
480
481
482
483
484
485
486
487
488
def delete_group(self, group_id: str) -> int:
    """Delete a group; its conversations detach (``ON DELETE SET NULL``).

    Args:
        group_id: The group to delete (unknown id is a no-op).

    Returns:
        The number of ``conversation_groups`` rows deleted (0 or 1).
    """
    cursor = self._conn.execute(
        "DELETE FROM conversation_groups WHERE group_id = %s", (group_id,)
    )
    return cursor.rowcount

get_conversation(conversation_id)

Fetch a full transcript: messages plus snapshotted sources.

Parameters:

Name Type Description Default
conversation_id str

The conversation to fetch.

 required

Returns:

Type Description
ConversationDetail | None

The transcript, or None for an unknown id.
Source code in varagity/stores/conversation_store.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
def get_conversation(self, conversation_id: str) -> ConversationDetail | None:
    """Fetch a full transcript: messages plus snapshotted sources.

    Args:
        conversation_id: The conversation to fetch.

    Returns:
        The transcript, or ``None`` for an unknown id.
    """
    conversation_row = self._conn.execute(
        "SELECT title, created_at, updated_at FROM conversations WHERE conversation_id = %s",
        (conversation_id,),
    ).fetchone()
    if conversation_row is None:
        return None
    sources_by_message: dict[str, list[StoredSource]] = {}
    for message_id, rank, chunk_id, trace in self._conn.execute(
        _SELECT_SOURCES_SQL, (conversation_id,)
    ):
        sources_by_message.setdefault(message_id, []).append(
            StoredSource(rank=rank, chunk_id=chunk_id, trace=trace)
        )
    messages = [
        MessageRecord(
            message_id=row[0],
            role=row[1],
            content=row[2],
            created_at=row[3],
            retrieval_method=row[4],
            latency_ms=row[5],
            reasoning=row[6],
            condensed_query=row[7],
            chat_engine=row[8],
            target_corpus=row[9],
            graph_evidence=row[10],
            sources=sources_by_message.get(row[0], []),
        )
        for row in self._conn.execute(_SELECT_MESSAGES_SQL, (conversation_id,))
    ]
    return ConversationDetail(
        conversation_id=conversation_id,
        title=conversation_row[0],
        created_at=conversation_row[1],
        updated_at=conversation_row[2],
        messages=messages,
    )

group_exists(group_id)

Check whether a group id is known.

Parameters:

Name Type Description Default
group_id str

The id to look up.

 required

Returns:

Type Description
bool

True if a matching row exists.
Source code in varagity/stores/conversation_store.py 
462
463
464
465
466
467
468
469
470
471
472
473
474
def group_exists(self, group_id: str) -> bool:
    """Check whether a group id is known.

    Args:
        group_id: The id to look up.

    Returns:
        ``True`` if a matching row exists.
    """
    row = self._conn.execute(
        "SELECT 1 FROM conversation_groups WHERE group_id = %s", (group_id,)
    ).fetchone()
    return row is not None

list_conversations()

List every conversation, most recently updated first.

Returns:

Type Description
list[ConversationSummary]

Summaries with message counts (spec_v2 §4.2).
Source code in varagity/stores/conversation_store.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def list_conversations(self) -> list[ConversationSummary]:
    """List every conversation, most recently updated first.

    Returns:
        Summaries with message counts (spec_v2 §4.2).
    """
    rows = self._conn.execute(_LIST_CONVERSATIONS_SQL).fetchall()
    return [
        ConversationSummary(
            conversation_id=row[0],
            title=row[1],
            created_at=row[2],
            updated_at=row[3],
            group_id=row[4],
            message_count=int(row[5]),
        )
        for row in rows
    ]

list_groups()

List every conversation group, name order (case-folded).

Returns:

Type Description
list[ConversationGroup]

All groups, including empty ones — an empty folder is still a
list[ConversationGroup]

drop target in the sidebar.
Source code in varagity/stores/conversation_store.py 
434
435
436
437
438
439
440
441
442
def list_groups(self) -> list[ConversationGroup]:
    """List every conversation group, name order (case-folded).

    Returns:
        All groups, including empty ones — an empty folder is still a
        drop target in the sidebar.
    """
    rows = self._conn.execute(_LIST_GROUPS_SQL).fetchall()
    return [ConversationGroup(group_id=row[0], name=row[1], created_at=row[2]) for row in rows]

recent_turns(conversation_id, limit)

Fetch a conversation's newest turns, re-ordered oldest first.

The chat engine's history feed (spec_v3 §4.4) — deliberately not :meth:get_conversation, which hydrates every message's snapshotted sources and traces: evidence the condenser must not see, on the pre-first-token hot path, growing with conversation length. Here the bound is applied in SQL (newest limit rows) and only role/content are read. Primitive pairs keep varagity/stores/ from importing varagity/chat/ — callers map them to their own turn type.

Parameters:

Name Type Description Default
conversation_id str

The conversation to read (an unknown id yields the empty list — existence is the caller's check).

 required
limit int

Maximum turns returned; non-positive yields the empty list without touching the database.

 required

Returns:

Type Description
list[tuple[str, str]]

Up to limit newest (role, content) pairs, oldest first.
Source code in varagity/stores/conversation_store.py 
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
def recent_turns(self, conversation_id: str, limit: int) -> list[tuple[str, str]]:
    """Fetch a conversation's newest turns, re-ordered oldest first.

    The chat engine's history feed (spec_v3 §4.4) — deliberately not
    :meth:`get_conversation`, which hydrates every message's snapshotted
    sources and traces: evidence the condenser must not see, on the
    pre-first-token hot path, growing with conversation length. Here the
    bound is applied **in SQL** (newest ``limit`` rows) and only
    ``role``/``content`` are read. Primitive pairs keep
    ``varagity/stores/`` from importing ``varagity/chat/`` — callers map
    them to their own turn type.

    Args:
        conversation_id: The conversation to read (an unknown id yields
            the empty list — existence is the caller's check).
        limit: Maximum turns returned; non-positive yields the empty
            list without touching the database.

    Returns:
        Up to ``limit`` newest ``(role, content)`` pairs, oldest first.
    """
    if limit <= 0:
        return []
    rows = self._conn.execute(_RECENT_TURNS_SQL, (conversation_id, limit)).fetchall()
    return [(row[0], row[1]) for row in reversed(rows)]

set_conversation_group(conversation_id, group_id)

File a conversation under a group, or ungroup it.

Deliberately no updated_at bump: moving a conversation must not re-order the recency-sorted sidebar list.

Parameters:

Name Type Description Default
conversation_id str

The conversation to move (unknown id is a no-op).

 required
group_id str | None

Target group id, or None to ungroup.

 required

Returns:

Type Description
int

The number of conversations rows updated (0 or 1).

Raises:

Type Description
ForeignKeyViolation

If group_id is unknown (callers validate first; the FK is the backstop).
Source code in varagity/stores/conversation_store.py 
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
def set_conversation_group(self, conversation_id: str, group_id: str | None) -> int:
    """File a conversation under a group, or ungroup it.

    Deliberately no ``updated_at`` bump: moving a conversation must not
    re-order the recency-sorted sidebar list.

    Args:
        conversation_id: The conversation to move (unknown id is a
            no-op).
        group_id: Target group id, or ``None`` to ungroup.

    Returns:
        The number of ``conversations`` rows updated (0 or 1).

    Raises:
        psycopg.errors.ForeignKeyViolation: If ``group_id`` is unknown
            (callers validate first; the FK is the backstop).
    """
    cursor = self._conn.execute(
        "UPDATE conversations SET group_id = %s WHERE conversation_id = %s",
        (group_id, conversation_id),
    )
    return cursor.rowcount

ConversationSummary

Bases: BaseModel

One conversation as listed in the sidebar (spec_v2 §4.2).

Attributes:

Name Type Description
conversation_id str

The app-generated id.
title str

Current title (auto-generated or the creation default).
created_at datetime

Creation timestamp.
updated_at datetime

Last-turn timestamp (list ordering key).
message_count int

Number of persisted messages (user + assistant).
group_id str | None

Sidebar group the conversation is filed under (None = ungrouped).
Source code in varagity/stores/conversation_store.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class ConversationSummary(BaseModel):
    """One conversation as listed in the sidebar (spec_v2 §4.2).

    Attributes:
        conversation_id: The app-generated id.
        title: Current title (auto-generated or the creation default).
        created_at: Creation timestamp.
        updated_at: Last-turn timestamp (list ordering key).
        message_count: Number of persisted messages (user + assistant).
        group_id: Sidebar group the conversation is filed under (``None``
            = ungrouped).
    """

    conversation_id: str
    title: str
    created_at: datetime
    updated_at: datetime
    message_count: int
    group_id: str | None = None

MessageRecord

Bases: BaseModel

One persisted message, with the assistant turn's provenance.

Attributes:

Name Type Description
message_id str

The app-generated id.
role str

"user" or "assistant".
content str

The question or the generated answer.
created_at datetime

Persistence timestamp.
retrieval_method str | None

Retrieval method that produced an assistant turn (None for user turns).
latency_ms dict[str, Any] | None

Per-stage timings of an assistant turn (None for user turns).
reasoning str | None

Captured <think> stream, if any.
condensed_query str | None

The standalone search query that drove an assistant turn's retrieval (spec_v3 §8; None = the search used the user's words verbatim).
chat_engine str | None

Registry name of the chat engine that produced an assistant turn (None for user turns and pre-v3 rows).
target_corpus str | None

Which corpus the turn asked for ("rag" | "graph"); None for user turns and pre-stage-2 history, which read as chunk RAG. Recorded even when a graph turn degraded, so the request stays attributable (ADR-017).
graph_evidence dict[str, Any] | None

The graph retrieval's entities/relations and the engine query mode, or None — which on a "graph" turn is the honest record of a degrade.
sources list[StoredSource]

The turn's snapshotted evidence, rank order. Chunk turns snapshot retrieved chunks; graph turns snapshot the cited transcript days under trace.kind = "graph_transcript".
Source code in varagity/stores/conversation_store.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class MessageRecord(BaseModel):
    """One persisted message, with the assistant turn's provenance.

    Attributes:
        message_id: The app-generated id.
        role: ``"user"`` or ``"assistant"``.
        content: The question or the generated answer.
        created_at: Persistence timestamp.
        retrieval_method: Retrieval method that produced an assistant turn
            (``None`` for user turns).
        latency_ms: Per-stage timings of an assistant turn (``None`` for
            user turns).
        reasoning: Captured ``<think>`` stream, if any.
        condensed_query: The standalone search query that drove an
            assistant turn's retrieval (spec_v3 §8; ``None`` = the search
            used the user's words verbatim).
        chat_engine: Registry name of the chat engine that produced an
            assistant turn (``None`` for user turns and pre-v3 rows).
        target_corpus: Which corpus the turn *asked* for (``"rag"`` |
            ``"graph"``); ``None`` for user turns and pre-stage-2 history,
            which read as chunk RAG. Recorded even when a graph turn
            degraded, so the request stays attributable (ADR-017).
        graph_evidence: The graph retrieval's entities/relations and the
            engine query mode, or ``None`` — which on a ``"graph"`` turn is
            the honest record of a degrade.
        sources: The turn's snapshotted evidence, rank order. Chunk turns
            snapshot retrieved chunks; graph turns snapshot the cited
            transcript days under ``trace.kind = "graph_transcript"``.
    """

    message_id: str
    role: str
    content: str
    created_at: datetime
    retrieval_method: str | None = None
    latency_ms: dict[str, Any] | None = None
    reasoning: str | None = None
    condensed_query: str | None = None
    chat_engine: str | None = None
    target_corpus: str | None = None
    graph_evidence: dict[str, Any] | None = None
    sources: list[StoredSource] = []

StoredSource

Bases: BaseModel

One snapshotted evidence row of an assistant turn (spec_v2 §9.1).

Attributes:

Name Type Description
rank int

Final rank in the answer's evidence (1-based).
chunk_id str

Soft reference to the chunk that produced the snapshot — or, on a graph turn, to the cited transcript day's doc_key.
trace dict[str, Any]

The persisted snapshot: score, content, context, source provenance fields, and the serialized :class:~varagity.stores.records.RetrievalTrace under "trace" (None when the retriever attached none). A graph turn's rows carry the other shape instead — kind="graph_transcript" plus thread/span/excerpt/guids (:func:_graph_source_snapshot) — so readers branch on kind.
Source code in varagity/stores/conversation_store.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class StoredSource(BaseModel):
    """One snapshotted evidence row of an assistant turn (spec_v2 §9.1).

    Attributes:
        rank: Final rank in the answer's evidence (1-based).
        chunk_id: Soft reference to the chunk that produced the snapshot —
            or, on a graph turn, to the cited transcript day's ``doc_key``.
        trace: The persisted snapshot: ``score``, ``content``, ``context``,
            source provenance fields, and the serialized
            :class:`~varagity.stores.records.RetrievalTrace` under
            ``"trace"`` (``None`` when the retriever attached none). A graph
            turn's rows carry the other shape instead —
            ``kind="graph_transcript"`` plus thread/span/excerpt/guids
            (:func:`_graph_source_snapshot`) — so readers branch on
            ``kind``.
    """

    rank: int
    chunk_id: str
    trace: dict[str, Any]

varagity.stores.app_settings_store

AppSettingsStore — persisted runtime setting overrides (spec_v2 §4.7).

A thin psycopg wrapper over the app_settings table (migration 002): one row per overridden :class:~varagity.config.Settings field, keyed by the field name with the value in its env-string form — exactly what pydantic-settings parses from the environment, so the override layer (:mod:varagity.api.runtime_settings) replays rows as env vars verbatim.

Keys beginning with _ are reserved for app metadata and never surface as overrides. Two are in use, both "the stores no longer match what would be produced today" flags behind a persistent GUI affordance: :data:CORPUS_STALE_KEY (an ingest-affecting setting changed since the last reingest) and :data:GRAPH_STALE_KEY (a graph source file was deleted, so the graph still holds messages whose source is gone — stage-2 decision #16). Each is cleared only by a completed API-driven rebuild of its own corpus.

CORPUS_STALE_KEY = '_corpus_stale' module-attribute

Reserved metadata key flagging the corpus stale (needs --reingest).

GRAPH_STALE_KEY = '_graph_stale' module-attribute

Reserved metadata key flagging the graph stale (needs a rebuild).

AppSettingsStore

Bases: ClosingContextMixin

Persisted key/value overrides over PostgreSQL.

Owns one autocommit connection (every operation is a single statement). Use as a context manager or call :meth:close when done.

Source code in varagity/stores/app_settings_store.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class AppSettingsStore(ClosingContextMixin):
    """Persisted key/value overrides over PostgreSQL.

    Owns one autocommit connection (every operation is a single statement).
    Use as a context manager or call :meth:`close` when done.
    """

    def __init__(self, conninfo: str | None = None) -> None:
        """Connect to the settings database.

        Args:
            conninfo: libpq connection string; defaults to the
                ``POSTGRES_*`` settings.

        Raises:
            psycopg.OperationalError: If the database is unreachable.
        """
        self._conn = psycopg.connect(conninfo or default_conninfo(), autocommit=True)

    def close(self) -> None:
        """Close the underlying connection (idempotent)."""
        if not self._conn.closed:
            self._conn.close()

    def load_overrides(self) -> dict[str, str]:
        """Read every persisted setting override.

        Returns:
            Setting name → env-string value, excluding reserved (``_``-
            prefixed) metadata rows.
        """
        rows = self._conn.execute(
            "SELECT key, value FROM app_settings WHERE key NOT LIKE '\\_%'"
        ).fetchall()
        return {key: value for key, value in rows}

    def set_override(self, name: str, value: str) -> None:
        """Insert or update one override row.

        Args:
            name: The Settings field name (e.g. ``"RETRIEVAL_METHOD"``).
            value: The value in env-string form (e.g. ``"reranked"``,
                ``"true"``, ``"0.7"``).
        """
        self._conn.execute(_UPSERT_SQL, {"key": name, "value": value})

    def delete_override(self, name: str) -> None:
        """Remove one override row (reverting the setting to its env value).

        Args:
            name: The Settings field name (deleting an absent row is a
                no-op).
        """
        self._conn.execute("DELETE FROM app_settings WHERE key = %s", (name,))

    def is_corpus_stale(self) -> bool:
        """Read the corpus-stale metadata flag.

        Returns:
            ``True`` when a reingest-affecting override changed after the
            last reingest (set by ``PATCH /api/settings``, cleared by a
            completed ``POST /api/ingest`` with ``reingest=true``).
        """
        row = self._conn.execute(
            "SELECT value FROM app_settings WHERE key = %s", (CORPUS_STALE_KEY,)
        ).fetchone()
        return row is not None and row[0] == "true"

    def set_corpus_stale(self, stale: bool) -> None:
        """Write the corpus-stale metadata flag.

        Args:
            stale: The new flag value (stored as ``"true"``/``"false"``).
        """
        self._conn.execute(
            _UPSERT_SQL, {"key": CORPUS_STALE_KEY, "value": "true" if stale else "false"}
        )

    def is_graph_stale(self) -> bool:
        """Read the graph-stale metadata flag.

        Returns:
            ``True`` when the graph no longer matches its source corpus —
            set by deleting a graph source file (the graph keeps that
            file's messages until it is rebuilt), cleared **only** by a
            completed ``POST /api/graph/build`` with ``reingest=true``.
        """
        row = self._conn.execute(
            "SELECT value FROM app_settings WHERE key = %s", (GRAPH_STALE_KEY,)
        ).fetchone()
        return row is not None and row[0] == "true"

    def set_graph_stale(self, stale: bool) -> None:
        """Write the graph-stale metadata flag.

        Args:
            stale: The new flag value (stored as ``"true"``/``"false"``).
        """
        self._conn.execute(
            _UPSERT_SQL, {"key": GRAPH_STALE_KEY, "value": "true" if stale else "false"}
        )

__init__(conninfo=None)

Connect to the settings database.

Parameters:

Name Type Description Default
conninfo str | None

libpq connection string; defaults to the POSTGRES_* settings.

 None

Raises:

Type Description
OperationalError

If the database is unreachable.
Source code in varagity/stores/app_settings_store.py 
48
49
50
51
52
53
54
55
56
57
58
def __init__(self, conninfo: str | None = None) -> None:
    """Connect to the settings database.

    Args:
        conninfo: libpq connection string; defaults to the
            ``POSTGRES_*`` settings.

    Raises:
        psycopg.OperationalError: If the database is unreachable.
    """
    self._conn = psycopg.connect(conninfo or default_conninfo(), autocommit=True)

close()

Close the underlying connection (idempotent).

Source code in varagity/stores/app_settings_store.py 
60
61
62
63
def close(self) -> None:
    """Close the underlying connection (idempotent)."""
    if not self._conn.closed:
        self._conn.close()

delete_override(name)

Remove one override row (reverting the setting to its env value).

Parameters:

Name Type Description Default
name str

The Settings field name (deleting an absent row is a no-op).

 required
Source code in varagity/stores/app_settings_store.py 
87
88
89
90
91
92
93
94
def delete_override(self, name: str) -> None:
    """Remove one override row (reverting the setting to its env value).

    Args:
        name: The Settings field name (deleting an absent row is a
            no-op).
    """
    self._conn.execute("DELETE FROM app_settings WHERE key = %s", (name,))

is_corpus_stale()

Read the corpus-stale metadata flag.

Returns:

Type Description
bool

True when a reingest-affecting override changed after the
bool

last reingest (set by PATCH /api/settings, cleared by a
bool

completed POST /api/ingest with reingest=true).
Source code in varagity/stores/app_settings_store.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def is_corpus_stale(self) -> bool:
    """Read the corpus-stale metadata flag.

    Returns:
        ``True`` when a reingest-affecting override changed after the
        last reingest (set by ``PATCH /api/settings``, cleared by a
        completed ``POST /api/ingest`` with ``reingest=true``).
    """
    row = self._conn.execute(
        "SELECT value FROM app_settings WHERE key = %s", (CORPUS_STALE_KEY,)
    ).fetchone()
    return row is not None and row[0] == "true"

is_graph_stale()

Read the graph-stale metadata flag.

Returns:

Type Description
bool

True when the graph no longer matches its source corpus —
bool

set by deleting a graph source file (the graph keeps that
bool

file's messages until it is rebuilt), cleared only by a
bool

completed POST /api/graph/build with reingest=true.
Source code in varagity/stores/app_settings_store.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
def is_graph_stale(self) -> bool:
    """Read the graph-stale metadata flag.

    Returns:
        ``True`` when the graph no longer matches its source corpus —
        set by deleting a graph source file (the graph keeps that
        file's messages until it is rebuilt), cleared **only** by a
        completed ``POST /api/graph/build`` with ``reingest=true``.
    """
    row = self._conn.execute(
        "SELECT value FROM app_settings WHERE key = %s", (GRAPH_STALE_KEY,)
    ).fetchone()
    return row is not None and row[0] == "true"

load_overrides()

Read every persisted setting override.

Returns:

Type Description
dict[str, str]

Setting name → env-string value, excluding reserved (_-
dict[str, str]

prefixed) metadata rows.
Source code in varagity/stores/app_settings_store.py 
65
66
67
68
69
70
71
72
73
74
75
def load_overrides(self) -> dict[str, str]:
    """Read every persisted setting override.

    Returns:
        Setting name → env-string value, excluding reserved (``_``-
        prefixed) metadata rows.
    """
    rows = self._conn.execute(
        "SELECT key, value FROM app_settings WHERE key NOT LIKE '\\_%'"
    ).fetchall()
    return {key: value for key, value in rows}

set_corpus_stale(stale)

Write the corpus-stale metadata flag.

Parameters:

Name Type Description Default
stale bool

The new flag value (stored as "true"/"false").

 required
Source code in varagity/stores/app_settings_store.py 
109
110
111
112
113
114
115
116
117
def set_corpus_stale(self, stale: bool) -> None:
    """Write the corpus-stale metadata flag.

    Args:
        stale: The new flag value (stored as ``"true"``/``"false"``).
    """
    self._conn.execute(
        _UPSERT_SQL, {"key": CORPUS_STALE_KEY, "value": "true" if stale else "false"}
    )

set_graph_stale(stale)

Write the graph-stale metadata flag.

Parameters:

Name Type Description Default
stale bool

The new flag value (stored as "true"/"false").

 required
Source code in varagity/stores/app_settings_store.py 
133
134
135
136
137
138
139
140
141
def set_graph_stale(self, stale: bool) -> None:
    """Write the graph-stale metadata flag.

    Args:
        stale: The new flag value (stored as ``"true"``/``"false"``).
    """
    self._conn.execute(
        _UPSERT_SQL, {"key": GRAPH_STALE_KEY, "value": "true" if stale else "false"}
    )

set_override(name, value)

Insert or update one override row.

Parameters:

Name Type Description Default
name str

The Settings field name (e.g. "RETRIEVAL_METHOD").

 required
value str

The value in env-string form (e.g. "reranked", "true", "0.7").

 required
Source code in varagity/stores/app_settings_store.py 
77
78
79
80
81
82
83
84
85
def set_override(self, name: str, value: str) -> None:
    """Insert or update one override row.

    Args:
        name: The Settings field name (e.g. ``"RETRIEVAL_METHOD"``).
        value: The value in env-string form (e.g. ``"reranked"``,
            ``"true"``, ``"0.7"``).
    """
    self._conn.execute(_UPSERT_SQL, {"key": name, "value": value})

varagity.stores.migrate

Idempotent ordered-SQL migration runner (spec_v2 §9.3).

schema.sql runs only on the postgres container's first boot (docker-entrypoint-initdb.d), so schema added after a volume exists never reaches it. This runner reconciles: ordered migrations/NNN_*.sql files — each written IF NOT EXISTS-safe — are applied in a transaction apiece and tracked by filename in a schema_migrations table. Running it repeatedly is a no-op; running it against a fresh volume and a v1 volume converges both to the same schema. It is invoked from the API's startup lifespan (varagity.api.main).

Alembic is the noted heavier alternative if migrations ever get non-trivial (plan decision #8; recorded in the runbook and ADR-005).

MIGRATIONS_PATH = Path(__file__).parent / 'migrations' module-attribute

Directory holding the ordered NNN_*.sql migration files.

run_migrations(conn, migrations_path=None)

Apply every unapplied migration file, in filename order.

Each unapplied file runs in its own transaction together with its schema_migrations bookkeeping row, so a failing migration rolls back atomically and a re-run re-attempts it.

Parameters:

Name Type Description Default
conn Connection

An open (autocommit) connection to the target database.

 required
migrations_path Path | None

Directory of NNN_*.sql files; defaults to :data:MIGRATIONS_PATH.

 None

Returns:

Type Description
list[str]

The names of the migrations applied by this call, in order
list[str]

(empty when the database was already current).

Raises:

Type Description
ValueError

If the directory contains a .sql file that does not match the NNN_name.sql convention (it would silently escape ordering otherwise).
Error

If a migration's SQL fails (the transaction for that file is rolled back).
Source code in varagity/stores/migrate.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def run_migrations(conn: psycopg.Connection, migrations_path: Path | None = None) -> list[str]:
    """Apply every unapplied migration file, in filename order.

    Each unapplied file runs in its own transaction together with its
    ``schema_migrations`` bookkeeping row, so a failing migration rolls
    back atomically and a re-run re-attempts it.

    Args:
        conn: An open (autocommit) connection to the target database.
        migrations_path: Directory of ``NNN_*.sql`` files; defaults to
            :data:`MIGRATIONS_PATH`.

    Returns:
        The names of the migrations applied by this call, in order
        (empty when the database was already current).

    Raises:
        ValueError: If the directory contains a ``.sql`` file that does not
            match the ``NNN_name.sql`` convention (it would silently escape
            ordering otherwise).
        psycopg.Error: If a migration's SQL fails (the transaction for that
            file is rolled back).
    """
    path = MIGRATIONS_PATH if migrations_path is None else migrations_path
    sql_files = sorted(path.glob("*.sql"))
    misnamed = [f.name for f in sql_files if not _MIGRATION_NAME_RE.match(f.name)]
    if misnamed:
        raise ValueError(f"migration files must be named NNN_name.sql; got {misnamed} in {path}")

    conn.execute(_CREATE_TRACKING_TABLE_SQL)
    applied = {name for (name,) in conn.execute("SELECT name FROM schema_migrations")}

    ran: list[str] = []
    for sql_file in sql_files:
        if sql_file.name in applied:
            continue
        with conn.transaction():
            conn.execute(sql_file.read_text())
            conn.execute("INSERT INTO schema_migrations (name) VALUES (%s)", (sql_file.name,))
        logger.info("applied migration %s", sql_file.name)
        ran.append(sql_file.name)
    if not ran:
        logger.info("schema is current (%d migration(s) already applied)", len(applied))
    return ran