Backups

Corpus backups (ADR-019): one zip bundle per backup, written into the corpus folder's reserved varagity-exports/ so it survives docker compose down -v beside the sources. The container (bundle.json manifest, the all-or-nothing writer, the one-member manifest read, the folder listing with each bundle's cheap restore verdict — the graph engine's file format included — and the full sha256 verification), the RAG export cut from one consistent corpus snapshot, the RAG restore that turns a bundle back into rows and a rebuilt BM25 index with no model call, the graph export that copies the engine's working directory verbatim under the graph service's write lock, and the graph restore that extracts it beside the live one and swaps it in. The HTTP side (the corpus gate, the backup runner, the routes) is in the HTTP API reference, the advisory corpus lock is on the vector store in Stores, the swap itself (GraphService.replace_workdir) is in GraphRAG, and the rag-export / rag-restore / graph-export / graph-restore flows are in Orchestration. The format, and how a restore reads it back, are described in Data model.

varagity.backup

Corpus backups: one zip bundle per backup, in each corpus's reserved folder (ADR-019).

docker compose down -v deletes every processed byte — the chunk rows, their contextual blurbs and embeddings, the BM25 index, the extracted message graph — while the source folders, being host binds, survive. A bundle written into <corpus root>/varagity-exports/ survives with them, and holds what a rebuild would otherwise have to recompute with the LLM and the embedder.

• :mod:varagity.backup.bundle — the container: the bundle.json manifest, the all-or-nothing writer (.partial → os.replace, a sha256 per member), the one-member manifest read, the folder listing with each bundle's cheap restore verdict, and the full sha256 verification.
• :mod:varagity.backup.rag — what a RAG bundle holds, how it is cut from one consistent corpus snapshot, and how a restore turns it back into rows (and a rebuilt BM25 index) with no model call.
• :mod:varagity.backup.graph — what a graph bundle holds (the engine's working directory, verbatim), how it is copied under the graph service's write lock while chat keeps answering, and how a restore extracts it and swaps it in for the live workdir with no model call.

The package knows nothing about HTTP: the API's backup runner and routes (:mod:varagity.api.backup_runner, :mod:varagity.api.routes.backups) drive it through the rag-export, rag-restore, graph-export and graph-restore Prefect flows.

BUNDLE_FORMAT = 'varagity-backup' module-attribute

The manifest's format marker: what makes a zip a varagity bundle.

BUNDLE_VERSION = 1 module-attribute

The bundle layout this code writes (and, in time, the versions it reads).

Corpus = Literal['rag', 'graph'] module-attribute

Which corpus a bundle backs up — ChatRequest.corpus's vocabulary.

MANIFEST_NAME = 'bundle.json' module-attribute

The manifest member — always the zip's last entry.

BM25RebuildFailed

Bases: RuntimeError

The rows are restored, but the BM25 index could not be rebuilt.

The Postgres commit already happened, so the restore took effect; restoring the same bundle again rebuilds the index (and rewrites the identical rows on the way).

Attributes:

Name	Type	Description
manifest		The restored bundle's manifest — the rows now in the corpus are its rows, so the corpus-stale flag must still be derived from it (None only for an instance rebuilt from its message alone, e.g. unpickled).

Source code in varagity/backup/rag.py

class BM25RebuildFailed(RuntimeError):
    """The rows are restored, but the BM25 index could not be rebuilt.

    The Postgres commit already happened, so the restore took effect;
    restoring the same bundle again rebuilds the index (and rewrites the
    identical rows on the way).

    Attributes:
        manifest: The restored bundle's manifest — the rows now in the
            corpus are its rows, so the corpus-stale flag must still be
            derived from it (``None`` only for an instance rebuilt from its
            message alone, e.g. unpickled).
    """

    def __init__(self, message: str, manifest: BundleManifest | None = None) -> None:
        """Word the failure and keep the manifest of the rows that did commit.

        Args:
            message: What failed, and how to repair it.
            manifest: The restored bundle's manifest.
        """
        super().__init__(message)
        self.manifest = manifest

__init__(message, manifest=None)

Word the failure and keep the manifest of the rows that did commit.

Parameters:

Name	Type	Description	Default
message	str	What failed, and how to repair it.	required
manifest	BundleManifest | None	The restored bundle's manifest.	None

Source code in varagity/backup/rag.py

def __init__(self, message: str, manifest: BundleManifest | None = None) -> None:
    """Word the failure and keep the manifest of the rows that did commit.

    Args:
        message: What failed, and how to repair it.
        manifest: The restored bundle's manifest.
    """
    super().__init__(message)
    self.manifest = manifest

BundleDamaged

Bases: BundleInvalid

A member's bytes don't match what its bundle recorded.

The one defect the cheap verdict can't see: the zip and its manifest are intact, but a member's sha256 (or the zip's own CRC-32, or its deflate stream) says the content changed after it was written.

Source code in varagity/backup/bundle.py

class BundleDamaged(BundleInvalid):
    """A member's bytes don't match what its bundle recorded.

    The one defect the cheap verdict can't see: the zip and its manifest are
    intact, but a member's sha256 (or the zip's own CRC-32, or its deflate
    stream) says the content changed after it was written.
    """

BundleError

Bases: Exception

Base class for a bundle that can't be read or used.

Source code in varagity/backup/bundle.py

class BundleError(Exception):
    """Base class for a bundle that can't be read or used."""

BundleIncompatible

Bases: BundleError

A sound bundle this stack must not restore.

Another bundle format version, another corpus, vectors from another embedding model or dimension — or, for a graph bundle, files another graph engine (or another major.minor version of this one) wrote. Neither store checks the embedding model (pgvector compares only the vector size, LightRAG only embedding_dim), so restoring a mismatch would succeed and then retrieve the wrong passages — it is refused instead, never converted.

Source code in varagity/backup/bundle.py

class BundleIncompatible(BundleError):
    """A sound bundle this stack must not restore.

    Another bundle format version, another corpus, vectors from another
    embedding model or dimension — or, for a graph bundle, files another
    graph engine (or another major.minor version of this one) wrote. Neither
    store checks the embedding model (pgvector compares only the vector
    size, LightRAG only ``embedding_dim``), so restoring a mismatch would
    succeed and then retrieve the wrong passages — it is refused instead,
    never converted.
    """

BundleInvalid

Bases: BundleError

The file is not a readable varagity bundle (damaged, or not one at all).

Source code in varagity/backup/bundle.py

class BundleInvalid(BundleError):
    """The file is not a readable varagity bundle (damaged, or not one at all)."""

BundleListing dataclass

One bundle file found in an exports folder.

Attributes:

Name	Type	Description
name	str	The file name.
path	Path	Its resolved path.
size_bytes	int	Its size on disk.
modified_at	datetime	Its mtime (UTC).
mtime_ns	int	Its mtime in nanoseconds — with size_bytes, the identity a remembered restore failure is keyed by.
verdict	BundleVerdict	The cheap inspection's verdict for the listing's corpus.

Source code in varagity/backup/bundle.py

@dataclass(frozen=True)
class BundleListing:
    """One bundle file found in an exports folder.

    Attributes:
        name: The file name.
        path: Its resolved path.
        size_bytes: Its size on disk.
        modified_at: Its mtime (UTC).
        mtime_ns: Its mtime in nanoseconds — with ``size_bytes``, the
            identity a remembered restore failure is keyed by.
        verdict: The cheap inspection's verdict for the listing's corpus.
    """

    name: str
    path: Path
    size_bytes: int
    modified_at: datetime
    mtime_ns: int
    verdict: BundleVerdict

    @property
    def manifest(self) -> BundleManifest | None:
        """The bundle's manifest.

        Returns:
            The manifest, or ``None`` when the file isn't a readable
            varagity bundle (copied in by hand, damaged, another format …).
        """
        return self.verdict.manifest

manifest property

The bundle's manifest.

Returns:

Type	Description
BundleManifest | None	The manifest, or None when the file isn't a readable
BundleManifest | None	varagity bundle (copied in by hand, damaged, another format …).

BundleManifest

Bases: BaseModel

bundle.json: what a bundle holds and what it was made under.

Every later phase judges bundles against this record — a restore checks compatibility against it and derives the stale flag from it — so it carries the export-time facts no row records.

Attributes:

Name	Type	Description
format	Literal['varagity-backup']	Always "varagity-backup".
version	int	The bundle layout version (:data:BUNDLE_VERSION).
corpus	Corpus	Which corpus the bundle backs up.
created_at	datetime	When the export's snapshot was taken (UTC).
embedding_model	str	EMBEDDING_MODEL at export — the store checks only vector size, so the model name is the only guard against restoring vectors from another embedding space.
embedding_dim	int	EMBEDDING_DIM at export.
stale	bool	The corpus's stale flag at export (_corpus_stale or _graph_stale).
counts	dict[str, int]	What the bundle holds — rag: documents, chunks; graph: thread_days, messages, and the entities and relations the graph reported (each only when known).
ingest_settings	dict[str, str] | None	The reingest-affecting settings at export, in env-string form (rag only; None for a corpus without them).
engine	EngineFormat | None	The graph engine's file format at export — the workdir is copied verbatim, so it is only as portable as the engine version that wrote it (graph only; None for the RAG corpus).
members	list[BundleMember]	Every data member with its size and sha256, in write order.

Source code in varagity/backup/bundle.py

class BundleManifest(BaseModel):
    """``bundle.json``: what a bundle holds and what it was made under.

    Every later phase judges bundles against this record — a restore checks
    compatibility against it and derives the stale flag from it — so it
    carries the export-time facts no row records.

    Attributes:
        format: Always ``"varagity-backup"``.
        version: The bundle layout version (:data:`BUNDLE_VERSION`).
        corpus: Which corpus the bundle backs up.
        created_at: When the export's snapshot was taken (UTC).
        embedding_model: ``EMBEDDING_MODEL`` at export — the store checks
            only vector size, so the model name is the only guard against
            restoring vectors from another embedding space.
        embedding_dim: ``EMBEDDING_DIM`` at export.
        stale: The corpus's stale flag at export (``_corpus_stale`` or
            ``_graph_stale``).
        counts: What the bundle holds — rag: ``documents``, ``chunks``;
            graph: ``thread_days``, ``messages``, and the ``entities`` and
            ``relations`` the graph reported (each only when known).
        ingest_settings: The reingest-affecting settings at export, in
            env-string form (rag only; ``None`` for a corpus without them).
        engine: The graph engine's file format at export — the workdir is
            copied verbatim, so it is only as portable as the engine version
            that wrote it (graph only; ``None`` for the RAG corpus).
        members: Every data member with its size and sha256, in write order.
    """

    format: Literal["varagity-backup"]
    version: int
    corpus: Corpus
    created_at: datetime
    embedding_model: str
    embedding_dim: int
    stale: bool
    counts: dict[str, int]
    ingest_settings: dict[str, str] | None = None
    engine: EngineFormat | None = None
    members: list[BundleMember]

BundleMember

Bases: BaseModel

One data member of a bundle, as its manifest records it.

Attributes:

Name	Type	Description
name	str	The member's path inside the zip.
size_bytes	int	Its uncompressed size.
sha256	str	The hex sha256 of its uncompressed bytes — what a restore verifies before it touches a store.

Source code in varagity/backup/bundle.py

class BundleMember(BaseModel):
    """One data member of a bundle, as its manifest records it.

    Attributes:
        name: The member's path inside the zip.
        size_bytes: Its uncompressed size.
        sha256: The hex sha256 of its uncompressed bytes — what a restore
            verifies before it touches a store.
    """

    name: str
    size_bytes: int
    sha256: str

BundleSource

An open bundle whose members are read through verifying streams.

Attributes:

Name	Type	Description
manifest		The manifest the members are checked against.

Source code in varagity/backup/bundle.py

class BundleSource:
    """An open bundle whose members are read through verifying streams.

    Attributes:
        manifest: The manifest the members are checked against.
    """

    def __init__(self, archive: zipfile.ZipFile, manifest: BundleManifest) -> None:
        """Bind the open zip to the manifest it is held to.

        Args:
            archive: The open bundle.
            manifest: Its (already inspected) manifest.
        """
        self._archive = archive
        self._records = {member.name: member for member in manifest.members}
        self.manifest = manifest

    @contextmanager
    def member(self, name: str) -> Iterator[MemberStream]:
        """Open one data member for reading, verified when the block exits.

        Several members may be open at once (zipfile reads each through its
        own position in the shared file), which is how a restore reads the
        chunk lines and their vectors side by side.

        Args:
            name: The member's name.

        Yields:
            The member's verifying stream. When the ``with`` block exits
            cleanly, whatever the reader left is drained and the member's
            size and sha256 are checked against the manifest.

        Raises:
            BundleInvalid: If the manifest records no such member, or the zip
                can't open it.
            BundleDamaged: If the member's bytes don't match its record.
        """
        record = self._records.get(name)
        if record is None:
            raise BundleInvalid(f"Missing member(s): {name}.")
        try:
            raw = self._archive.open(name)
        except KeyError:
            raise BundleInvalid(f"Missing member(s): {name}.") from None
        except _UNREADABLE_ZIP as error:
            raise BundleInvalid(f"{name} could not be opened ({error}).") from error
        with raw:
            stream = MemberStream(raw, record)
            yield stream
            stream.verify()

__init__(archive, manifest)

Bind the open zip to the manifest it is held to.

Parameters:

Name	Type	Description	Default
archive	ZipFile	The open bundle.	required
manifest	BundleManifest	Its (already inspected) manifest.	required

Source code in varagity/backup/bundle.py

def __init__(self, archive: zipfile.ZipFile, manifest: BundleManifest) -> None:
    """Bind the open zip to the manifest it is held to.

    Args:
        archive: The open bundle.
        manifest: Its (already inspected) manifest.
    """
    self._archive = archive
    self._records = {member.name: member for member in manifest.members}
    self.manifest = manifest

member(name)

Open one data member for reading, verified when the block exits.

Several members may be open at once (zipfile reads each through its own position in the shared file), which is how a restore reads the chunk lines and their vectors side by side.

Parameters:

Name	Type	Description	Default
name	str	The member's name.	required

Yields:

Type	Description
MemberStream	The member's verifying stream. When the with block exits
MemberStream	cleanly, whatever the reader left is drained and the member's
MemberStream	size and sha256 are checked against the manifest.

Raises:

Type	Description
BundleInvalid	If the manifest records no such member, or the zip can't open it.
BundleDamaged	If the member's bytes don't match its record.

Source code in varagity/backup/bundle.py

@contextmanager
def member(self, name: str) -> Iterator[MemberStream]:
    """Open one data member for reading, verified when the block exits.

    Several members may be open at once (zipfile reads each through its
    own position in the shared file), which is how a restore reads the
    chunk lines and their vectors side by side.

    Args:
        name: The member's name.

    Yields:
        The member's verifying stream. When the ``with`` block exits
        cleanly, whatever the reader left is drained and the member's
        size and sha256 are checked against the manifest.

    Raises:
        BundleInvalid: If the manifest records no such member, or the zip
            can't open it.
        BundleDamaged: If the member's bytes don't match its record.
    """
    record = self._records.get(name)
    if record is None:
        raise BundleInvalid(f"Missing member(s): {name}.")
    try:
        raw = self._archive.open(name)
    except KeyError:
        raise BundleInvalid(f"Missing member(s): {name}.") from None
    except _UNREADABLE_ZIP as error:
        raise BundleInvalid(f"{name} could not be opened ({error}).") from error
    with raw:
        stream = MemberStream(raw, record)
        yield stream
        stream.verify()

BundleVerdict dataclass

What the cheap inspection concluded about one bundle.

Attributes:

Name	Type	Description
manifest	BundleManifest | None	The bundle's manifest, or None when none could be read (also for a format version this stack can't parse).
problem	BundleError | None	Why the bundle can't be restored, or None when it can.

Source code in varagity/backup/bundle.py

@dataclass(frozen=True)
class BundleVerdict:
    """What the cheap inspection concluded about one bundle.

    Attributes:
        manifest: The bundle's manifest, or ``None`` when none could be read
            (also for a format version this stack can't parse).
        problem: Why the bundle can't be restored, or ``None`` when it can.
    """

    manifest: BundleManifest | None
    problem: BundleError | None = None

    @property
    def restorable(self) -> bool:
        """Whether nothing the inspection could see refuses the bundle.

        Returns:
            ``True`` when there is no problem (the restore still hashes it).
        """
        return self.problem is None

    @property
    def reason(self) -> str | None:
        """The problem, worded for a person.

        Returns:
            The refusal's message, or ``None`` for a restorable bundle.
        """
        return None if self.problem is None else str(self.problem)

    def manifest_or_raise(self) -> BundleManifest:
        """Return the manifest of a restorable bundle, or raise its problem.

        Returns:
            The manifest.

        Raises:
            BundleInvalid: If the bundle is not a sound bundle.
            BundleIncompatible: If this stack must not restore it.
        """
        if self.problem is not None:
            raise self.problem
        if self.manifest is None:  # pragma: no cover — a verdict always pairs one or the other
            raise BundleInvalid("the bundle has no readable manifest")
        return self.manifest

reason property

The problem, worded for a person.

Returns:

Type	Description
str | None	The refusal's message, or None for a restorable bundle.

restorable property

Whether nothing the inspection could see refuses the bundle.

Returns:

Type	Description
bool	True when there is no problem (the restore still hashes it).

manifest_or_raise()

Return the manifest of a restorable bundle, or raise its problem.

Returns:

Type	Description
BundleManifest	The manifest.

Raises:

Type	Description
BundleInvalid	If the bundle is not a sound bundle.
BundleIncompatible	If this stack must not restore it.

Source code in varagity/backup/bundle.py

def manifest_or_raise(self) -> BundleManifest:
    """Return the manifest of a restorable bundle, or raise its problem.

    Returns:
        The manifest.

    Raises:
        BundleInvalid: If the bundle is not a sound bundle.
        BundleIncompatible: If this stack must not restore it.
    """
    if self.problem is not None:
        raise self.problem
    if self.manifest is None:  # pragma: no cover — a verdict always pairs one or the other
        raise BundleInvalid("the bundle has no readable manifest")
    return self.manifest

BundleWriter

Fills one bundle: data members first, then the sealed manifest.

Obtained from :func:write_bundle; :attr:result is available once its with block exits cleanly.

Attributes:

Name	Type	Description
corpus		The corpus being backed up.
created_at		The export's instant — the name's stamp and the manifest's created_at.

Source code in varagity/backup/bundle.py

class BundleWriter:
    """Fills one bundle: data members first, then the sealed manifest.

    Obtained from :func:`write_bundle`; :attr:`result` is available once its
    ``with`` block exits cleanly.

    Attributes:
        corpus: The corpus being backed up.
        created_at: The export's instant — the name's stamp and the
            manifest's ``created_at``.
    """

    def __init__(self, archive: zipfile.ZipFile, corpus: Corpus, created_at: datetime) -> None:
        """Bind the writer to its open partial archive.

        Args:
            archive: The partial zip, open for writing.
            corpus: The corpus being backed up.
            created_at: The export's instant (UTC).
        """
        self._archive = archive
        self.corpus = corpus
        self.created_at = created_at
        self._members: list[BundleMember] = []
        self._manifest: BundleManifest | None = None
        self._result: WrittenBundle | None = None
        # Zip stores a naive DOS timestamp; the UTC wall clock is the honest
        # one (the manifest carries the aware instant regardless).
        self._date_time = created_at.astimezone(UTC).timetuple()[:6]

    @contextmanager
    def member(self, name: str, *, compress: bool = True) -> Iterator[MemberSink]:
        """Stream one data member into the bundle.

        Args:
            name: The member's path inside the zip.
            compress: Deflate it (text) or store it as-is (already-dense
                bytes like a float32 matrix, which deflate can't shrink).

        Yields:
            The sink to write the member's bytes to; its size and sha256 are
            recorded for the manifest when the block exits.

        Raises:
            ValueError: If the name is the manifest's or already used.
            RuntimeError: If the manifest is already sealed.
        """
        if self._manifest is not None:
            raise RuntimeError("the manifest is sealed — no member may follow it")
        if name == MANIFEST_NAME or any(member.name == name for member in self._members):
            raise ValueError(f"bundle member {name!r} is reserved or already written")
        info = zipfile.ZipInfo(name, date_time=self._date_time)
        info.compress_type = zipfile.ZIP_DEFLATED if compress else zipfile.ZIP_STORED
        info.external_attr = _MEMBER_MODE
        # force_zip64: the size is unknown up front, and a corpus's vectors
        # can pass the 2 GiB a plain entry can describe.
        with self._archive.open(info, mode="w", force_zip64=True) as raw:
            sink = MemberSink(raw)
            yield sink
        self._members.append(
            BundleMember(name=name, size_bytes=sink.size_bytes, sha256=sink.hexdigest())
        )

    def seal(
        self,
        *,
        embedding_model: str,
        embedding_dim: int,
        stale: bool,
        counts: dict[str, int],
        ingest_settings: dict[str, str] | None = None,
        engine: EngineFormat | None = None,
    ) -> BundleManifest:
        """Write ``bundle.json`` as the bundle's last member.

        Args:
            embedding_model: ``EMBEDDING_MODEL`` at export.
            embedding_dim: ``EMBEDDING_DIM`` at export.
            stale: The corpus's stale flag at export.
            counts: What the bundle holds.
            ingest_settings: The reingest-affecting settings at export, if
                the corpus has them.
            engine: The graph engine's file format, if the corpus is the
                graph.

        Returns:
            The manifest, as written.

        Raises:
            RuntimeError: If the manifest is already sealed.
        """
        if self._manifest is not None:
            raise RuntimeError("the manifest is already sealed")
        manifest = BundleManifest(
            format=BUNDLE_FORMAT,
            version=BUNDLE_VERSION,
            corpus=self.corpus,
            created_at=self.created_at,
            embedding_model=embedding_model,
            embedding_dim=embedding_dim,
            stale=stale,
            counts=dict(counts),
            ingest_settings=dict(ingest_settings) if ingest_settings is not None else None,
            engine=engine,
            members=list(self._members),
        )
        info = zipfile.ZipInfo(MANIFEST_NAME, date_time=self._date_time)
        info.compress_type = zipfile.ZIP_DEFLATED
        info.external_attr = _MEMBER_MODE
        self._archive.writestr(info, manifest.model_dump_json(indent=2) + "\n")
        self._manifest = manifest
        return manifest

    @property
    def manifest(self) -> BundleManifest | None:
        """The sealed manifest (``None`` until :meth:`seal` runs).

        Returns:
            The manifest, or ``None``.
        """
        return self._manifest

    @property
    def result(self) -> WrittenBundle:
        """The published bundle.

        Returns:
            What :func:`write_bundle` moved into place.

        Raises:
            RuntimeError: If the bundle was not (yet) published.
        """
        if self._result is None:
            raise RuntimeError("the bundle has not been published")
        return self._result

    def _publish(self, result: WrittenBundle) -> None:
        """Record the published bundle (called by :func:`write_bundle`).

        Args:
            result: The bundle now in place.
        """
        self._result = result

manifest property

The sealed manifest (None until :meth:seal runs).

Returns:

Type	Description
BundleManifest | None	The manifest, or None.

result property

The published bundle.

Returns:

Name	Type	Description
What	WrittenBundle	func:write_bundle moved into place.

Raises:

Type	Description
RuntimeError	If the bundle was not (yet) published.

__init__(archive, corpus, created_at)

Bind the writer to its open partial archive.

Parameters:

Name	Type	Description	Default
archive	ZipFile	The partial zip, open for writing.	required
corpus	Corpus	The corpus being backed up.	required
created_at	datetime	The export's instant (UTC).	required

Source code in varagity/backup/bundle.py

def __init__(self, archive: zipfile.ZipFile, corpus: Corpus, created_at: datetime) -> None:
    """Bind the writer to its open partial archive.

    Args:
        archive: The partial zip, open for writing.
        corpus: The corpus being backed up.
        created_at: The export's instant (UTC).
    """
    self._archive = archive
    self.corpus = corpus
    self.created_at = created_at
    self._members: list[BundleMember] = []
    self._manifest: BundleManifest | None = None
    self._result: WrittenBundle | None = None
    # Zip stores a naive DOS timestamp; the UTC wall clock is the honest
    # one (the manifest carries the aware instant regardless).
    self._date_time = created_at.astimezone(UTC).timetuple()[:6]

member(name, *, compress=True)

Stream one data member into the bundle.

Parameters:

Name	Type	Description	Default
name	str	The member's path inside the zip.	required
compress	bool	Deflate it (text) or store it as-is (already-dense bytes like a float32 matrix, which deflate can't shrink).	True

Yields:

Type	Description
MemberSink	The sink to write the member's bytes to; its size and sha256 are
MemberSink	recorded for the manifest when the block exits.

Raises:

Type	Description
ValueError	If the name is the manifest's or already used.
RuntimeError	If the manifest is already sealed.

Source code in varagity/backup/bundle.py

@contextmanager
def member(self, name: str, *, compress: bool = True) -> Iterator[MemberSink]:
    """Stream one data member into the bundle.

    Args:
        name: The member's path inside the zip.
        compress: Deflate it (text) or store it as-is (already-dense
            bytes like a float32 matrix, which deflate can't shrink).

    Yields:
        The sink to write the member's bytes to; its size and sha256 are
        recorded for the manifest when the block exits.

    Raises:
        ValueError: If the name is the manifest's or already used.
        RuntimeError: If the manifest is already sealed.
    """
    if self._manifest is not None:
        raise RuntimeError("the manifest is sealed — no member may follow it")
    if name == MANIFEST_NAME or any(member.name == name for member in self._members):
        raise ValueError(f"bundle member {name!r} is reserved or already written")
    info = zipfile.ZipInfo(name, date_time=self._date_time)
    info.compress_type = zipfile.ZIP_DEFLATED if compress else zipfile.ZIP_STORED
    info.external_attr = _MEMBER_MODE
    # force_zip64: the size is unknown up front, and a corpus's vectors
    # can pass the 2 GiB a plain entry can describe.
    with self._archive.open(info, mode="w", force_zip64=True) as raw:
        sink = MemberSink(raw)
        yield sink
    self._members.append(
        BundleMember(name=name, size_bytes=sink.size_bytes, sha256=sink.hexdigest())
    )

seal(*, embedding_model, embedding_dim, stale, counts, ingest_settings=None, engine=None)

Write bundle.json as the bundle's last member.

Parameters:

Name	Type	Description	Default
embedding_model	str	EMBEDDING_MODEL at export.	required
embedding_dim	int	EMBEDDING_DIM at export.	required
stale	bool	The corpus's stale flag at export.	required
counts	dict[str, int]	What the bundle holds.	required
ingest_settings	dict[str, str] | None	The reingest-affecting settings at export, if the corpus has them.	None
engine	EngineFormat | None	The graph engine's file format, if the corpus is the graph.	None

Returns:

Type	Description
BundleManifest	The manifest, as written.

Raises:

Type	Description
RuntimeError	If the manifest is already sealed.

Source code in varagity/backup/bundle.py

def seal(
    self,
    *,
    embedding_model: str,
    embedding_dim: int,
    stale: bool,
    counts: dict[str, int],
    ingest_settings: dict[str, str] | None = None,
    engine: EngineFormat | None = None,
) -> BundleManifest:
    """Write ``bundle.json`` as the bundle's last member.

    Args:
        embedding_model: ``EMBEDDING_MODEL`` at export.
        embedding_dim: ``EMBEDDING_DIM`` at export.
        stale: The corpus's stale flag at export.
        counts: What the bundle holds.
        ingest_settings: The reingest-affecting settings at export, if
            the corpus has them.
        engine: The graph engine's file format, if the corpus is the
            graph.

    Returns:
        The manifest, as written.

    Raises:
        RuntimeError: If the manifest is already sealed.
    """
    if self._manifest is not None:
        raise RuntimeError("the manifest is already sealed")
    manifest = BundleManifest(
        format=BUNDLE_FORMAT,
        version=BUNDLE_VERSION,
        corpus=self.corpus,
        created_at=self.created_at,
        embedding_model=embedding_model,
        embedding_dim=embedding_dim,
        stale=stale,
        counts=dict(counts),
        ingest_settings=dict(ingest_settings) if ingest_settings is not None else None,
        engine=engine,
        members=list(self._members),
    )
    info = zipfile.ZipInfo(MANIFEST_NAME, date_time=self._date_time)
    info.compress_type = zipfile.ZIP_DEFLATED
    info.external_attr = _MEMBER_MODE
    self._archive.writestr(info, manifest.model_dump_json(indent=2) + "\n")
    self._manifest = manifest
    return manifest

GraphExportReport dataclass

What one graph export wrote.

Attributes:

Name	Type	Description
bundle	WrittenBundle	The published bundle (name, path, size, manifest).
files	int	How many workdir files it holds.
wall_clock_s	float	Seconds from the export's start (the write lock is taken at once) to the published bundle.

Source code in varagity/backup/graph.py

@dataclass(frozen=True)
class GraphExportReport:
    """What one graph export wrote.

    Attributes:
        bundle: The published bundle (name, path, size, manifest).
        files: How many workdir files it holds.
        wall_clock_s: Seconds from the export's start (the write lock is
            taken at once) to the published bundle.
    """

    bundle: WrittenBundle
    files: int
    wall_clock_s: float

    @property
    def counts(self) -> dict[str, int]:
        """What the bundle holds, by the graph's own summary (see :func:`graph_counts`).

        Returns:
            The manifest's counts.
        """
        return dict(self.bundle.manifest.counts)

counts property

What the bundle holds, by the graph's own summary (see :func:graph_counts).

Returns:

Type	Description
dict[str, int]	The manifest's counts.

GraphRestore dataclass

What one graph restore did.

Attributes:

Name	Type	Description
bundle_name	str	The restored bundle's file name.
size_bytes	int	The bundle's size on disk.
manifest	BundleManifest	Its manifest.
files	int	How many workdir files it restored.
stats	GraphStats	The restored graph's size, as its freshly opened session reported it.
wall_clock_s	float	Seconds from validation to an open graph.
stale	bool | None	The graph-stale flag the restore left — the bundle's own, written by the graph-restore flow; None until it has.

Source code in varagity/backup/graph.py

@dataclass(frozen=True)
class GraphRestore:
    """What one graph restore did.

    Attributes:
        bundle_name: The restored bundle's file name.
        size_bytes: The bundle's size on disk.
        manifest: Its manifest.
        files: How many workdir files it restored.
        stats: The restored graph's size, as its freshly opened session
            reported it.
        wall_clock_s: Seconds from validation to an open graph.
        stale: The graph-stale flag the restore left — the bundle's own,
            written by the ``graph-restore`` flow; ``None`` until it has.
    """

    bundle_name: str
    size_bytes: int
    manifest: BundleManifest
    files: int
    stats: GraphStats
    wall_clock_s: float
    stale: bool | None = None

    @property
    def counts(self) -> dict[str, int]:
        """What the restored graph holds, by the bundle's manifest (its size summary at export).

        Returns:
            ``thread_days``, ``messages`` and — when the graph reported them
            at export — ``entities`` and ``relations``.
        """
        return dict(self.manifest.counts)

counts property

What the restored graph holds, by the bundle's manifest (its size summary at export).

Returns:

Type	Description
dict[str, int]	thread_days, messages and — when the graph reported them
dict[str, int]	at export — entities and relations.

MemberStream

The read end of one bundle member, hashed and counted as it is read.

Every byte handed to the reader passes through :meth:read or :meth:readline, so once the member is drained the digest covers it whole; :meth:verify then holds it to the manifest's record.

Attributes:

Name	Type	Description
name		The member's name.

Source code in varagity/backup/bundle.py

class MemberStream:
    """The read end of one bundle member, hashed and counted as it is read.

    Every byte handed to the reader passes through :meth:`read` or
    :meth:`readline`, so once the member is drained the digest covers it
    whole; :meth:`verify` then holds it to the manifest's record.

    Attributes:
        name: The member's name.
    """

    def __init__(self, raw: IO[bytes], record: BundleMember) -> None:
        """Wrap a member's decompressing read stream.

        Args:
            raw: The stream ``ZipFile.open(name)`` returned.
            record: What the manifest recorded for the member.
        """
        self.name = record.name
        self._raw = raw
        self._record = record
        self._hash = hashlib.sha256()
        self._size = 0

    def _failure(self, error: BaseException) -> BundleError:
        """Classify a read failure as damage or as an unreadable bundle.

        Args:
            error: What reading the member raised.

        Returns:
            ``BundleDamaged`` for a CRC or deflate failure (the bytes
            changed), ``BundleInvalid`` otherwise.
        """
        if isinstance(error, _DAMAGED_MEMBER):
            return BundleDamaged(
                f"{self.name} could not be read ({error}) — the bundle is damaged."
            )
        return BundleInvalid(f"{self.name} could not be read ({error}).")

    def _take(self, data: bytes) -> bytes:
        """Count and hash bytes on their way to the reader.

        Args:
            data: What the underlying stream returned.

        Returns:
            The same bytes.
        """
        self._hash.update(data)
        self._size += len(data)
        return data

    def read(self, size: int = -1) -> bytes:
        """Read up to ``size`` bytes (all that is left when negative).

        Args:
            size: The most bytes to return.

        Returns:
            The bytes; ``b""`` at the end of the member.

        Raises:
            BundleDamaged: If the member's bytes fail the zip's own checks.
            BundleInvalid: If the member can't be read for another reason.
        """
        try:
            data = self._raw.read(size)
        except _UNREADABLE_ZIP as error:
            raise self._failure(error) from error
        return self._take(data)

    def readline(self) -> bytes:
        """Read one line, newline included.

        Returns:
            The line; ``b""`` at the end of the member.

        Raises:
            BundleDamaged: If the member's bytes fail the zip's own checks.
            BundleInvalid: If the member can't be read for another reason.
        """
        try:
            data = self._raw.readline()
        except _UNREADABLE_ZIP as error:
            raise self._failure(error) from error
        return self._take(data)

    def __iter__(self) -> Iterator[bytes]:
        """Iterate the member's lines (a JSONL member's records).

        Yields:
            Each line, newline included.
        """
        while line := self.readline():
            yield line

    def verify(self) -> None:
        """Drain what the reader left, then hold the member to its record.

        Raises:
            BundleDamaged: If its size or sha256 differs from what the
                manifest recorded.
        """
        while self.read(_READ_BYTES):
            pass
        if self._size != self._record.size_bytes or self._hash.hexdigest() != self._record.sha256:
            raise BundleDamaged(
                f"{self.name} does not match its recorded sha256 — the bundle is damaged."
            )

__init__(raw, record)

Wrap a member's decompressing read stream.

Parameters:

Name	Type	Description	Default
raw	IO[bytes]	The stream ZipFile.open(name) returned.	required
record	BundleMember	What the manifest recorded for the member.	required

Source code in varagity/backup/bundle.py

def __init__(self, raw: IO[bytes], record: BundleMember) -> None:
    """Wrap a member's decompressing read stream.

    Args:
        raw: The stream ``ZipFile.open(name)`` returned.
        record: What the manifest recorded for the member.
    """
    self.name = record.name
    self._raw = raw
    self._record = record
    self._hash = hashlib.sha256()
    self._size = 0

__iter__()

Iterate the member's lines (a JSONL member's records).

Yields:

Type	Description
bytes	Each line, newline included.

Source code in varagity/backup/bundle.py

def __iter__(self) -> Iterator[bytes]:
    """Iterate the member's lines (a JSONL member's records).

    Yields:
        Each line, newline included.
    """
    while line := self.readline():
        yield line

read(size=-1)

Read up to size bytes (all that is left when negative).

Parameters:

Name	Type	Description	Default
size	int	The most bytes to return.	-1

Returns:

Type	Description
bytes	The bytes; b"" at the end of the member.

Raises:

Type	Description
BundleDamaged	If the member's bytes fail the zip's own checks.
BundleInvalid	If the member can't be read for another reason.

Source code in varagity/backup/bundle.py

def read(self, size: int = -1) -> bytes:
    """Read up to ``size`` bytes (all that is left when negative).

    Args:
        size: The most bytes to return.

    Returns:
        The bytes; ``b""`` at the end of the member.

    Raises:
        BundleDamaged: If the member's bytes fail the zip's own checks.
        BundleInvalid: If the member can't be read for another reason.
    """
    try:
        data = self._raw.read(size)
    except _UNREADABLE_ZIP as error:
        raise self._failure(error) from error
    return self._take(data)

readline()

Read one line, newline included.

Returns:

Type	Description
bytes	The line; b"" at the end of the member.

Raises:

Type	Description
BundleDamaged	If the member's bytes fail the zip's own checks.
BundleInvalid	If the member can't be read for another reason.

Source code in varagity/backup/bundle.py

def readline(self) -> bytes:
    """Read one line, newline included.

    Returns:
        The line; ``b""`` at the end of the member.

    Raises:
        BundleDamaged: If the member's bytes fail the zip's own checks.
        BundleInvalid: If the member can't be read for another reason.
    """
    try:
        data = self._raw.readline()
    except _UNREADABLE_ZIP as error:
        raise self._failure(error) from error
    return self._take(data)

verify()

Drain what the reader left, then hold the member to its record.

Raises:

Type	Description
BundleDamaged	If its size or sha256 differs from what the manifest recorded.

Source code in varagity/backup/bundle.py

def verify(self) -> None:
    """Drain what the reader left, then hold the member to its record.

    Raises:
        BundleDamaged: If its size or sha256 differs from what the
            manifest recorded.
    """
    while self.read(_READ_BYTES):
        pass
    if self._size != self._record.size_bytes or self._hash.hexdigest() != self._record.sha256:
        raise BundleDamaged(
            f"{self.name} does not match its recorded sha256 — the bundle is damaged."
        )

NothingToExport

Bases: RuntimeError

Raised when a corpus holds nothing to back up — an empty bundle is not a backup.

The RAG corpus with no documents row, or a graph whose working directory has no file to copy (nothing has ever been built).

Source code in varagity/backup/bundle.py

class NothingToExport(RuntimeError):
    """Raised when a corpus holds nothing to back up — an empty bundle is not a backup.

    The RAG corpus with no ``documents`` row, or a graph whose working
    directory has no file to copy (nothing has ever been built).
    """

RagExport dataclass

What one RAG export wrote.

Attributes:

Name	Type	Description
bundle	WrittenBundle	The published bundle (name, path, size, manifest).
wall_clock_s	float	Seconds from snapshot to publish.

Source code in varagity/backup/rag.py

@dataclass(frozen=True)
class RagExport:
    """What one RAG export wrote.

    Attributes:
        bundle: The published bundle (name, path, size, manifest).
        wall_clock_s: Seconds from snapshot to publish.
    """

    bundle: WrittenBundle
    wall_clock_s: float

    @property
    def counts(self) -> dict[str, int]:
        """The bundle's row counts (``documents``, ``chunks``).

        Returns:
            The manifest's counts.
        """
        return dict(self.bundle.manifest.counts)

counts property

The bundle's row counts (documents, chunks).

Returns:

Type	Description
dict[str, int]	The manifest's counts.

RagRestore dataclass

What one RAG restore did.

Attributes:

Name	Type	Description
bundle_name	str	The restored bundle's file name.
size_bytes	int	The bundle's size on disk.
manifest	BundleManifest	Its manifest.
documents	int	documents rows written.
chunks	int	chunks rows written.
bm25_indexed	int	Chunks the rebuilt BM25 index holds.
wall_clock_s	float	Seconds from validation to a searchable index.
stale	bool | None	The corpus-stale flag the restore left — derived from the bundle by the rag-restore flow; None until it has.

Source code in varagity/backup/rag.py

@dataclass(frozen=True)
class RagRestore:
    """What one RAG restore did.

    Attributes:
        bundle_name: The restored bundle's file name.
        size_bytes: The bundle's size on disk.
        manifest: Its manifest.
        documents: ``documents`` rows written.
        chunks: ``chunks`` rows written.
        bm25_indexed: Chunks the rebuilt BM25 index holds.
        wall_clock_s: Seconds from validation to a searchable index.
        stale: The corpus-stale flag the restore left — derived from the
            bundle by the ``rag-restore`` flow; ``None`` until it has.
    """

    bundle_name: str
    size_bytes: int
    manifest: BundleManifest
    documents: int
    chunks: int
    bm25_indexed: int
    wall_clock_s: float
    stale: bool | None = None

    @property
    def counts(self) -> dict[str, int]:
        """What the restore wrote (``documents``, ``chunks``, ``bm25``).

        Returns:
            The row counts, plus the chunks the BM25 index now holds.
        """
        return {"documents": self.documents, "chunks": self.chunks, "bm25": self.bm25_indexed}

counts property

What the restore wrote (documents, chunks, bm25).

Returns:

Type	Description
dict[str, int]	The row counts, plus the chunks the BM25 index now holds.

WrittenBundle dataclass

A bundle :func:write_bundle published.

Attributes:

Name	Type	Description
name	str	The bundle's file name.
path	Path	Where it now lives.
size_bytes	int	Its size on disk.
manifest	BundleManifest	The manifest sealed into it.

Source code in varagity/backup/bundle.py

@dataclass(frozen=True)
class WrittenBundle:
    """A bundle :func:`write_bundle` published.

    Attributes:
        name: The bundle's file name.
        path: Where it now lives.
        size_bytes: Its size on disk.
        manifest: The manifest sealed into it.
    """

    name: str
    path: Path
    size_bytes: int
    manifest: BundleManifest

bundle_name(directory, corpus, *, now=None)

Choose the file name for a new bundle — never one already taken.

varagity-<corpus>-<UTC yyyymmdd-HHMM>.zip; a repeat within the same minute gets a -2, -3, … suffix rather than replacing a backup.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder the bundle will land in.	required
corpus	Corpus	The corpus being backed up.	required
now	datetime | None	The export's instant (defaults to now); rendered in UTC.	None

Returns:

Type	Description
str	A name free in directory (also free of an in-flight partial).

Source code in varagity/backup/bundle.py

def bundle_name(directory: Path, corpus: Corpus, *, now: datetime | None = None) -> str:
    """Choose the file name for a new bundle — never one already taken.

    ``varagity-<corpus>-<UTC yyyymmdd-HHMM>.zip``; a repeat within the same
    minute gets a ``-2``, ``-3``, … suffix rather than replacing a backup.

    Args:
        directory: The exports folder the bundle will land in.
        corpus: The corpus being backed up.
        now: The export's instant (defaults to now); rendered in UTC.

    Returns:
        A name free in ``directory`` (also free of an in-flight partial).
    """
    stamp = (now or datetime.now(UTC)).astimezone(UTC).strftime("%Y%m%d-%H%M")
    base = f"varagity-{corpus}-{stamp}"
    candidate = f"{base}{BUNDLE_SUFFIX}"
    attempt = 2
    while _taken(directory, candidate):
        candidate = f"{base}-{attempt}{BUNDLE_SUFFIX}"
        attempt += 1
    return candidate

corpus_root(corpus)

The configured root directory of one corpus.

Parameters:

Name	Type	Description	Default
corpus	Corpus	The corpus.	required

Returns:

Type	Description
Path	DOCS_PATH for rag, GRAPH_DOCS_PATH for graph.

Source code in varagity/backup/bundle.py

def corpus_root(corpus: Corpus) -> Path:
    """The configured root directory of one corpus.

    Args:
        corpus: The corpus.

    Returns:
        ``DOCS_PATH`` for ``rag``, ``GRAPH_DOCS_PATH`` for ``graph``.
    """
    settings = get_settings()
    return Path(settings.DOCS_PATH if corpus == "rag" else settings.GRAPH_DOCS_PATH)

derive_corpus_stale(manifest, current)

The corpus-stale flag a restored RAG bundle leaves.

Stale when the bundle was stale at export, or when the settings its rows were made under differ from this stack's — the same "the corpus may not match the current settings" the flag always means.

Parameters:

Name	Type	Description	Default
manifest	BundleManifest	The restored bundle's manifest.	required
current	Mapping[str, str]	This stack's reingest-affecting settings, in env-string form.	required

Returns:

Type	Description
bool	Whether the restored corpus is stale.

Source code in varagity/backup/rag.py

def derive_corpus_stale(manifest: BundleManifest, current: Mapping[str, str]) -> bool:
    """The corpus-stale flag a restored RAG bundle leaves.

    Stale when the bundle was stale at export, or when the settings its
    rows were made under differ from this stack's — the same "the corpus
    may not match the current settings" the flag always means.

    Args:
        manifest: The restored bundle's manifest.
        current: This stack's reingest-affecting settings, in env-string
            form.

    Returns:
        Whether the restored corpus is stale.
    """
    return manifest.stale or bool(ingest_drift(manifest, current))

display_path(corpus, name=None)

The human form of a bundle's location — the text "Copy path" copies.

In compose the API sees /app/docs while the user's checkout has ./docs (the bind's host side), so an absolute path would be wrong on the one machine that matters. The corpus folder's own name, relative, reads right from the checkout root in both host and container mode.

Parameters:

Name	Type	Description	Default
corpus	Corpus	The corpus.	required
name	str | None	A bundle file name; None names the folder itself.	None

Returns:

Type	Description
str	E.g. "docs/varagity-exports/varagity-rag-20261108-1312.zip" or
str	"docs/varagity-exports/".

Source code in varagity/backup/bundle.py

def display_path(corpus: Corpus, name: str | None = None) -> str:
    """The human form of a bundle's location — the text "Copy path" copies.

    In compose the API sees ``/app/docs`` while the user's checkout has
    ``./docs`` (the bind's host side), so an absolute path would be wrong on
    the one machine that matters. The corpus folder's own name, relative,
    reads right from the checkout root in both host and container mode.

    Args:
        corpus: The corpus.
        name: A bundle file name; ``None`` names the folder itself.

    Returns:
        E.g. ``"docs/varagity-exports/varagity-rag-20261108-1312.zip"`` or
        ``"docs/varagity-exports/"``.
    """
    # A root with no name is "/" itself — and "/varagity-exports/" is then
    # exactly right.
    folder = f"{corpus_root(corpus).resolve().name}/{EXPORTS_DIRNAME}/"
    return folder if name is None else f"{folder}{name}"

export_graph(service, dest, *, read_stale, on_progress=None)

Export the message graph's working directory into a new bundle in dest.

Holds the service's write lock for the whole copy — no build, resume or delete can start until the bundle is published, while reads (graph chat, /graph) are answered as usual. No model is called and no session is opened: the files are copied as they are on disk, which is exactly what an API restart would load.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (its workdir, write lock, and engine format).	required
dest	Path	The exports folder (<GRAPH_DOCS_PATH>/varagity-exports).	required
read_stale	Callable[[], bool]	Reads the graph-stale flag (the flow passes a Postgres read). Called once, after the last file is copied and before the manifest is sealed, so a source deleted mid-export is recorded; if it raises, the export fails and leaves no bundle.	required
on_progress	GraphExportProgressCallback | None	Called with (stage, current, total) as the export moves through snapshot (take the lock, list the files) → pack (bytes copied of the files' total) → finalize.	None

Returns:

Type	Description
GraphExportReport	The published bundle and how long it took.

Raises:

Type	Description
GraphBuildInProgress	If a build, resume or delete holds the write lock.
NothingToExport	If the workdir holds no file to copy (nothing has been built).
ValueError	If the engine's name could not be the bundle's folder.
OSError	If a file can't be read or the bundle can't be written.

Source code in varagity/backup/graph.py

def export_graph(
    service: GraphService,
    dest: Path,
    *,
    read_stale: Callable[[], bool],
    on_progress: GraphExportProgressCallback | None = None,
) -> GraphExportReport:
    """Export the message graph's working directory into a new bundle in ``dest``.

    Holds the service's write lock for the whole copy — no build, resume or
    delete can start until the bundle is published, while reads (graph chat,
    ``/graph``) are answered as usual. No model is called and no session is
    opened: the files are copied as they are on disk, which is exactly what
    an API restart would load.

    Args:
        service: The process's graph service (its workdir, write lock, and
            engine format).
        dest: The exports folder (``<GRAPH_DOCS_PATH>/varagity-exports``).
        read_stale: Reads the graph-stale flag (the flow passes a Postgres
            read). Called once, after the last file is copied and before the
            manifest is sealed, so a source deleted mid-export is recorded;
            if it raises, the export fails and leaves no bundle.
        on_progress: Called with ``(stage, current, total)`` as the export
            moves through ``snapshot`` (take the lock, list the files) →
            ``pack`` (bytes copied of the files' total) → ``finalize``.

    Returns:
        The published bundle and how long it took.

    Raises:
        varagity.graph.service.GraphBuildInProgress: If a build, resume or
            delete holds the write lock.
        NothingToExport: If the workdir holds no file to copy (nothing has
            been built).
        ValueError: If the engine's name could not be the bundle's folder.
        OSError: If a file can't be read or the bundle can't be written.
    """
    settings = get_settings()
    report: GraphExportProgressCallback = on_progress if on_progress is not None else _ignore
    started = time.perf_counter()
    created_at = datetime.now(UTC)

    report("snapshot", None, None)
    with service.writing():
        files, skipped = _workdir_entries(service.workdir)
        for name, why in skipped:
            logger.warning("not exporting %s from the graph workdir — %s", name, why)
        if not files:
            raise NothingToExport(
                "nothing has been built into the graph — there is nothing to export"
            )
        engine = service.storage_format()
        if not is_graph_file_name(engine.name):
            raise ValueError(f"the graph engine {engine.name!r} can't name a bundle folder")
        counts = graph_counts(service.workdir)
        total = sum(path.stat().st_size for path in files)
        # File names only in these lines: they are relayed into the panel's
        # log, where an in-container path (/app/graph-data/…) would read as
        # nonsense beside the checkout-relative "Saved to" path.
        logger.info(
            "exporting %d graph file(s), %s bytes, written by %s %s",
            len(files),
            f"{total:,}",
            engine.package,
            engine.version,
        )
        with write_bundle(dest, "graph", created_at=created_at) as bundle:
            progress = _ByteProgress(report, total)
            for path in files:
                with (
                    path.open("rb") as handle,
                    bundle.member(f"{engine.name}/{path.name}") as member,
                ):
                    while block := handle.read(_COPY_BYTES):
                        member.write(block)
                        progress.advance(len(block))
            progress.finish()
            report("finalize", None, None)
            # Last, not first: a source deleted during the copy has flagged
            # the graph stale by now (see the module docstring).
            stale = read_stale()
            bundle.seal(
                embedding_model=settings.EMBEDDING_MODEL,
                embedding_dim=settings.EMBEDDING_DIM,
                stale=stale,
                counts=counts,
                engine=engine,
            )
    written = bundle.result
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "exported %d graph file(s) to %s (%s bytes) in %.1fs",
        len(files),
        written.name,
        f"{written.size_bytes:,}",
        wall_clock_s,
    )
    return GraphExportReport(bundle=written, files=len(files), wall_clock_s=wall_clock_s)

export_rag(dest, store, ingest_settings, *, stale, on_progress=None)

Export the whole chunk corpus into a new bundle in dest.

Parameters:

Name	Type	Description	Default
dest	Path	The exports folder (<DOCS_PATH>/varagity-exports).	required
store	ContextualVectorDB	The vector store to snapshot.	required
ingest_settings	Mapping[str, str]	The reingest-affecting settings at export, in env-string form — the runner reads them, so this package never imports the API layer.	required
stale	bool	The corpus-stale flag at export.	required
on_progress	ProgressCallback | None	Called with (stage, current, total) as the export moves through snapshot → documents → chunks → finalize.	None

Returns:

Type	Description
RagExport	The published bundle and how long it took.

Raises:

Type	Description
NothingToExport	If the snapshot holds no documents.
ValueError	If a stored vector's dimension isn't EMBEDDING_DIM.
OSError	If the bundle can't be written.

Source code in varagity/backup/rag.py

def export_rag(
    dest: Path,
    store: ContextualVectorDB,
    ingest_settings: Mapping[str, str],
    *,
    stale: bool,
    on_progress: ProgressCallback | None = None,
) -> RagExport:
    """Export the whole chunk corpus into a new bundle in ``dest``.

    Args:
        dest: The exports folder (``<DOCS_PATH>/varagity-exports``).
        store: The vector store to snapshot.
        ingest_settings: The reingest-affecting settings at export, in
            env-string form — the runner reads them, so this package never
            imports the API layer.
        stale: The corpus-stale flag at export.
        on_progress: Called with ``(stage, current, total)`` as the export
            moves through ``snapshot`` → ``documents`` → ``chunks`` →
            ``finalize``.

    Returns:
        The published bundle and how long it took.

    Raises:
        NothingToExport: If the snapshot holds no documents.
        ValueError: If a stored vector's dimension isn't ``EMBEDDING_DIM``.
        OSError: If the bundle can't be written.
    """
    settings = get_settings()
    report: ProgressCallback = on_progress if on_progress is not None else _ignore
    docs_root = Path(settings.DOCS_PATH).resolve()
    dim = settings.EMBEDDING_DIM
    started = time.perf_counter()
    created_at = datetime.now(UTC)

    report("snapshot", None, None)
    with store.iter_corpus() as snapshot:
        if snapshot.document_count == 0:
            raise NothingToExport("the corpus holds no documents — there is nothing to export")
        # File names only in these lines: they are relayed into the panel's
        # log, where an in-container path (/app/docs/…) would contradict the
        # checkout-relative "Saved to" path beside it.
        logger.info(
            "exporting %d document(s) and %d chunk(s) from one snapshot",
            snapshot.document_count,
            snapshot.chunk_count,
        )
        with write_bundle(dest, "rag", created_at=created_at) as bundle:
            documents = _write_documents(bundle, snapshot, docs_root, report)
            # The spool sits beside the bundle (same disk) and is hidden and
            # unnamed where the OS allows: it is never listed, and it
            # vanishes with the process if the export dies.
            with tempfile.TemporaryFile(dir=dest, prefix=".varagity-rag-spool-") as spool:
                chunks, models = _write_chunks(
                    bundle, snapshot, dim=dim, spool=spool, report=report
                )
                report("finalize", None, None)
                _write_embeddings(bundle, spool)
            bundle.seal(
                embedding_model=settings.EMBEDDING_MODEL,
                embedding_dim=dim,
                stale=stale,
                counts={"documents": documents, "chunks": chunks},
                ingest_settings=dict(ingest_settings),
            )
    _warn_on_model_drift(models, settings.EMBEDDING_MODEL)
    written = bundle.result
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "exported %d document(s) and %d chunk(s) to %s (%d bytes) in %.1fs",
        documents,
        chunks,
        written.name,
        written.size_bytes,
        wall_clock_s,
    )
    return RagExport(bundle=written, wall_clock_s=wall_clock_s)

exports_dir(corpus)

The reserved folder a corpus's bundles live in.

Parameters:

Name	Type	Description	Default
corpus	Corpus	The corpus.	required

Returns:

Type	Description
Path	<corpus root>/varagity-exports (it may not exist yet).

Source code in varagity/backup/bundle.py

def exports_dir(corpus: Corpus) -> Path:
    """The reserved folder a corpus's bundles live in.

    Args:
        corpus: The corpus.

    Returns:
        ``<corpus root>/varagity-exports`` (it may not exist yet).
    """
    return corpus_root(corpus) / EXPORTS_DIRNAME

graph_counts(workdir)

What a graph bundle holds, by the workdir's own size summary.

The summary sidecar is the graph's size cache (what the status route and the gauges read) — never the graphml, which a count would have to parse.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory.	required

Returns:

Type	Description
dict[str, int]	thread_days (transcript documents) and messages (distinct
dict[str, int]	guids), plus entities and relations when the graph reported
dict[str, int]	them — or nothing at all when no build has written a summary yet (a
dict[str, int]	first build killed partway leaves files, but no summary).

Source code in varagity/backup/graph.py

def graph_counts(workdir: Path) -> dict[str, int]:
    """What a graph bundle holds, by the workdir's own size summary.

    The summary sidecar is the graph's size cache (what the status route and
    the gauges read) — never the graphml, which a count would have to parse.

    Args:
        workdir: The engine's working directory.

    Returns:
        ``thread_days`` (transcript documents) and ``messages`` (distinct
        guids), plus ``entities`` and ``relations`` when the graph reported
        them — or nothing at all when no build has written a summary yet (a
        first build killed partway leaves files, but no summary).
    """
    summary = load_summary(workdir)
    if summary is None:
        return {}
    counts = {"thread_days": summary.docs, "messages": summary.message_guids}
    if summary.entities is not None:
        counts["entities"] = summary.entities
    if summary.relations is not None:
        counts["relations"] = summary.relations
    return counts

graph_folder(manifest)

The folder a graph bundle's files ride under: the name of the engine that wrote them.

Shared by the verdict's member rule and the restore, which extracts <folder>/<file> into the workdir as <file>.

Parameters:

Name	Type	Description	Default
manifest	BundleManifest	The graph bundle's manifest.	required

Returns:

Type	Description
str	The engine's registry name (lightrag).

Raises:

Type	Description
BundleInvalid	If the manifest records no engine, or one whose name couldn't be a folder.

Source code in varagity/backup/bundle.py

def graph_folder(manifest: BundleManifest) -> str:
    """The folder a graph bundle's files ride under: the name of the engine that wrote them.

    Shared by the verdict's member rule and the restore, which extracts
    ``<folder>/<file>`` into the workdir as ``<file>``.

    Args:
        manifest: The graph bundle's manifest.

    Returns:
        The engine's registry name (``lightrag``).

    Raises:
        BundleInvalid: If the manifest records no engine, or one whose name
            couldn't be a folder.
    """
    engine = manifest.engine
    if engine is None:
        raise BundleInvalid(f"{MANIFEST_NAME} does not record the graph engine that wrote it.")
    if not is_graph_file_name(engine.name):
        raise BundleInvalid(
            f"{MANIFEST_NAME} records the graph engine {engine.name!r}, which is not a folder name."
        )
    return engine.name

ingest_drift(manifest, current)

Name the reingest-affecting settings a RAG bundle differs from this stack on.

Parameters:

Name	Type	Description	Default
manifest	BundleManifest	The bundle's manifest.	required
current	Mapping[str, str]	This stack's reingest-affecting settings, in env-string form (varagity.api.backup_runner.current_ingest_settings).	required

Returns:

Type	Description
list[str]	The differing setting names, sorted — a setting recorded on only one
list[str]	side counts as differing, and a bundle that recorded no settings
list[str]	differs on all of them.

Source code in varagity/backup/rag.py

def ingest_drift(manifest: BundleManifest, current: Mapping[str, str]) -> list[str]:
    """Name the reingest-affecting settings a RAG bundle differs from this stack on.

    Args:
        manifest: The bundle's manifest.
        current: This stack's reingest-affecting settings, in env-string
            form (``varagity.api.backup_runner.current_ingest_settings``).

    Returns:
        The differing setting names, sorted — a setting recorded on only one
        side counts as differing, and a bundle that recorded no settings
        differs on all of them.
    """
    recorded = manifest.ingest_settings
    if recorded is None:
        return sorted(current)
    return sorted(
        name for name in set(recorded) | set(current) if recorded.get(name) != current.get(name)
    )

inspect_bundle(path, corpus)

The cheap verdict the backups list shows for one bundle.

Reads the zip's central directory and bundle.json — never a data member — and checks everything those can tell: the format version, the corpus, the member allowlist and zip-slip rules, each member's recorded size against the zip's, the embedding model and dimension against this stack's settings, and for a graph bundle the engine that wrote its files against this stack's (:func:running_engine_format). What it can't see is a member damaged inside an intact zip with unchanged sizes: the restore's verifying streams (:func:open_bundle) — or :func:verify_bundle, standalone — hash for that.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required
corpus	Corpus	The corpus it would restore.	required

Returns:

Type	Description
BundleVerdict	The verdict (it never raises for a bad file — the problem rides
BundleVerdict	the verdict).

Source code in varagity/backup/bundle.py

def inspect_bundle(path: Path, corpus: Corpus) -> BundleVerdict:
    """The cheap verdict the backups list shows for one bundle.

    Reads the zip's central directory and ``bundle.json`` — never a data
    member — and checks everything those can tell: the format version, the
    corpus, the member allowlist and zip-slip rules, each member's recorded
    size against the zip's, the embedding model and dimension against this
    stack's settings, and for a graph bundle the engine that wrote its files
    against this stack's (:func:`running_engine_format`). What it can't see
    is a member damaged *inside* an intact zip with unchanged sizes: the
    restore's verifying streams (:func:`open_bundle`) — or
    :func:`verify_bundle`, standalone — hash for that.

    Args:
        path: The bundle file.
        corpus: The corpus it would restore.

    Returns:
        The verdict (it never raises for a bad file — the problem rides
        the verdict).
    """
    try:
        with zipfile.ZipFile(path) as archive:
            return _inspect_archive(archive, corpus)
    except _UNREADABLE_ZIP as error:
        return BundleVerdict(manifest=None, problem=_unreadable(error))

is_bundle_name(name)

Report whether a file name is a listable, servable bundle name.

Parameters:

Name	Type	Description	Default
name	str	A bare file name (no directories).	required

Returns:

Type	Description
bool	True for a visible *.zip name — never a hidden file, so an
bool	in-flight .{name}.partial (or anything else dot-prefixed) is
bool	never listed or served.

Source code in varagity/backup/bundle.py

def is_bundle_name(name: str) -> bool:
    """Report whether a file name is a listable, servable bundle name.

    Args:
        name: A bare file name (no directories).

    Returns:
        ``True`` for a visible ``*.zip`` name — never a hidden file, so an
        in-flight ``.{name}.partial`` (or anything else dot-prefixed) is
        never listed or served.
    """
    return (
        bool(name)
        and not name.startswith(".")
        and "/" not in name
        and "\\" not in name
        and name.lower().endswith(BUNDLE_SUFFIX)
    )

is_graph_file_name(name)

Report whether a working-directory file name may ride in a graph bundle.

The graph bundle's member rule (and the rule for the engine folder those members sit under), shared by the export — which must never write a member the restore's verdict would refuse — and the verdict itself.

Parameters:

Name	Type	Description	Default
name	str	A bare file (or folder) name.	required

Returns:

Type	Description
bool	True for a visible name of at most 255 ASCII letters, digits,
bool	_, . and - — every name an engine writes, and never one
bool	that could be a path, a hidden file, . or ...

Source code in varagity/backup/bundle.py

def is_graph_file_name(name: str) -> bool:
    """Report whether a working-directory file name may ride in a graph bundle.

    The graph bundle's member rule (and the rule for the engine folder those
    members sit under), shared by the export — which must never write a
    member the restore's verdict would refuse — and the verdict itself.

    Args:
        name: A bare file (or folder) name.

    Returns:
        ``True`` for a visible name of at most 255 ASCII letters, digits,
        ``_``, ``.`` and ``-`` — every name an engine writes, and never one
        that could be a path, a hidden file, ``.`` or ``..``.
    """
    return _GRAPH_FILE_NAME.fullmatch(name) is not None

list_bundles(directory, corpus)

List the bundles in an exports folder, newest first, each with its verdict.

Hidden files (in-flight partials included) and non-.zip files are skipped, and so is anything that resolves outside the folder (a symlink pointing elsewhere is not a bundle this folder holds). A visible zip that isn't a readable bundle is still listed — with no manifest and its reason — because the user put it there and the panel should say what it is.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder (a missing folder lists nothing).	required
corpus	Corpus	The corpus the folder belongs to (what each bundle is judged against).	required

Returns:

Type	Description
list[BundleListing]	One listing per bundle file, newest mtime first.

Source code in varagity/backup/bundle.py

def list_bundles(directory: Path, corpus: Corpus) -> list[BundleListing]:
    """List the bundles in an exports folder, newest first, each with its verdict.

    Hidden files (in-flight partials included) and non-``.zip`` files
    are skipped, and so is anything that resolves outside the folder (a
    symlink pointing elsewhere is not a bundle this folder holds). A visible
    zip that isn't a readable bundle is still listed — with no manifest and
    its reason — because the user put it there and the panel should say what
    it is.

    Args:
        directory: The exports folder (a missing folder lists nothing).
        corpus: The corpus the folder belongs to (what each bundle is judged
            against).

    Returns:
        One listing per bundle file, newest mtime first.
    """
    if not directory.is_dir():
        return []
    root = directory.resolve()
    listings: list[BundleListing] = []
    for entry in directory.iterdir():
        if not is_bundle_name(entry.name):
            continue
        resolved = resolve_contained(entry, root)
        if resolved is None or not resolved.is_file():
            continue
        try:
            info = resolved.stat()
        except OSError:
            continue  # vanished between the listing and the stat
        verdict = inspect_bundle(resolved, corpus)
        if verdict.problem is not None:
            logger.debug("listing %s as not restorable: %s", entry.name, verdict.reason)
        listings.append(
            BundleListing(
                name=entry.name,
                path=resolved,
                size_bytes=info.st_size,
                modified_at=datetime.fromtimestamp(info.st_mtime, tz=UTC),
                mtime_ns=info.st_mtime_ns,
                verdict=verdict,
            )
        )
    listings.sort(key=lambda listing: (listing.modified_at, listing.name), reverse=True)
    return listings

open_bundle(path, manifest)

Open a bundle to read its members against its manifest.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required
manifest	BundleManifest	Its manifest (from :func:inspect_bundle or :func:verify_bundle).	required

Yields:

Type	Description
BundleSource	The open bundle.

Raises:

Type	Description
BundleInvalid	If the file can't be opened as a zip.

Source code in varagity/backup/bundle.py

@contextmanager
def open_bundle(path: Path, manifest: BundleManifest) -> Iterator[BundleSource]:
    """Open a bundle to read its members against its manifest.

    Args:
        path: The bundle file.
        manifest: Its manifest (from :func:`inspect_bundle` or
            :func:`verify_bundle`).

    Yields:
        The open bundle.

    Raises:
        BundleInvalid: If the file can't be opened as a zip.
    """
    try:
        archive = zipfile.ZipFile(path)
    except _UNREADABLE_ZIP as error:
        raise _unreadable(error) from error
    with archive:
        yield BundleSource(archive, manifest)

read_manifest(path)

Read a bundle's manifest — and only its manifest.

The zip's central directory locates bundle.json; no other member is opened, so listing a folder of large bundles stays cheap.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required

Returns:

Type	Description
BundleManifest	The validated manifest.

Raises:

Type	Description
BundleInvalid	If the file is not a readable zip, has no bundle.json, or its manifest is not a varagity manifest.
BundleIncompatible	If the manifest is of another format version.

Source code in varagity/backup/bundle.py

def read_manifest(path: Path) -> BundleManifest:
    """Read a bundle's manifest — and only its manifest.

    The zip's central directory locates ``bundle.json``; no other member is
    opened, so listing a folder of large bundles stays cheap.

    Args:
        path: The bundle file.

    Returns:
        The validated manifest.

    Raises:
        BundleInvalid: If the file is not a readable zip, has no
            ``bundle.json``, or its manifest is not a varagity manifest.
        BundleIncompatible: If the manifest is of another format version.
    """
    try:
        with zipfile.ZipFile(path) as archive:
            raw = _manifest_bytes(archive)
    except BundleError as error:
        raise type(error)(f"{path.name}: {error}") from error
    except _UNREADABLE_ZIP as error:
        raise BundleInvalid(f"{path.name}: {_unreadable(error)}") from error
    try:
        return _parse_manifest(raw)
    except BundleError as error:
        raise type(error)(f"{path.name}: {error}") from error

restore_graph(service, bundle, *, run_id, on_progress=None)

Replace the message graph's working directory with a bundle's.

No model is called and nothing is re-extracted — the bundle is a workdir:

1. validate — the cheap verdict (the route checked it before the run; this guards a direct call and a file swapped since);
2. extract — every file into a fresh staging folder beside the workdir (:meth:~varagity.graph.service.GraphService.staging_dir, on the same volume), each member's sha256 checked as it is read — the restore's one read of the bundle, and its only hash — and each file synced to disk, so a damaged bundle can never go live;
3. swap and open — :meth:~varagity.graph.service.GraphService.replace_workdir closes the session, swaps the folders under the service's write and open locks, and opens the restored graph; if it won't open, the previous graph is put back.

Whatever is left of the staging folder is removed however the restore ends; until the swap, the live graph is untouched.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (the workdir's single writer).	required
bundle	Path	The bundle file.	required
run_id	str	The restore run's id (names the folders beside the workdir).	required
on_progress	GraphRestoreProgressCallback | None	Called with (stage, current, total) per stage; extract counts bytes.	None

Returns:

Type	Description
GraphRestore	What was restored, and how long it took.

Raises:

Type	Description
BundleInvalid	If the bundle is unsound or damaged (the live graph is untouched).
BundleIncompatible	If this stack must not restore it (untouched).
GraphBuildInProgress	If a write holds the graph's write lock at the swap (untouched).
GraphUnavailable	If the restored graph would not open (the previous graph is back in place).
OSError	If a file can't be extracted or a folder can't be moved (the previous graph stays, or is put back).

Source code in varagity/backup/graph.py

def restore_graph(
    service: GraphService,
    bundle: Path,
    *,
    run_id: str,
    on_progress: GraphRestoreProgressCallback | None = None,
) -> GraphRestore:
    """Replace the message graph's working directory with a bundle's.

    No model is called and nothing is re-extracted — the bundle *is* a
    workdir:

    1. ``validate`` — the cheap verdict (the route checked it before the
       run; this guards a direct call and a file swapped since);
    2. ``extract`` — every file into a fresh staging folder beside the
       workdir (:meth:`~varagity.graph.service.GraphService.staging_dir`, on
       the same volume), each member's sha256 checked as it is read — the
       restore's one read of the bundle, and its only hash — and each file
       synced to disk, so a damaged bundle can never go live;
    3. ``swap`` and ``open`` —
       :meth:`~varagity.graph.service.GraphService.replace_workdir` closes
       the session, swaps the folders under the service's write and open
       locks, and opens the restored graph; if it won't open, the previous
       graph is put back.

    Whatever is left of the staging folder is removed however the restore
    ends; until the swap, the live graph is untouched.

    Args:
        service: The process's graph service (the workdir's single writer).
        bundle: The bundle file.
        run_id: The restore run's id (names the folders beside the workdir).
        on_progress: Called with ``(stage, current, total)`` per stage;
            ``extract`` counts bytes.

    Returns:
        What was restored, and how long it took.

    Raises:
        BundleInvalid: If the bundle is unsound or damaged (the live graph
            is untouched).
        BundleIncompatible: If this stack must not restore it (untouched).
        varagity.graph.service.GraphBuildInProgress: If a write holds the
            graph's write lock at the swap (untouched).
        varagity.graph.service.GraphUnavailable: If the restored graph would
            not open (the previous graph is back in place).
        OSError: If a file can't be extracted or a folder can't be moved
            (the previous graph stays, or is put back).
    """
    report: GraphRestoreProgressCallback = on_progress if on_progress is not None else _ignore
    started = time.perf_counter()

    report("validate", None, None)
    manifest = inspect_bundle(bundle, "graph").manifest_or_raise()
    folder = f"{graph_folder(manifest)}/"
    engine = manifest.engine
    size_bytes = bundle.stat().st_size
    total = sum(member.size_bytes for member in manifest.members)
    # File names only, as the export logs: these lines are relayed into the
    # panel, where an in-container path would read as nonsense.
    logger.info(
        "restoring %d graph file(s), %s bytes, written by %s, from %s",
        len(manifest.members),
        f"{total:,}",
        f"{engine.package} {engine.version}" if engine is not None else "an unknown engine",
        bundle.name,
    )
    staged = service.staging_dir(run_id)
    staged.mkdir(parents=True)  # a fresh folder, never one that already holds files
    try:
        progress = _ByteProgress(report, total, stage="extract")
        with open_bundle(bundle, manifest) as source:
            for member in manifest.members:
                # The verdict held every member to "<folder><flat safe name>".
                target = staged / member.name.removeprefix(folder)
                with source.member(member.name) as stream, target.open("xb") as out:
                    while block := stream.read(_COPY_BYTES):
                        out.write(block)
                        progress.advance(len(block))
                    out.flush()
                    os.fsync(out.fileno())
        progress.finish()
        report("swap", None, None)
        try:
            stats = service.replace_workdir(
                staged, run_id, on_open=lambda: report("open", None, None)
            )
        except GraphUnavailable:
            logger.warning(
                "the restored graph would not open, so the restore was undone — the graph is "
                "as it was before"
            )
            raise
    finally:
        # Gone already after a swap; a failed restore's extracted files otherwise.
        shutil.rmtree(staged, ignore_errors=True)
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "restored %d graph file(s) from %s in %.1fs",
        len(manifest.members),
        bundle.name,
        wall_clock_s,
    )
    return GraphRestore(
        bundle_name=bundle.name,
        size_bytes=size_bytes,
        manifest=manifest,
        files=len(manifest.members),
        stats=stats,
        wall_clock_s=wall_clock_s,
    )

restore_rag(bundle, store, bm25, docs_root, *, on_progress=None)

Replace the chunk corpus with a bundle's rows, then rebuild the BM25 index.

No model is called — nothing is parsed, chunked, contextualized or embedded. Under the corpus's exclusive advisory lock, held until the BM25 rebuild ends (so no ingest in any process can write rows the rebuild would miss, or touch the index while it is gone):

1. validate — the cheap verdict (the route checked it before the run; this guards a direct call and a file swapped since).
2. load — one transaction empties both tables and streams the bundle's lines into binary COPY, re-rooting each source; every member's sha256 is checked as it is read — the restore's one read of the bundle, and its only hash — so a damaged bundle (or one a line of which no row can be made from) can never commit.
3. index — the HNSW index is rebuilt and the transaction commits: the restore has taken effect. Any failure before this rolled back, leaving the previous corpus untouched.
4. bm25 — the index is recreated and refilled from the restored rows.

Parameters:

Name	Type	Description	Default
bundle	Path	The bundle file.	required
store	ContextualVectorDB	The vector store to replace (its connection holds the lock).	required
bm25	ElasticsearchBM25	The BM25 store to rebuild.	required
docs_root	Path	The restoring stack's DOCS_PATH (re-rooting target).	required
on_progress	RestoreProgressCallback | None	Called with (stage, current, total) per stage; the load and bm25 stages count chunks.	None

Returns:

Type	Description
RagRestore	What was restored, and how long it took.

Raises:

Type	Description
BundleInvalid	If the bundle is unsound or damaged (nothing was replaced — the transaction rolled back).
BundleIncompatible	If this stack must not restore it (nothing was replaced).
CorpusLocked	If an ingest holds the corpus lock (nothing was replaced).
BM25RebuildFailed	If the rows were restored but the BM25 index could not be rebuilt.

Source code in varagity/backup/rag.py

def restore_rag(
    bundle: Path,
    store: ContextualVectorDB,
    bm25: ElasticsearchBM25,
    docs_root: Path,
    *,
    on_progress: RestoreProgressCallback | None = None,
) -> RagRestore:
    """Replace the chunk corpus with a bundle's rows, then rebuild the BM25 index.

    No model is called — nothing is parsed, chunked, contextualized or
    embedded. Under the corpus's **exclusive** advisory lock, held until the
    BM25 rebuild ends (so no ingest in any process can write rows the
    rebuild would miss, or touch the index while it is gone):

    1. ``validate`` — the cheap verdict (the route checked it before the
       run; this guards a direct call and a file swapped since).
    2. ``load`` — one transaction empties both tables and streams the
       bundle's lines into binary ``COPY``, re-rooting each ``source``;
       every member's sha256 is checked as it is read — the restore's one
       read of the bundle, and its only hash — so a damaged bundle (or one
       a line of which no row can be made from) can never commit.
    3. ``index`` — the HNSW index is rebuilt and the transaction commits:
       the restore has taken effect. Any failure before this rolled back,
       leaving the previous corpus untouched.
    4. ``bm25`` — the index is recreated and refilled from the restored
       rows.

    Args:
        bundle: The bundle file.
        store: The vector store to replace (its connection holds the lock).
        bm25: The BM25 store to rebuild.
        docs_root: The restoring stack's ``DOCS_PATH`` (re-rooting target).
        on_progress: Called with ``(stage, current, total)`` per stage; the
            ``load`` and ``bm25`` stages count chunks.

    Returns:
        What was restored, and how long it took.

    Raises:
        BundleInvalid: If the bundle is unsound or damaged (nothing was
            replaced — the transaction rolled back).
        BundleIncompatible: If this stack must not restore it (nothing was
            replaced).
        varagity.stores.vector_store.CorpusLocked: If an ingest holds the
            corpus lock (nothing was replaced).
        BM25RebuildFailed: If the rows were restored but the BM25 index
            could not be rebuilt.
    """
    settings = get_settings()
    report: RestoreProgressCallback = on_progress if on_progress is not None else _ignore
    dim = settings.EMBEDDING_DIM
    root = docs_root.resolve()
    started = time.perf_counter()

    report("validate", None, None)
    manifest = inspect_bundle(bundle, "rag").manifest_or_raise()
    size_bytes = bundle.stat().st_size
    expected_documents = manifest.counts["documents"]
    expected_chunks = manifest.counts["chunks"]
    logger.info(
        "restoring %d document(s) and %d chunk(s) from %s",
        expected_documents,
        expected_chunks,
        bundle.name,
    )
    with store.corpus_lock(exclusive=True):
        sources: dict[str, str] = {}
        with open_bundle(bundle, manifest) as source:
            documents, chunks = store.replace_corpus(
                _document_rows(source, root, sources, expected_documents),
                _chunk_rows(
                    source,
                    sources,
                    dim=dim,
                    expected=expected_chunks,
                    progress=_Progress(report, "load", expected_chunks),
                ),
                on_indexing=lambda: report("index", None, None),
            )
        logger.info("loaded %d document(s) and %d chunk(s) — committed", documents, chunks)
        progress = _Progress(report, "bm25", chunks)
        try:
            with closing(store.iter_index_rows()) as rows:
                indexed = bm25.rebuild(rows, on_indexed=progress.tick)
        except Exception as error:
            raise BM25RebuildFailed(
                f"the rows are restored, but the BM25 index could not be rebuilt "
                f"({type(error).__name__}: {error}) — restore the same bundle again to "
                "repair it",
                manifest,
            ) from error
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "restored %d document(s) and %d chunk(s) from %s in %.1fs; BM25 holds %d",
        documents,
        chunks,
        bundle.name,
        wall_clock_s,
        indexed,
    )
    return RagRestore(
        bundle_name=bundle.name,
        size_bytes=size_bytes,
        manifest=manifest,
        documents=documents,
        chunks=chunks,
        bm25_indexed=indexed,
        wall_clock_s=wall_clock_s,
    )

restored_source(source, relative_path, docs_root)

Re-root one document's source onto the restoring stack's corpus directory.

doc_id already hashes the corpus-relative path, so the row's identity is portable; source (the absolute path the readers resolve — the document list, previews, "also remove the file") is rebuilt the way ingest builds it, (DOCS_PATH / relative_path).resolve(). A document exported from outside the corpus has no relative path and keeps its original absolute source.

Parameters:

Name	Type	Description	Default
source	str	The source recorded at export.	required
relative_path	str | None	Its path relative to the exporting DOCS_PATH, or None.	required
docs_root	Path	The resolved DOCS_PATH of the restoring stack.	required

Returns:

Type	Description
str	The source to restore.

Raises:

Type	Description
BundleInvalid	If the relative path is absolute, empty, or climbs out of the corpus (..) — no export writes one.

Source code in varagity/backup/rag.py

def restored_source(source: str, relative_path: str | None, docs_root: Path) -> str:
    """Re-root one document's ``source`` onto the restoring stack's corpus directory.

    ``doc_id`` already hashes the corpus-relative path, so the row's
    identity is portable; ``source`` (the absolute path the readers resolve
    — the document list, previews, "also remove the file") is rebuilt the
    way ingest builds it, ``(DOCS_PATH / relative_path).resolve()``. A
    document exported from outside the corpus has no relative path and
    keeps its original absolute ``source``.

    Args:
        source: The ``source`` recorded at export.
        relative_path: Its path relative to the exporting ``DOCS_PATH``, or
            ``None``.
        docs_root: The resolved ``DOCS_PATH`` of the restoring stack.

    Returns:
        The ``source`` to restore.

    Raises:
        BundleInvalid: If the relative path is absolute, empty, or climbs
            out of the corpus (``..``) — no export writes one.
    """
    if relative_path is None:
        return source
    relative = PurePosixPath(relative_path)
    if relative.is_absolute() or not relative.parts or ".." in relative.parts:
        raise BundleInvalid(
            f"{DOCUMENTS_MEMBER} records relative_path {relative_path!r}, which is not a path "
            "inside the corpus."
        )
    return str(docs_root.joinpath(*relative.parts).resolve())

running_engine_format()

The file format this stack's graph engine reads — what a graph bundle is held to.

The engine reports its own format (GraphEngine.storage_format(), read from package metadata, so the engine library is never imported), which keeps the backup code from hardcoding any library's name or version.

Returns:

Type	Description
EngineFormat	GRAPH_ENGINE's registered engine's format.

Raises:

Type	Description
KeyError	If GRAPH_ENGINE names no registered engine (the config validator makes this unreachable).

Source code in varagity/backup/bundle.py

def running_engine_format() -> EngineFormat:
    """The file format this stack's graph engine reads — what a graph bundle is held to.

    The engine reports its own format (``GraphEngine.storage_format()``, read
    from package metadata, so the engine library is never imported), which
    keeps the backup code from hardcoding any library's name or version.

    Returns:
        ``GRAPH_ENGINE``'s registered engine's format.

    Raises:
        KeyError: If ``GRAPH_ENGINE`` names no registered engine (the config
            validator makes this unreachable).
    """
    # Here rather than at the top: importing the engines package registers
    # its adapters, which this module — imported by the API schemas — needs
    # only once a graph bundle is actually judged.
    from varagity.graph.engines import get_graph_engine

    return get_graph_engine(get_settings().GRAPH_ENGINE).storage_format()

verify_bundle(path, corpus)

Run every check a restore makes, without restoring anything.

:func:inspect_bundle's verdict, then the sha256 of every member — every byte of the bundle is read once. A restore makes the same checks as it loads (through :func:open_bundle's verifying streams, so it reads the bundle only once); this is the standalone form, for a caller that wants the full verdict without a restore.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required
corpus	Corpus	The corpus it would restore.	required

Returns:

Type	Description
BundleManifest	The bundle's manifest, now fully verified.

Raises:

Type	Description
BundleIncompatible	If this stack must not restore the bundle.
BundleDamaged	If a member's bytes don't match its recorded sha256.
BundleInvalid	If the bundle is otherwise unsound.

Source code in varagity/backup/bundle.py

def verify_bundle(path: Path, corpus: Corpus) -> BundleManifest:
    """Run every check a restore makes, without restoring anything.

    :func:`inspect_bundle`'s verdict, then the sha256 of every member —
    every byte of the bundle is read once. A restore makes the same checks
    as it loads (through :func:`open_bundle`'s verifying streams, so it
    reads the bundle only once); this is the standalone form, for a caller
    that wants the full verdict without a restore.

    Args:
        path: The bundle file.
        corpus: The corpus it would restore.

    Returns:
        The bundle's manifest, now fully verified.

    Raises:
        BundleIncompatible: If this stack must not restore the bundle.
        BundleDamaged: If a member's bytes don't match its recorded sha256.
        BundleInvalid: If the bundle is otherwise unsound.
    """
    manifest = inspect_bundle(path, corpus).manifest_or_raise()
    with open_bundle(path, manifest) as source:
        for member in manifest.members:
            # Opening and closing a member is the whole check: the clean
            # exit drains it through the hash and compares the digest.
            with source.member(member.name):
                pass
    return manifest

workdir_files(workdir)

List the files a graph export would copy.

What the export route's nothing_to_export check asks — the same rule the export applies, so the route and the run can never disagree on whether there is anything to back up.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory (a missing one holds none).	required

Returns:

Type	Description
list[Path]	The workdir's own files, in name order, temp files and anything a
list[Path]	bundle can't carry left out.

Source code in varagity/backup/graph.py

def workdir_files(workdir: Path) -> list[Path]:
    """List the files a graph export would copy.

    What the export route's ``nothing_to_export`` check asks — the same
    rule the export applies, so the route and the run can never disagree on
    whether there is anything to back up.

    Args:
        workdir: The engine's working directory (a missing one holds none).

    Returns:
        The workdir's own files, in name order, temp files and anything a
        bundle can't carry left out.
    """
    return _workdir_entries(workdir)[0]

write_bundle(directory, corpus, *, created_at=None)

Write a new bundle into directory, all-or-nothing.

Sweeps this corpus's stale partials, chooses a free name (:func:bundle_name), and opens .{name}.partial. When the with block exits cleanly and the writer was sealed, the partial is synced and renamed into place with os.replace — same directory, so the rename is atomic and a listing sees the whole bundle or nothing. Any exception, or an unsealed writer, deletes the partial instead.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder (created if missing).	required
corpus	Corpus	The corpus being backed up.	required
created_at	datetime | None	The export's instant (defaults to now).	None

Yields:

Type	Description
BundleWriter	The writer to fill; read :attr:BundleWriter.result after the block.

Raises:

Type	Description
RuntimeError	If the block exits without sealing the manifest (the partial is removed; nothing is published).
OSError	If the folder or the bundle can't be written.

Source code in varagity/backup/bundle.py

@contextmanager
def write_bundle(
    directory: Path, corpus: Corpus, *, created_at: datetime | None = None
) -> Iterator[BundleWriter]:
    """Write a new bundle into ``directory``, all-or-nothing.

    Sweeps this corpus's stale partials, chooses a free name
    (:func:`bundle_name`), and opens ``.{name}.partial``. When the ``with``
    block exits cleanly *and* the writer was sealed, the partial is synced
    and renamed into place with ``os.replace`` — same directory, so the
    rename is atomic and a listing sees the whole bundle or nothing. Any
    exception, or an unsealed writer, deletes the partial instead.

    Args:
        directory: The exports folder (created if missing).
        corpus: The corpus being backed up.
        created_at: The export's instant (defaults to now).

    Yields:
        The writer to fill; read :attr:`BundleWriter.result` after the block.

    Raises:
        RuntimeError: If the block exits without sealing the manifest (the
            partial is removed; nothing is published).
        OSError: If the folder or the bundle can't be written.
    """
    created = (created_at or datetime.now(UTC)).astimezone(UTC)
    directory.mkdir(parents=True, exist_ok=True)
    sweep_partials(directory, corpus)
    name = bundle_name(directory, corpus, now=created)
    partial = directory / f".{name}{PARTIAL_SUFFIX}"
    try:
        # "x": never clobber a partial some other writer owns.
        with zipfile.ZipFile(partial, mode="x", compression=zipfile.ZIP_DEFLATED) as archive:
            writer = BundleWriter(archive, corpus, created)
            yield writer
            manifest = writer.manifest
            if manifest is None:
                raise RuntimeError(
                    "the bundle was never sealed — a bundle without its manifest is not published"
                )
        _fsync(partial)
        final = directory / name
        if final.exists(follow_symlinks=False):
            # Something claimed the name while we wrote — never replace it.
            final = directory / bundle_name(directory, corpus, now=created)
        os.replace(partial, final)
    except BaseException:
        partial.unlink(missing_ok=True)
        raise
    fsync_directory(directory)
    size = final.stat().st_size
    writer._publish(WrittenBundle(name=final.name, path=final, size_bytes=size, manifest=manifest))
    # DEBUG: the corpus export logs its own one-line summary at INFO.
    logger.debug("published %s (%d bytes)", final.name, size)

varagity.backup.bundle

The backup bundle: one zip per backup, its manifest written last (ADR-019).

A bundle is an ordinary .zip in a corpus root's reserved varagity-exports/ folder. Its data members stream in first, each hashed as it is written, and bundle.json — the :class:BundleManifest — is the last member, so it can record every data member's size and sha256. The zip's central directory then lets :func:read_manifest open that one member without touching the (possibly hundreds of megabytes of) rest: listing the backups panel costs a directory read, not an unpack.

Writing never exposes a half-made file. :func:write_bundle writes into a hidden .{name}.partial sibling and renames it into place (os.replace) only once the manifest is sealed and the bytes are synced; any failure removes the partial instead, and the next export sweeps whatever a crash left behind. Listings skip hidden files, so a bundle is either whole or invisible.

Reading has two depths, because hashing is the expensive part:

• :func:inspect_bundle is the cheap verdict every listed bundle gets — the zip's central directory and bundle.json alone: the format version, the corpus, the member allowlist and safety rules, each member's recorded size against the zip's, the embedding model and dimension against this stack's, and for a graph bundle the engine format its files are in. No data member is opened, so a folder of large bundles still lists in milliseconds.
• :func:open_bundle hands out verifying streams: every byte a restore loads passes through a member's sha256, which is checked when the member is drained — the only check that sees a bundle damaged inside an intact zip. A restore reads the bundle exactly once this way, and its writes only take effect after every member checked out (the RAG restore's transaction commits, the graph restore's staged folder is swapped in), so a damaged bundle can never restore. :func:verify_bundle runs the same check standalone — every member hashed, nothing loaded.

Two refusals come out of both: :class:BundleInvalid (not a sound bundle; :class:BundleDamaged when only its bytes are wrong) and :class:BundleIncompatible (a sound bundle this stack must not restore — another embedding space, or a graph engine format this stack's engine may not read — refused, never converted).

This module knows the container, not the corpora: what goes inside a bundle is :mod:varagity.backup.rag's and :mod:varagity.backup.graph's business — except each corpus's member allowlist, which the verdict needs — and nothing here knows about HTTP. The two allowlists differ in kind: a RAG bundle holds exactly three named members, while a graph bundle holds whatever files its engine's working directory held, so its rule is a pattern — flat files with safe names, under one folder named for the engine (lightrag/…) — and its manifest records that engine's file format (:attr:BundleManifest.engine).

BUNDLE_FORMAT = 'varagity-backup' module-attribute

The manifest's format marker: what makes a zip a varagity bundle.

BUNDLE_VERSION = 1 module-attribute

The bundle layout this code writes (and, in time, the versions it reads).

Corpus = Literal['rag', 'graph'] module-attribute

Which corpus a bundle backs up — ChatRequest.corpus's vocabulary.

MANIFEST_NAME = 'bundle.json' module-attribute

The manifest member — always the zip's last entry.

BundleDamaged

Bases: BundleInvalid

A member's bytes don't match what its bundle recorded.

The one defect the cheap verdict can't see: the zip and its manifest are intact, but a member's sha256 (or the zip's own CRC-32, or its deflate stream) says the content changed after it was written.

Source code in varagity/backup/bundle.py

class BundleDamaged(BundleInvalid):
    """A member's bytes don't match what its bundle recorded.

    The one defect the cheap verdict can't see: the zip and its manifest are
    intact, but a member's sha256 (or the zip's own CRC-32, or its deflate
    stream) says the content changed after it was written.
    """

BundleError

Bases: Exception

Base class for a bundle that can't be read or used.

Source code in varagity/backup/bundle.py

class BundleError(Exception):
    """Base class for a bundle that can't be read or used."""

BundleIncompatible

Bases: BundleError

A sound bundle this stack must not restore.

Another bundle format version, another corpus, vectors from another embedding model or dimension — or, for a graph bundle, files another graph engine (or another major.minor version of this one) wrote. Neither store checks the embedding model (pgvector compares only the vector size, LightRAG only embedding_dim), so restoring a mismatch would succeed and then retrieve the wrong passages — it is refused instead, never converted.

Source code in varagity/backup/bundle.py

class BundleIncompatible(BundleError):
    """A sound bundle this stack must not restore.

    Another bundle format version, another corpus, vectors from another
    embedding model or dimension — or, for a graph bundle, files another
    graph engine (or another major.minor version of this one) wrote. Neither
    store checks the embedding model (pgvector compares only the vector
    size, LightRAG only ``embedding_dim``), so restoring a mismatch would
    succeed and then retrieve the wrong passages — it is refused instead,
    never converted.
    """

BundleInvalid

Bases: BundleError

The file is not a readable varagity bundle (damaged, or not one at all).

Source code in varagity/backup/bundle.py

class BundleInvalid(BundleError):
    """The file is not a readable varagity bundle (damaged, or not one at all)."""

BundleListing dataclass

One bundle file found in an exports folder.

Attributes:

Name	Type	Description
name	str	The file name.
path	Path	Its resolved path.
size_bytes	int	Its size on disk.
modified_at	datetime	Its mtime (UTC).
mtime_ns	int	Its mtime in nanoseconds — with size_bytes, the identity a remembered restore failure is keyed by.
verdict	BundleVerdict	The cheap inspection's verdict for the listing's corpus.

Source code in varagity/backup/bundle.py

@dataclass(frozen=True)
class BundleListing:
    """One bundle file found in an exports folder.

    Attributes:
        name: The file name.
        path: Its resolved path.
        size_bytes: Its size on disk.
        modified_at: Its mtime (UTC).
        mtime_ns: Its mtime in nanoseconds — with ``size_bytes``, the
            identity a remembered restore failure is keyed by.
        verdict: The cheap inspection's verdict for the listing's corpus.
    """

    name: str
    path: Path
    size_bytes: int
    modified_at: datetime
    mtime_ns: int
    verdict: BundleVerdict

    @property
    def manifest(self) -> BundleManifest | None:
        """The bundle's manifest.

        Returns:
            The manifest, or ``None`` when the file isn't a readable
            varagity bundle (copied in by hand, damaged, another format …).
        """
        return self.verdict.manifest

manifest property

The bundle's manifest.

Returns:

Type	Description
BundleManifest | None	The manifest, or None when the file isn't a readable
BundleManifest | None	varagity bundle (copied in by hand, damaged, another format …).

BundleManifest

Bases: BaseModel

bundle.json: what a bundle holds and what it was made under.

Every later phase judges bundles against this record — a restore checks compatibility against it and derives the stale flag from it — so it carries the export-time facts no row records.

Attributes:

Name	Type	Description
format	Literal['varagity-backup']	Always "varagity-backup".
version	int	The bundle layout version (:data:BUNDLE_VERSION).
corpus	Corpus	Which corpus the bundle backs up.
created_at	datetime	When the export's snapshot was taken (UTC).
embedding_model	str	EMBEDDING_MODEL at export — the store checks only vector size, so the model name is the only guard against restoring vectors from another embedding space.
embedding_dim	int	EMBEDDING_DIM at export.
stale	bool	The corpus's stale flag at export (_corpus_stale or _graph_stale).
counts	dict[str, int]	What the bundle holds — rag: documents, chunks; graph: thread_days, messages, and the entities and relations the graph reported (each only when known).
ingest_settings	dict[str, str] | None	The reingest-affecting settings at export, in env-string form (rag only; None for a corpus without them).
engine	EngineFormat | None	The graph engine's file format at export — the workdir is copied verbatim, so it is only as portable as the engine version that wrote it (graph only; None for the RAG corpus).
members	list[BundleMember]	Every data member with its size and sha256, in write order.

Source code in varagity/backup/bundle.py

class BundleManifest(BaseModel):
    """``bundle.json``: what a bundle holds and what it was made under.

    Every later phase judges bundles against this record — a restore checks
    compatibility against it and derives the stale flag from it — so it
    carries the export-time facts no row records.

    Attributes:
        format: Always ``"varagity-backup"``.
        version: The bundle layout version (:data:`BUNDLE_VERSION`).
        corpus: Which corpus the bundle backs up.
        created_at: When the export's snapshot was taken (UTC).
        embedding_model: ``EMBEDDING_MODEL`` at export — the store checks
            only vector size, so the model name is the only guard against
            restoring vectors from another embedding space.
        embedding_dim: ``EMBEDDING_DIM`` at export.
        stale: The corpus's stale flag at export (``_corpus_stale`` or
            ``_graph_stale``).
        counts: What the bundle holds — rag: ``documents``, ``chunks``;
            graph: ``thread_days``, ``messages``, and the ``entities`` and
            ``relations`` the graph reported (each only when known).
        ingest_settings: The reingest-affecting settings at export, in
            env-string form (rag only; ``None`` for a corpus without them).
        engine: The graph engine's file format at export — the workdir is
            copied verbatim, so it is only as portable as the engine version
            that wrote it (graph only; ``None`` for the RAG corpus).
        members: Every data member with its size and sha256, in write order.
    """

    format: Literal["varagity-backup"]
    version: int
    corpus: Corpus
    created_at: datetime
    embedding_model: str
    embedding_dim: int
    stale: bool
    counts: dict[str, int]
    ingest_settings: dict[str, str] | None = None
    engine: EngineFormat | None = None
    members: list[BundleMember]

BundleMember

Bases: BaseModel

One data member of a bundle, as its manifest records it.

Attributes:

Name	Type	Description
name	str	The member's path inside the zip.
size_bytes	int	Its uncompressed size.
sha256	str	The hex sha256 of its uncompressed bytes — what a restore verifies before it touches a store.

Source code in varagity/backup/bundle.py

class BundleMember(BaseModel):
    """One data member of a bundle, as its manifest records it.

    Attributes:
        name: The member's path inside the zip.
        size_bytes: Its uncompressed size.
        sha256: The hex sha256 of its uncompressed bytes — what a restore
            verifies before it touches a store.
    """

    name: str
    size_bytes: int
    sha256: str

BundleSource

An open bundle whose members are read through verifying streams.

Attributes:

Name	Type	Description
manifest		The manifest the members are checked against.

Source code in varagity/backup/bundle.py

class BundleSource:
    """An open bundle whose members are read through verifying streams.

    Attributes:
        manifest: The manifest the members are checked against.
    """

    def __init__(self, archive: zipfile.ZipFile, manifest: BundleManifest) -> None:
        """Bind the open zip to the manifest it is held to.

        Args:
            archive: The open bundle.
            manifest: Its (already inspected) manifest.
        """
        self._archive = archive
        self._records = {member.name: member for member in manifest.members}
        self.manifest = manifest

    @contextmanager
    def member(self, name: str) -> Iterator[MemberStream]:
        """Open one data member for reading, verified when the block exits.

        Several members may be open at once (zipfile reads each through its
        own position in the shared file), which is how a restore reads the
        chunk lines and their vectors side by side.

        Args:
            name: The member's name.

        Yields:
            The member's verifying stream. When the ``with`` block exits
            cleanly, whatever the reader left is drained and the member's
            size and sha256 are checked against the manifest.

        Raises:
            BundleInvalid: If the manifest records no such member, or the zip
                can't open it.
            BundleDamaged: If the member's bytes don't match its record.
        """
        record = self._records.get(name)
        if record is None:
            raise BundleInvalid(f"Missing member(s): {name}.")
        try:
            raw = self._archive.open(name)
        except KeyError:
            raise BundleInvalid(f"Missing member(s): {name}.") from None
        except _UNREADABLE_ZIP as error:
            raise BundleInvalid(f"{name} could not be opened ({error}).") from error
        with raw:
            stream = MemberStream(raw, record)
            yield stream
            stream.verify()

__init__(archive, manifest)

Bind the open zip to the manifest it is held to.

Parameters:

Name	Type	Description	Default
archive	ZipFile	The open bundle.	required
manifest	BundleManifest	Its (already inspected) manifest.	required

Source code in varagity/backup/bundle.py

def __init__(self, archive: zipfile.ZipFile, manifest: BundleManifest) -> None:
    """Bind the open zip to the manifest it is held to.

    Args:
        archive: The open bundle.
        manifest: Its (already inspected) manifest.
    """
    self._archive = archive
    self._records = {member.name: member for member in manifest.members}
    self.manifest = manifest

member(name)

Open one data member for reading, verified when the block exits.

Several members may be open at once (zipfile reads each through its own position in the shared file), which is how a restore reads the chunk lines and their vectors side by side.

Parameters:

Name	Type	Description	Default
name	str	The member's name.	required

Yields:

Type	Description
MemberStream	The member's verifying stream. When the with block exits
MemberStream	cleanly, whatever the reader left is drained and the member's
MemberStream	size and sha256 are checked against the manifest.

Raises:

Type	Description
BundleInvalid	If the manifest records no such member, or the zip can't open it.
BundleDamaged	If the member's bytes don't match its record.

Source code in varagity/backup/bundle.py

@contextmanager
def member(self, name: str) -> Iterator[MemberStream]:
    """Open one data member for reading, verified when the block exits.

    Several members may be open at once (zipfile reads each through its
    own position in the shared file), which is how a restore reads the
    chunk lines and their vectors side by side.

    Args:
        name: The member's name.

    Yields:
        The member's verifying stream. When the ``with`` block exits
        cleanly, whatever the reader left is drained and the member's
        size and sha256 are checked against the manifest.

    Raises:
        BundleInvalid: If the manifest records no such member, or the zip
            can't open it.
        BundleDamaged: If the member's bytes don't match its record.
    """
    record = self._records.get(name)
    if record is None:
        raise BundleInvalid(f"Missing member(s): {name}.")
    try:
        raw = self._archive.open(name)
    except KeyError:
        raise BundleInvalid(f"Missing member(s): {name}.") from None
    except _UNREADABLE_ZIP as error:
        raise BundleInvalid(f"{name} could not be opened ({error}).") from error
    with raw:
        stream = MemberStream(raw, record)
        yield stream
        stream.verify()

BundleVerdict dataclass

What the cheap inspection concluded about one bundle.

Attributes:

Name	Type	Description
manifest	BundleManifest | None	The bundle's manifest, or None when none could be read (also for a format version this stack can't parse).
problem	BundleError | None	Why the bundle can't be restored, or None when it can.

Source code in varagity/backup/bundle.py

@dataclass(frozen=True)
class BundleVerdict:
    """What the cheap inspection concluded about one bundle.

    Attributes:
        manifest: The bundle's manifest, or ``None`` when none could be read
            (also for a format version this stack can't parse).
        problem: Why the bundle can't be restored, or ``None`` when it can.
    """

    manifest: BundleManifest | None
    problem: BundleError | None = None

    @property
    def restorable(self) -> bool:
        """Whether nothing the inspection could see refuses the bundle.

        Returns:
            ``True`` when there is no problem (the restore still hashes it).
        """
        return self.problem is None

    @property
    def reason(self) -> str | None:
        """The problem, worded for a person.

        Returns:
            The refusal's message, or ``None`` for a restorable bundle.
        """
        return None if self.problem is None else str(self.problem)

    def manifest_or_raise(self) -> BundleManifest:
        """Return the manifest of a restorable bundle, or raise its problem.

        Returns:
            The manifest.

        Raises:
            BundleInvalid: If the bundle is not a sound bundle.
            BundleIncompatible: If this stack must not restore it.
        """
        if self.problem is not None:
            raise self.problem
        if self.manifest is None:  # pragma: no cover — a verdict always pairs one or the other
            raise BundleInvalid("the bundle has no readable manifest")
        return self.manifest

reason property

The problem, worded for a person.

Returns:

Type	Description
str | None	The refusal's message, or None for a restorable bundle.

restorable property

Whether nothing the inspection could see refuses the bundle.

Returns:

Type	Description
bool	True when there is no problem (the restore still hashes it).

manifest_or_raise()

Return the manifest of a restorable bundle, or raise its problem.

Returns:

Type	Description
BundleManifest	The manifest.

Raises:

Type	Description
BundleInvalid	If the bundle is not a sound bundle.
BundleIncompatible	If this stack must not restore it.

Source code in varagity/backup/bundle.py

def manifest_or_raise(self) -> BundleManifest:
    """Return the manifest of a restorable bundle, or raise its problem.

    Returns:
        The manifest.

    Raises:
        BundleInvalid: If the bundle is not a sound bundle.
        BundleIncompatible: If this stack must not restore it.
    """
    if self.problem is not None:
        raise self.problem
    if self.manifest is None:  # pragma: no cover — a verdict always pairs one or the other
        raise BundleInvalid("the bundle has no readable manifest")
    return self.manifest

BundleWriter

Fills one bundle: data members first, then the sealed manifest.

Obtained from :func:write_bundle; :attr:result is available once its with block exits cleanly.

Attributes:

Name	Type	Description
corpus		The corpus being backed up.
created_at		The export's instant — the name's stamp and the manifest's created_at.

Source code in varagity/backup/bundle.py

class BundleWriter:
    """Fills one bundle: data members first, then the sealed manifest.

    Obtained from :func:`write_bundle`; :attr:`result` is available once its
    ``with`` block exits cleanly.

    Attributes:
        corpus: The corpus being backed up.
        created_at: The export's instant — the name's stamp and the
            manifest's ``created_at``.
    """

    def __init__(self, archive: zipfile.ZipFile, corpus: Corpus, created_at: datetime) -> None:
        """Bind the writer to its open partial archive.

        Args:
            archive: The partial zip, open for writing.
            corpus: The corpus being backed up.
            created_at: The export's instant (UTC).
        """
        self._archive = archive
        self.corpus = corpus
        self.created_at = created_at
        self._members: list[BundleMember] = []
        self._manifest: BundleManifest | None = None
        self._result: WrittenBundle | None = None
        # Zip stores a naive DOS timestamp; the UTC wall clock is the honest
        # one (the manifest carries the aware instant regardless).
        self._date_time = created_at.astimezone(UTC).timetuple()[:6]

    @contextmanager
    def member(self, name: str, *, compress: bool = True) -> Iterator[MemberSink]:
        """Stream one data member into the bundle.

        Args:
            name: The member's path inside the zip.
            compress: Deflate it (text) or store it as-is (already-dense
                bytes like a float32 matrix, which deflate can't shrink).

        Yields:
            The sink to write the member's bytes to; its size and sha256 are
            recorded for the manifest when the block exits.

        Raises:
            ValueError: If the name is the manifest's or already used.
            RuntimeError: If the manifest is already sealed.
        """
        if self._manifest is not None:
            raise RuntimeError("the manifest is sealed — no member may follow it")
        if name == MANIFEST_NAME or any(member.name == name for member in self._members):
            raise ValueError(f"bundle member {name!r} is reserved or already written")
        info = zipfile.ZipInfo(name, date_time=self._date_time)
        info.compress_type = zipfile.ZIP_DEFLATED if compress else zipfile.ZIP_STORED
        info.external_attr = _MEMBER_MODE
        # force_zip64: the size is unknown up front, and a corpus's vectors
        # can pass the 2 GiB a plain entry can describe.
        with self._archive.open(info, mode="w", force_zip64=True) as raw:
            sink = MemberSink(raw)
            yield sink
        self._members.append(
            BundleMember(name=name, size_bytes=sink.size_bytes, sha256=sink.hexdigest())
        )

    def seal(
        self,
        *,
        embedding_model: str,
        embedding_dim: int,
        stale: bool,
        counts: dict[str, int],
        ingest_settings: dict[str, str] | None = None,
        engine: EngineFormat | None = None,
    ) -> BundleManifest:
        """Write ``bundle.json`` as the bundle's last member.

        Args:
            embedding_model: ``EMBEDDING_MODEL`` at export.
            embedding_dim: ``EMBEDDING_DIM`` at export.
            stale: The corpus's stale flag at export.
            counts: What the bundle holds.
            ingest_settings: The reingest-affecting settings at export, if
                the corpus has them.
            engine: The graph engine's file format, if the corpus is the
                graph.

        Returns:
            The manifest, as written.

        Raises:
            RuntimeError: If the manifest is already sealed.
        """
        if self._manifest is not None:
            raise RuntimeError("the manifest is already sealed")
        manifest = BundleManifest(
            format=BUNDLE_FORMAT,
            version=BUNDLE_VERSION,
            corpus=self.corpus,
            created_at=self.created_at,
            embedding_model=embedding_model,
            embedding_dim=embedding_dim,
            stale=stale,
            counts=dict(counts),
            ingest_settings=dict(ingest_settings) if ingest_settings is not None else None,
            engine=engine,
            members=list(self._members),
        )
        info = zipfile.ZipInfo(MANIFEST_NAME, date_time=self._date_time)
        info.compress_type = zipfile.ZIP_DEFLATED
        info.external_attr = _MEMBER_MODE
        self._archive.writestr(info, manifest.model_dump_json(indent=2) + "\n")
        self._manifest = manifest
        return manifest

    @property
    def manifest(self) -> BundleManifest | None:
        """The sealed manifest (``None`` until :meth:`seal` runs).

        Returns:
            The manifest, or ``None``.
        """
        return self._manifest

    @property
    def result(self) -> WrittenBundle:
        """The published bundle.

        Returns:
            What :func:`write_bundle` moved into place.

        Raises:
            RuntimeError: If the bundle was not (yet) published.
        """
        if self._result is None:
            raise RuntimeError("the bundle has not been published")
        return self._result

    def _publish(self, result: WrittenBundle) -> None:
        """Record the published bundle (called by :func:`write_bundle`).

        Args:
            result: The bundle now in place.
        """
        self._result = result

manifest property

The sealed manifest (None until :meth:seal runs).

Returns:

Type	Description
BundleManifest | None	The manifest, or None.

result property

The published bundle.

Returns:

Name	Type	Description
What	WrittenBundle	func:write_bundle moved into place.

Raises:

Type	Description
RuntimeError	If the bundle was not (yet) published.

__init__(archive, corpus, created_at)

Bind the writer to its open partial archive.

Parameters:

Name	Type	Description	Default
archive	ZipFile	The partial zip, open for writing.	required
corpus	Corpus	The corpus being backed up.	required
created_at	datetime	The export's instant (UTC).	required

Source code in varagity/backup/bundle.py

def __init__(self, archive: zipfile.ZipFile, corpus: Corpus, created_at: datetime) -> None:
    """Bind the writer to its open partial archive.

    Args:
        archive: The partial zip, open for writing.
        corpus: The corpus being backed up.
        created_at: The export's instant (UTC).
    """
    self._archive = archive
    self.corpus = corpus
    self.created_at = created_at
    self._members: list[BundleMember] = []
    self._manifest: BundleManifest | None = None
    self._result: WrittenBundle | None = None
    # Zip stores a naive DOS timestamp; the UTC wall clock is the honest
    # one (the manifest carries the aware instant regardless).
    self._date_time = created_at.astimezone(UTC).timetuple()[:6]

member(name, *, compress=True)

Stream one data member into the bundle.

Parameters:

Name	Type	Description	Default
name	str	The member's path inside the zip.	required
compress	bool	Deflate it (text) or store it as-is (already-dense bytes like a float32 matrix, which deflate can't shrink).	True

Yields:

Type	Description
MemberSink	The sink to write the member's bytes to; its size and sha256 are
MemberSink	recorded for the manifest when the block exits.

Raises:

Type	Description
ValueError	If the name is the manifest's or already used.
RuntimeError	If the manifest is already sealed.

Source code in varagity/backup/bundle.py

@contextmanager
def member(self, name: str, *, compress: bool = True) -> Iterator[MemberSink]:
    """Stream one data member into the bundle.

    Args:
        name: The member's path inside the zip.
        compress: Deflate it (text) or store it as-is (already-dense
            bytes like a float32 matrix, which deflate can't shrink).

    Yields:
        The sink to write the member's bytes to; its size and sha256 are
        recorded for the manifest when the block exits.

    Raises:
        ValueError: If the name is the manifest's or already used.
        RuntimeError: If the manifest is already sealed.
    """
    if self._manifest is not None:
        raise RuntimeError("the manifest is sealed — no member may follow it")
    if name == MANIFEST_NAME or any(member.name == name for member in self._members):
        raise ValueError(f"bundle member {name!r} is reserved or already written")
    info = zipfile.ZipInfo(name, date_time=self._date_time)
    info.compress_type = zipfile.ZIP_DEFLATED if compress else zipfile.ZIP_STORED
    info.external_attr = _MEMBER_MODE
    # force_zip64: the size is unknown up front, and a corpus's vectors
    # can pass the 2 GiB a plain entry can describe.
    with self._archive.open(info, mode="w", force_zip64=True) as raw:
        sink = MemberSink(raw)
        yield sink
    self._members.append(
        BundleMember(name=name, size_bytes=sink.size_bytes, sha256=sink.hexdigest())
    )

seal(*, embedding_model, embedding_dim, stale, counts, ingest_settings=None, engine=None)

Write bundle.json as the bundle's last member.

Parameters:

Name	Type	Description	Default
embedding_model	str	EMBEDDING_MODEL at export.	required
embedding_dim	int	EMBEDDING_DIM at export.	required
stale	bool	The corpus's stale flag at export.	required
counts	dict[str, int]	What the bundle holds.	required
ingest_settings	dict[str, str] | None	The reingest-affecting settings at export, if the corpus has them.	None
engine	EngineFormat | None	The graph engine's file format, if the corpus is the graph.	None

Returns:

Type	Description
BundleManifest	The manifest, as written.

Raises:

Type	Description
RuntimeError	If the manifest is already sealed.

Source code in varagity/backup/bundle.py

def seal(
    self,
    *,
    embedding_model: str,
    embedding_dim: int,
    stale: bool,
    counts: dict[str, int],
    ingest_settings: dict[str, str] | None = None,
    engine: EngineFormat | None = None,
) -> BundleManifest:
    """Write ``bundle.json`` as the bundle's last member.

    Args:
        embedding_model: ``EMBEDDING_MODEL`` at export.
        embedding_dim: ``EMBEDDING_DIM`` at export.
        stale: The corpus's stale flag at export.
        counts: What the bundle holds.
        ingest_settings: The reingest-affecting settings at export, if
            the corpus has them.
        engine: The graph engine's file format, if the corpus is the
            graph.

    Returns:
        The manifest, as written.

    Raises:
        RuntimeError: If the manifest is already sealed.
    """
    if self._manifest is not None:
        raise RuntimeError("the manifest is already sealed")
    manifest = BundleManifest(
        format=BUNDLE_FORMAT,
        version=BUNDLE_VERSION,
        corpus=self.corpus,
        created_at=self.created_at,
        embedding_model=embedding_model,
        embedding_dim=embedding_dim,
        stale=stale,
        counts=dict(counts),
        ingest_settings=dict(ingest_settings) if ingest_settings is not None else None,
        engine=engine,
        members=list(self._members),
    )
    info = zipfile.ZipInfo(MANIFEST_NAME, date_time=self._date_time)
    info.compress_type = zipfile.ZIP_DEFLATED
    info.external_attr = _MEMBER_MODE
    self._archive.writestr(info, manifest.model_dump_json(indent=2) + "\n")
    self._manifest = manifest
    return manifest

MemberSink

The write end of one bundle member, hashing and counting as it goes.

Attributes:

Name	Type	Description
size_bytes		Bytes written so far (uncompressed).

Source code in varagity/backup/bundle.py

class MemberSink:
    """The write end of one bundle member, hashing and counting as it goes.

    Attributes:
        size_bytes: Bytes written so far (uncompressed).
    """

    def __init__(self, raw: IO[bytes]) -> None:
        """Wrap a zip member's write stream.

        Args:
            raw: The stream ``ZipFile.open(..., "w")`` returned.
        """
        self._raw = raw
        self._hash = hashlib.sha256()
        self.size_bytes = 0

    def write(self, data: bytes | bytearray | memoryview) -> int:
        """Append bytes to the member.

        Args:
            data: The next slice of the member's content.

        Returns:
            The number of bytes written.
        """
        self._raw.write(data)
        self._hash.update(data)
        size = len(data)
        self.size_bytes += size
        return size

    def hexdigest(self) -> str:
        """The sha256 of everything written so far.

        Returns:
            The hex digest.
        """
        return self._hash.hexdigest()

__init__(raw)

Wrap a zip member's write stream.

Parameters:

Name	Type	Description	Default
raw	IO[bytes]	The stream ZipFile.open(..., "w") returned.	required

Source code in varagity/backup/bundle.py

def __init__(self, raw: IO[bytes]) -> None:
    """Wrap a zip member's write stream.

    Args:
        raw: The stream ``ZipFile.open(..., "w")`` returned.
    """
    self._raw = raw
    self._hash = hashlib.sha256()
    self.size_bytes = 0

hexdigest()

The sha256 of everything written so far.

Returns:

Type	Description
str	The hex digest.

Source code in varagity/backup/bundle.py

def hexdigest(self) -> str:
    """The sha256 of everything written so far.

    Returns:
        The hex digest.
    """
    return self._hash.hexdigest()

write(data)

Append bytes to the member.

Parameters:

Name	Type	Description	Default
data	bytes | bytearray | memoryview	The next slice of the member's content.	required

Returns:

Type	Description
int	The number of bytes written.

Source code in varagity/backup/bundle.py

def write(self, data: bytes | bytearray | memoryview) -> int:
    """Append bytes to the member.

    Args:
        data: The next slice of the member's content.

    Returns:
        The number of bytes written.
    """
    self._raw.write(data)
    self._hash.update(data)
    size = len(data)
    self.size_bytes += size
    return size

MemberStream

The read end of one bundle member, hashed and counted as it is read.

Every byte handed to the reader passes through :meth:read or :meth:readline, so once the member is drained the digest covers it whole; :meth:verify then holds it to the manifest's record.

Attributes:

Name	Type	Description
name		The member's name.

Source code in varagity/backup/bundle.py

class MemberStream:
    """The read end of one bundle member, hashed and counted as it is read.

    Every byte handed to the reader passes through :meth:`read` or
    :meth:`readline`, so once the member is drained the digest covers it
    whole; :meth:`verify` then holds it to the manifest's record.

    Attributes:
        name: The member's name.
    """

    def __init__(self, raw: IO[bytes], record: BundleMember) -> None:
        """Wrap a member's decompressing read stream.

        Args:
            raw: The stream ``ZipFile.open(name)`` returned.
            record: What the manifest recorded for the member.
        """
        self.name = record.name
        self._raw = raw
        self._record = record
        self._hash = hashlib.sha256()
        self._size = 0

    def _failure(self, error: BaseException) -> BundleError:
        """Classify a read failure as damage or as an unreadable bundle.

        Args:
            error: What reading the member raised.

        Returns:
            ``BundleDamaged`` for a CRC or deflate failure (the bytes
            changed), ``BundleInvalid`` otherwise.
        """
        if isinstance(error, _DAMAGED_MEMBER):
            return BundleDamaged(
                f"{self.name} could not be read ({error}) — the bundle is damaged."
            )
        return BundleInvalid(f"{self.name} could not be read ({error}).")

    def _take(self, data: bytes) -> bytes:
        """Count and hash bytes on their way to the reader.

        Args:
            data: What the underlying stream returned.

        Returns:
            The same bytes.
        """
        self._hash.update(data)
        self._size += len(data)
        return data

    def read(self, size: int = -1) -> bytes:
        """Read up to ``size`` bytes (all that is left when negative).

        Args:
            size: The most bytes to return.

        Returns:
            The bytes; ``b""`` at the end of the member.

        Raises:
            BundleDamaged: If the member's bytes fail the zip's own checks.
            BundleInvalid: If the member can't be read for another reason.
        """
        try:
            data = self._raw.read(size)
        except _UNREADABLE_ZIP as error:
            raise self._failure(error) from error
        return self._take(data)

    def readline(self) -> bytes:
        """Read one line, newline included.

        Returns:
            The line; ``b""`` at the end of the member.

        Raises:
            BundleDamaged: If the member's bytes fail the zip's own checks.
            BundleInvalid: If the member can't be read for another reason.
        """
        try:
            data = self._raw.readline()
        except _UNREADABLE_ZIP as error:
            raise self._failure(error) from error
        return self._take(data)

    def __iter__(self) -> Iterator[bytes]:
        """Iterate the member's lines (a JSONL member's records).

        Yields:
            Each line, newline included.
        """
        while line := self.readline():
            yield line

    def verify(self) -> None:
        """Drain what the reader left, then hold the member to its record.

        Raises:
            BundleDamaged: If its size or sha256 differs from what the
                manifest recorded.
        """
        while self.read(_READ_BYTES):
            pass
        if self._size != self._record.size_bytes or self._hash.hexdigest() != self._record.sha256:
            raise BundleDamaged(
                f"{self.name} does not match its recorded sha256 — the bundle is damaged."
            )

__init__(raw, record)

Wrap a member's decompressing read stream.

Parameters:

Name	Type	Description	Default
raw	IO[bytes]	The stream ZipFile.open(name) returned.	required
record	BundleMember	What the manifest recorded for the member.	required

Source code in varagity/backup/bundle.py

def __init__(self, raw: IO[bytes], record: BundleMember) -> None:
    """Wrap a member's decompressing read stream.

    Args:
        raw: The stream ``ZipFile.open(name)`` returned.
        record: What the manifest recorded for the member.
    """
    self.name = record.name
    self._raw = raw
    self._record = record
    self._hash = hashlib.sha256()
    self._size = 0

__iter__()

Iterate the member's lines (a JSONL member's records).

Yields:

Type	Description
bytes	Each line, newline included.

Source code in varagity/backup/bundle.py

def __iter__(self) -> Iterator[bytes]:
    """Iterate the member's lines (a JSONL member's records).

    Yields:
        Each line, newline included.
    """
    while line := self.readline():
        yield line

read(size=-1)

Read up to size bytes (all that is left when negative).

Parameters:

Name	Type	Description	Default
size	int	The most bytes to return.	-1

Returns:

Type	Description
bytes	The bytes; b"" at the end of the member.

Raises:

Type	Description
BundleDamaged	If the member's bytes fail the zip's own checks.
BundleInvalid	If the member can't be read for another reason.

Source code in varagity/backup/bundle.py

def read(self, size: int = -1) -> bytes:
    """Read up to ``size`` bytes (all that is left when negative).

    Args:
        size: The most bytes to return.

    Returns:
        The bytes; ``b""`` at the end of the member.

    Raises:
        BundleDamaged: If the member's bytes fail the zip's own checks.
        BundleInvalid: If the member can't be read for another reason.
    """
    try:
        data = self._raw.read(size)
    except _UNREADABLE_ZIP as error:
        raise self._failure(error) from error
    return self._take(data)

readline()

Read one line, newline included.

Returns:

Type	Description
bytes	The line; b"" at the end of the member.

Raises:

Type	Description
BundleDamaged	If the member's bytes fail the zip's own checks.
BundleInvalid	If the member can't be read for another reason.

Source code in varagity/backup/bundle.py

def readline(self) -> bytes:
    """Read one line, newline included.

    Returns:
        The line; ``b""`` at the end of the member.

    Raises:
        BundleDamaged: If the member's bytes fail the zip's own checks.
        BundleInvalid: If the member can't be read for another reason.
    """
    try:
        data = self._raw.readline()
    except _UNREADABLE_ZIP as error:
        raise self._failure(error) from error
    return self._take(data)

verify()

Drain what the reader left, then hold the member to its record.

Raises:

Type	Description
BundleDamaged	If its size or sha256 differs from what the manifest recorded.

Source code in varagity/backup/bundle.py

def verify(self) -> None:
    """Drain what the reader left, then hold the member to its record.

    Raises:
        BundleDamaged: If its size or sha256 differs from what the
            manifest recorded.
    """
    while self.read(_READ_BYTES):
        pass
    if self._size != self._record.size_bytes or self._hash.hexdigest() != self._record.sha256:
        raise BundleDamaged(
            f"{self.name} does not match its recorded sha256 — the bundle is damaged."
        )

NothingToExport

Bases: RuntimeError

Raised when a corpus holds nothing to back up — an empty bundle is not a backup.

The RAG corpus with no documents row, or a graph whose working directory has no file to copy (nothing has ever been built).

Source code in varagity/backup/bundle.py

class NothingToExport(RuntimeError):
    """Raised when a corpus holds nothing to back up — an empty bundle is not a backup.

    The RAG corpus with no ``documents`` row, or a graph whose working
    directory has no file to copy (nothing has ever been built).
    """

WrittenBundle dataclass

A bundle :func:write_bundle published.

Attributes:

Name	Type	Description
name	str	The bundle's file name.
path	Path	Where it now lives.
size_bytes	int	Its size on disk.
manifest	BundleManifest	The manifest sealed into it.

Source code in varagity/backup/bundle.py

@dataclass(frozen=True)
class WrittenBundle:
    """A bundle :func:`write_bundle` published.

    Attributes:
        name: The bundle's file name.
        path: Where it now lives.
        size_bytes: Its size on disk.
        manifest: The manifest sealed into it.
    """

    name: str
    path: Path
    size_bytes: int
    manifest: BundleManifest

bundle_name(directory, corpus, *, now=None)

Choose the file name for a new bundle — never one already taken.

varagity-<corpus>-<UTC yyyymmdd-HHMM>.zip; a repeat within the same minute gets a -2, -3, … suffix rather than replacing a backup.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder the bundle will land in.	required
corpus	Corpus	The corpus being backed up.	required
now	datetime | None	The export's instant (defaults to now); rendered in UTC.	None

Returns:

Type	Description
str	A name free in directory (also free of an in-flight partial).

Source code in varagity/backup/bundle.py

def bundle_name(directory: Path, corpus: Corpus, *, now: datetime | None = None) -> str:
    """Choose the file name for a new bundle — never one already taken.

    ``varagity-<corpus>-<UTC yyyymmdd-HHMM>.zip``; a repeat within the same
    minute gets a ``-2``, ``-3``, … suffix rather than replacing a backup.

    Args:
        directory: The exports folder the bundle will land in.
        corpus: The corpus being backed up.
        now: The export's instant (defaults to now); rendered in UTC.

    Returns:
        A name free in ``directory`` (also free of an in-flight partial).
    """
    stamp = (now or datetime.now(UTC)).astimezone(UTC).strftime("%Y%m%d-%H%M")
    base = f"varagity-{corpus}-{stamp}"
    candidate = f"{base}{BUNDLE_SUFFIX}"
    attempt = 2
    while _taken(directory, candidate):
        candidate = f"{base}-{attempt}{BUNDLE_SUFFIX}"
        attempt += 1
    return candidate

corpus_root(corpus)

The configured root directory of one corpus.

Parameters:

Name	Type	Description	Default
corpus	Corpus	The corpus.	required

Returns:

Type	Description
Path	DOCS_PATH for rag, GRAPH_DOCS_PATH for graph.

Source code in varagity/backup/bundle.py

def corpus_root(corpus: Corpus) -> Path:
    """The configured root directory of one corpus.

    Args:
        corpus: The corpus.

    Returns:
        ``DOCS_PATH`` for ``rag``, ``GRAPH_DOCS_PATH`` for ``graph``.
    """
    settings = get_settings()
    return Path(settings.DOCS_PATH if corpus == "rag" else settings.GRAPH_DOCS_PATH)

display_path(corpus, name=None)

The human form of a bundle's location — the text "Copy path" copies.

In compose the API sees /app/docs while the user's checkout has ./docs (the bind's host side), so an absolute path would be wrong on the one machine that matters. The corpus folder's own name, relative, reads right from the checkout root in both host and container mode.

Parameters:

Name	Type	Description	Default
corpus	Corpus	The corpus.	required
name	str | None	A bundle file name; None names the folder itself.	None

Returns:

Type	Description
str	E.g. "docs/varagity-exports/varagity-rag-20261108-1312.zip" or
str	"docs/varagity-exports/".

Source code in varagity/backup/bundle.py

def display_path(corpus: Corpus, name: str | None = None) -> str:
    """The human form of a bundle's location — the text "Copy path" copies.

    In compose the API sees ``/app/docs`` while the user's checkout has
    ``./docs`` (the bind's host side), so an absolute path would be wrong on
    the one machine that matters. The corpus folder's own name, relative,
    reads right from the checkout root in both host and container mode.

    Args:
        corpus: The corpus.
        name: A bundle file name; ``None`` names the folder itself.

    Returns:
        E.g. ``"docs/varagity-exports/varagity-rag-20261108-1312.zip"`` or
        ``"docs/varagity-exports/"``.
    """
    # A root with no name is "/" itself — and "/varagity-exports/" is then
    # exactly right.
    folder = f"{corpus_root(corpus).resolve().name}/{EXPORTS_DIRNAME}/"
    return folder if name is None else f"{folder}{name}"

exports_dir(corpus)

The reserved folder a corpus's bundles live in.

Parameters:

Name	Type	Description	Default
corpus	Corpus	The corpus.	required

Returns:

Type	Description
Path	<corpus root>/varagity-exports (it may not exist yet).

Source code in varagity/backup/bundle.py

def exports_dir(corpus: Corpus) -> Path:
    """The reserved folder a corpus's bundles live in.

    Args:
        corpus: The corpus.

    Returns:
        ``<corpus root>/varagity-exports`` (it may not exist yet).
    """
    return corpus_root(corpus) / EXPORTS_DIRNAME

graph_folder(manifest)

The folder a graph bundle's files ride under: the name of the engine that wrote them.

Shared by the verdict's member rule and the restore, which extracts <folder>/<file> into the workdir as <file>.

Parameters:

Name	Type	Description	Default
manifest	BundleManifest	The graph bundle's manifest.	required

Returns:

Type	Description
str	The engine's registry name (lightrag).

Raises:

Type	Description
BundleInvalid	If the manifest records no engine, or one whose name couldn't be a folder.

Source code in varagity/backup/bundle.py

def graph_folder(manifest: BundleManifest) -> str:
    """The folder a graph bundle's files ride under: the name of the engine that wrote them.

    Shared by the verdict's member rule and the restore, which extracts
    ``<folder>/<file>`` into the workdir as ``<file>``.

    Args:
        manifest: The graph bundle's manifest.

    Returns:
        The engine's registry name (``lightrag``).

    Raises:
        BundleInvalid: If the manifest records no engine, or one whose name
            couldn't be a folder.
    """
    engine = manifest.engine
    if engine is None:
        raise BundleInvalid(f"{MANIFEST_NAME} does not record the graph engine that wrote it.")
    if not is_graph_file_name(engine.name):
        raise BundleInvalid(
            f"{MANIFEST_NAME} records the graph engine {engine.name!r}, which is not a folder name."
        )
    return engine.name

inspect_bundle(path, corpus)

The cheap verdict the backups list shows for one bundle.

Reads the zip's central directory and bundle.json — never a data member — and checks everything those can tell: the format version, the corpus, the member allowlist and zip-slip rules, each member's recorded size against the zip's, the embedding model and dimension against this stack's settings, and for a graph bundle the engine that wrote its files against this stack's (:func:running_engine_format). What it can't see is a member damaged inside an intact zip with unchanged sizes: the restore's verifying streams (:func:open_bundle) — or :func:verify_bundle, standalone — hash for that.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required
corpus	Corpus	The corpus it would restore.	required

Returns:

Type	Description
BundleVerdict	The verdict (it never raises for a bad file — the problem rides
BundleVerdict	the verdict).

Source code in varagity/backup/bundle.py

def inspect_bundle(path: Path, corpus: Corpus) -> BundleVerdict:
    """The cheap verdict the backups list shows for one bundle.

    Reads the zip's central directory and ``bundle.json`` — never a data
    member — and checks everything those can tell: the format version, the
    corpus, the member allowlist and zip-slip rules, each member's recorded
    size against the zip's, the embedding model and dimension against this
    stack's settings, and for a graph bundle the engine that wrote its files
    against this stack's (:func:`running_engine_format`). What it can't see
    is a member damaged *inside* an intact zip with unchanged sizes: the
    restore's verifying streams (:func:`open_bundle`) — or
    :func:`verify_bundle`, standalone — hash for that.

    Args:
        path: The bundle file.
        corpus: The corpus it would restore.

    Returns:
        The verdict (it never raises for a bad file — the problem rides
        the verdict).
    """
    try:
        with zipfile.ZipFile(path) as archive:
            return _inspect_archive(archive, corpus)
    except _UNREADABLE_ZIP as error:
        return BundleVerdict(manifest=None, problem=_unreadable(error))

is_bundle_name(name)

Report whether a file name is a listable, servable bundle name.

Parameters:

Name	Type	Description	Default
name	str	A bare file name (no directories).	required

Returns:

Type	Description
bool	True for a visible *.zip name — never a hidden file, so an
bool	in-flight .{name}.partial (or anything else dot-prefixed) is
bool	never listed or served.

Source code in varagity/backup/bundle.py

def is_bundle_name(name: str) -> bool:
    """Report whether a file name is a listable, servable bundle name.

    Args:
        name: A bare file name (no directories).

    Returns:
        ``True`` for a visible ``*.zip`` name — never a hidden file, so an
        in-flight ``.{name}.partial`` (or anything else dot-prefixed) is
        never listed or served.
    """
    return (
        bool(name)
        and not name.startswith(".")
        and "/" not in name
        and "\\" not in name
        and name.lower().endswith(BUNDLE_SUFFIX)
    )

is_graph_file_name(name)

Report whether a working-directory file name may ride in a graph bundle.

The graph bundle's member rule (and the rule for the engine folder those members sit under), shared by the export — which must never write a member the restore's verdict would refuse — and the verdict itself.

Parameters:

Name	Type	Description	Default
name	str	A bare file (or folder) name.	required

Returns:

Type	Description
bool	True for a visible name of at most 255 ASCII letters, digits,
bool	_, . and - — every name an engine writes, and never one
bool	that could be a path, a hidden file, . or ...

Source code in varagity/backup/bundle.py

def is_graph_file_name(name: str) -> bool:
    """Report whether a working-directory file name may ride in a graph bundle.

    The graph bundle's member rule (and the rule for the engine folder those
    members sit under), shared by the export — which must never write a
    member the restore's verdict would refuse — and the verdict itself.

    Args:
        name: A bare file (or folder) name.

    Returns:
        ``True`` for a visible name of at most 255 ASCII letters, digits,
        ``_``, ``.`` and ``-`` — every name an engine writes, and never one
        that could be a path, a hidden file, ``.`` or ``..``.
    """
    return _GRAPH_FILE_NAME.fullmatch(name) is not None

list_bundles(directory, corpus)

List the bundles in an exports folder, newest first, each with its verdict.

Hidden files (in-flight partials included) and non-.zip files are skipped, and so is anything that resolves outside the folder (a symlink pointing elsewhere is not a bundle this folder holds). A visible zip that isn't a readable bundle is still listed — with no manifest and its reason — because the user put it there and the panel should say what it is.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder (a missing folder lists nothing).	required
corpus	Corpus	The corpus the folder belongs to (what each bundle is judged against).	required

Returns:

Type	Description
list[BundleListing]	One listing per bundle file, newest mtime first.

Source code in varagity/backup/bundle.py

def list_bundles(directory: Path, corpus: Corpus) -> list[BundleListing]:
    """List the bundles in an exports folder, newest first, each with its verdict.

    Hidden files (in-flight partials included) and non-``.zip`` files
    are skipped, and so is anything that resolves outside the folder (a
    symlink pointing elsewhere is not a bundle this folder holds). A visible
    zip that isn't a readable bundle is still listed — with no manifest and
    its reason — because the user put it there and the panel should say what
    it is.

    Args:
        directory: The exports folder (a missing folder lists nothing).
        corpus: The corpus the folder belongs to (what each bundle is judged
            against).

    Returns:
        One listing per bundle file, newest mtime first.
    """
    if not directory.is_dir():
        return []
    root = directory.resolve()
    listings: list[BundleListing] = []
    for entry in directory.iterdir():
        if not is_bundle_name(entry.name):
            continue
        resolved = resolve_contained(entry, root)
        if resolved is None or not resolved.is_file():
            continue
        try:
            info = resolved.stat()
        except OSError:
            continue  # vanished between the listing and the stat
        verdict = inspect_bundle(resolved, corpus)
        if verdict.problem is not None:
            logger.debug("listing %s as not restorable: %s", entry.name, verdict.reason)
        listings.append(
            BundleListing(
                name=entry.name,
                path=resolved,
                size_bytes=info.st_size,
                modified_at=datetime.fromtimestamp(info.st_mtime, tz=UTC),
                mtime_ns=info.st_mtime_ns,
                verdict=verdict,
            )
        )
    listings.sort(key=lambda listing: (listing.modified_at, listing.name), reverse=True)
    return listings

open_bundle(path, manifest)

Open a bundle to read its members against its manifest.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required
manifest	BundleManifest	Its manifest (from :func:inspect_bundle or :func:verify_bundle).	required

Yields:

Type	Description
BundleSource	The open bundle.

Raises:

Type	Description
BundleInvalid	If the file can't be opened as a zip.

Source code in varagity/backup/bundle.py

@contextmanager
def open_bundle(path: Path, manifest: BundleManifest) -> Iterator[BundleSource]:
    """Open a bundle to read its members against its manifest.

    Args:
        path: The bundle file.
        manifest: Its manifest (from :func:`inspect_bundle` or
            :func:`verify_bundle`).

    Yields:
        The open bundle.

    Raises:
        BundleInvalid: If the file can't be opened as a zip.
    """
    try:
        archive = zipfile.ZipFile(path)
    except _UNREADABLE_ZIP as error:
        raise _unreadable(error) from error
    with archive:
        yield BundleSource(archive, manifest)

read_manifest(path)

Read a bundle's manifest — and only its manifest.

The zip's central directory locates bundle.json; no other member is opened, so listing a folder of large bundles stays cheap.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required

Returns:

Type	Description
BundleManifest	The validated manifest.

Raises:

Type	Description
BundleInvalid	If the file is not a readable zip, has no bundle.json, or its manifest is not a varagity manifest.
BundleIncompatible	If the manifest is of another format version.

Source code in varagity/backup/bundle.py

def read_manifest(path: Path) -> BundleManifest:
    """Read a bundle's manifest — and only its manifest.

    The zip's central directory locates ``bundle.json``; no other member is
    opened, so listing a folder of large bundles stays cheap.

    Args:
        path: The bundle file.

    Returns:
        The validated manifest.

    Raises:
        BundleInvalid: If the file is not a readable zip, has no
            ``bundle.json``, or its manifest is not a varagity manifest.
        BundleIncompatible: If the manifest is of another format version.
    """
    try:
        with zipfile.ZipFile(path) as archive:
            raw = _manifest_bytes(archive)
    except BundleError as error:
        raise type(error)(f"{path.name}: {error}") from error
    except _UNREADABLE_ZIP as error:
        raise BundleInvalid(f"{path.name}: {_unreadable(error)}") from error
    try:
        return _parse_manifest(raw)
    except BundleError as error:
        raise type(error)(f"{path.name}: {error}") from error

running_engine_format()

The file format this stack's graph engine reads — what a graph bundle is held to.

The engine reports its own format (GraphEngine.storage_format(), read from package metadata, so the engine library is never imported), which keeps the backup code from hardcoding any library's name or version.

Returns:

Type	Description
EngineFormat	GRAPH_ENGINE's registered engine's format.

Raises:

Type	Description
KeyError	If GRAPH_ENGINE names no registered engine (the config validator makes this unreachable).

Source code in varagity/backup/bundle.py

def running_engine_format() -> EngineFormat:
    """The file format this stack's graph engine reads — what a graph bundle is held to.

    The engine reports its own format (``GraphEngine.storage_format()``, read
    from package metadata, so the engine library is never imported), which
    keeps the backup code from hardcoding any library's name or version.

    Returns:
        ``GRAPH_ENGINE``'s registered engine's format.

    Raises:
        KeyError: If ``GRAPH_ENGINE`` names no registered engine (the config
            validator makes this unreachable).
    """
    # Here rather than at the top: importing the engines package registers
    # its adapters, which this module — imported by the API schemas — needs
    # only once a graph bundle is actually judged.
    from varagity.graph.engines import get_graph_engine

    return get_graph_engine(get_settings().GRAPH_ENGINE).storage_format()

sweep_partials(directory, corpus)

Remove partial bundles an interrupted export of corpus left behind.

Safe because only one export per corpus runs at a time (the corpus gate), and only this corpus's prefix is touched — another corpus's export into the same folder is never disturbed.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder.	required
corpus	Corpus	The corpus whose partials to sweep.	required

Returns:

Type	Description
int	How many partials were removed.

Source code in varagity/backup/bundle.py

def sweep_partials(directory: Path, corpus: Corpus) -> int:
    """Remove partial bundles an interrupted export of ``corpus`` left behind.

    Safe because only one export per corpus runs at a time (the corpus gate),
    and only this corpus's prefix is touched — another corpus's export into
    the same folder is never disturbed.

    Args:
        directory: The exports folder.
        corpus: The corpus whose partials to sweep.

    Returns:
        How many partials were removed.
    """
    removed = 0
    if not directory.is_dir():
        return removed
    for path in directory.glob(f".varagity-{corpus}-*{BUNDLE_SUFFIX}{PARTIAL_SUFFIX}"):
        try:
            path.unlink()
        except OSError as error:
            logger.warning("could not remove the stale partial %s: %s", path.name, error)
            continue
        logger.info("removed %s, left behind by an interrupted export", path.name)
        removed += 1
    return removed

verify_bundle(path, corpus)

Run every check a restore makes, without restoring anything.

:func:inspect_bundle's verdict, then the sha256 of every member — every byte of the bundle is read once. A restore makes the same checks as it loads (through :func:open_bundle's verifying streams, so it reads the bundle only once); this is the standalone form, for a caller that wants the full verdict without a restore.

Parameters:

Name	Type	Description	Default
path	Path	The bundle file.	required
corpus	Corpus	The corpus it would restore.	required

Returns:

Type	Description
BundleManifest	The bundle's manifest, now fully verified.

Raises:

Type	Description
BundleIncompatible	If this stack must not restore the bundle.
BundleDamaged	If a member's bytes don't match its recorded sha256.
BundleInvalid	If the bundle is otherwise unsound.

Source code in varagity/backup/bundle.py

def verify_bundle(path: Path, corpus: Corpus) -> BundleManifest:
    """Run every check a restore makes, without restoring anything.

    :func:`inspect_bundle`'s verdict, then the sha256 of every member —
    every byte of the bundle is read once. A restore makes the same checks
    as it loads (through :func:`open_bundle`'s verifying streams, so it
    reads the bundle only once); this is the standalone form, for a caller
    that wants the full verdict without a restore.

    Args:
        path: The bundle file.
        corpus: The corpus it would restore.

    Returns:
        The bundle's manifest, now fully verified.

    Raises:
        BundleIncompatible: If this stack must not restore the bundle.
        BundleDamaged: If a member's bytes don't match its recorded sha256.
        BundleInvalid: If the bundle is otherwise unsound.
    """
    manifest = inspect_bundle(path, corpus).manifest_or_raise()
    with open_bundle(path, manifest) as source:
        for member in manifest.members:
            # Opening and closing a member is the whole check: the clean
            # exit drains it through the hash and compares the digest.
            with source.member(member.name):
                pass
    return manifest

write_bundle(directory, corpus, *, created_at=None)

Write a new bundle into directory, all-or-nothing.

Sweeps this corpus's stale partials, chooses a free name (:func:bundle_name), and opens .{name}.partial. When the with block exits cleanly and the writer was sealed, the partial is synced and renamed into place with os.replace — same directory, so the rename is atomic and a listing sees the whole bundle or nothing. Any exception, or an unsealed writer, deletes the partial instead.

Parameters:

Name	Type	Description	Default
directory	Path	The exports folder (created if missing).	required
corpus	Corpus	The corpus being backed up.	required
created_at	datetime | None	The export's instant (defaults to now).	None

Yields:

Type	Description
BundleWriter	The writer to fill; read :attr:BundleWriter.result after the block.

Raises:

Type	Description
RuntimeError	If the block exits without sealing the manifest (the partial is removed; nothing is published).
OSError	If the folder or the bundle can't be written.

Source code in varagity/backup/bundle.py

@contextmanager
def write_bundle(
    directory: Path, corpus: Corpus, *, created_at: datetime | None = None
) -> Iterator[BundleWriter]:
    """Write a new bundle into ``directory``, all-or-nothing.

    Sweeps this corpus's stale partials, chooses a free name
    (:func:`bundle_name`), and opens ``.{name}.partial``. When the ``with``
    block exits cleanly *and* the writer was sealed, the partial is synced
    and renamed into place with ``os.replace`` — same directory, so the
    rename is atomic and a listing sees the whole bundle or nothing. Any
    exception, or an unsealed writer, deletes the partial instead.

    Args:
        directory: The exports folder (created if missing).
        corpus: The corpus being backed up.
        created_at: The export's instant (defaults to now).

    Yields:
        The writer to fill; read :attr:`BundleWriter.result` after the block.

    Raises:
        RuntimeError: If the block exits without sealing the manifest (the
            partial is removed; nothing is published).
        OSError: If the folder or the bundle can't be written.
    """
    created = (created_at or datetime.now(UTC)).astimezone(UTC)
    directory.mkdir(parents=True, exist_ok=True)
    sweep_partials(directory, corpus)
    name = bundle_name(directory, corpus, now=created)
    partial = directory / f".{name}{PARTIAL_SUFFIX}"
    try:
        # "x": never clobber a partial some other writer owns.
        with zipfile.ZipFile(partial, mode="x", compression=zipfile.ZIP_DEFLATED) as archive:
            writer = BundleWriter(archive, corpus, created)
            yield writer
            manifest = writer.manifest
            if manifest is None:
                raise RuntimeError(
                    "the bundle was never sealed — a bundle without its manifest is not published"
                )
        _fsync(partial)
        final = directory / name
        if final.exists(follow_symlinks=False):
            # Something claimed the name while we wrote — never replace it.
            final = directory / bundle_name(directory, corpus, now=created)
        os.replace(partial, final)
    except BaseException:
        partial.unlink(missing_ok=True)
        raise
    fsync_directory(directory)
    size = final.stat().st_size
    writer._publish(WrittenBundle(name=final.name, path=final, size_bytes=size, manifest=manifest))
    # DEBUG: the corpus export logs its own one-line summary at INFO.
    logger.debug("published %s (%d bytes)", final.name, size)

varagity.backup.rag

The RAG bundle: every documents and chunks row, with its vector (ADR-019).

A RAG backup is the smallest thing that restores the chunk corpus with no model call: the rows pgvector holds (contextual blurbs included) and their embeddings. The BM25 index is deliberately absent — every field of an Elasticsearch document is a copy of a chunks column, so a restore rebuilds it from these same rows.

The bundle's data members, in write order::

documents.jsonl   one documents row per line, plus its relative_path    deflated
chunks.jsonl      one chunks row per line, every column but the vector  deflated
embeddings.f32    N × EMBEDDING_DIM little-endian float32 — row i is    stored
                  the vector of chunks.jsonl line i

relative_path is each document's source relative to DOCS_PATH (null for a document ingested from elsewhere): doc_id already hashes it, and it is what lets a restore re-root source onto whatever corpus directory the restoring stack has. Everything is read from one corpus snapshot (:meth:~varagity.stores.vector_store.ContextualVectorDB.iter_corpus), so the three members always describe the same moment.

A restore (:func:restore_rag) turns a bundle back into rows without a model call. Under the corpus's exclusive advisory lock, one Postgres transaction replaces both tables — the bundle's lines streamed straight into binary COPY, each member's sha256 checked again as it is read — and its commit is where the restore takes effect. The BM25 index is then rebuilt from the restored rows. The corpus-stale flag it leaves comes from the bundle (:func:derive_corpus_stale).

ExportStage = Literal['snapshot', 'documents', 'chunks', 'finalize'] module-attribute

The export's progress stages, in order.

ProgressCallback = Callable[[ExportStage, int | None, int | None], None] module-attribute

(stage, current, total) — how an export reports where it is.

RestoreProgressCallback = Callable[[RestoreStage, int | None, int | None], None] module-attribute

(stage, current, total) — how a restore reports where it is.

RestoreStage = Literal['validate', 'load', 'index', 'bm25'] module-attribute

The restore's progress stages, in order.

BM25RebuildFailed

Bases: RuntimeError

The rows are restored, but the BM25 index could not be rebuilt.

The Postgres commit already happened, so the restore took effect; restoring the same bundle again rebuilds the index (and rewrites the identical rows on the way).

Attributes:

Name	Type	Description
manifest		The restored bundle's manifest — the rows now in the corpus are its rows, so the corpus-stale flag must still be derived from it (None only for an instance rebuilt from its message alone, e.g. unpickled).

Source code in varagity/backup/rag.py

class BM25RebuildFailed(RuntimeError):
    """The rows are restored, but the BM25 index could not be rebuilt.

    The Postgres commit already happened, so the restore took effect;
    restoring the same bundle again rebuilds the index (and rewrites the
    identical rows on the way).

    Attributes:
        manifest: The restored bundle's manifest — the rows now in the
            corpus are its rows, so the corpus-stale flag must still be
            derived from it (``None`` only for an instance rebuilt from its
            message alone, e.g. unpickled).
    """

    def __init__(self, message: str, manifest: BundleManifest | None = None) -> None:
        """Word the failure and keep the manifest of the rows that did commit.

        Args:
            message: What failed, and how to repair it.
            manifest: The restored bundle's manifest.
        """
        super().__init__(message)
        self.manifest = manifest

__init__(message, manifest=None)

Word the failure and keep the manifest of the rows that did commit.

Parameters:

Name	Type	Description	Default
message	str	What failed, and how to repair it.	required
manifest	BundleManifest | None	The restored bundle's manifest.	None

Source code in varagity/backup/rag.py

def __init__(self, message: str, manifest: BundleManifest | None = None) -> None:
    """Word the failure and keep the manifest of the rows that did commit.

    Args:
        message: What failed, and how to repair it.
        manifest: The restored bundle's manifest.
    """
    super().__init__(message)
    self.manifest = manifest

RagExport dataclass

What one RAG export wrote.

Attributes:

Name	Type	Description
bundle	WrittenBundle	The published bundle (name, path, size, manifest).
wall_clock_s	float	Seconds from snapshot to publish.

Source code in varagity/backup/rag.py

@dataclass(frozen=True)
class RagExport:
    """What one RAG export wrote.

    Attributes:
        bundle: The published bundle (name, path, size, manifest).
        wall_clock_s: Seconds from snapshot to publish.
    """

    bundle: WrittenBundle
    wall_clock_s: float

    @property
    def counts(self) -> dict[str, int]:
        """The bundle's row counts (``documents``, ``chunks``).

        Returns:
            The manifest's counts.
        """
        return dict(self.bundle.manifest.counts)

counts property

The bundle's row counts (documents, chunks).

Returns:

Type	Description
dict[str, int]	The manifest's counts.

RagRestore dataclass

What one RAG restore did.

Attributes:

Name	Type	Description
bundle_name	str	The restored bundle's file name.
size_bytes	int	The bundle's size on disk.
manifest	BundleManifest	Its manifest.
documents	int	documents rows written.
chunks	int	chunks rows written.
bm25_indexed	int	Chunks the rebuilt BM25 index holds.
wall_clock_s	float	Seconds from validation to a searchable index.
stale	bool | None	The corpus-stale flag the restore left — derived from the bundle by the rag-restore flow; None until it has.

Source code in varagity/backup/rag.py

@dataclass(frozen=True)
class RagRestore:
    """What one RAG restore did.

    Attributes:
        bundle_name: The restored bundle's file name.
        size_bytes: The bundle's size on disk.
        manifest: Its manifest.
        documents: ``documents`` rows written.
        chunks: ``chunks`` rows written.
        bm25_indexed: Chunks the rebuilt BM25 index holds.
        wall_clock_s: Seconds from validation to a searchable index.
        stale: The corpus-stale flag the restore left — derived from the
            bundle by the ``rag-restore`` flow; ``None`` until it has.
    """

    bundle_name: str
    size_bytes: int
    manifest: BundleManifest
    documents: int
    chunks: int
    bm25_indexed: int
    wall_clock_s: float
    stale: bool | None = None

    @property
    def counts(self) -> dict[str, int]:
        """What the restore wrote (``documents``, ``chunks``, ``bm25``).

        Returns:
            The row counts, plus the chunks the BM25 index now holds.
        """
        return {"documents": self.documents, "chunks": self.chunks, "bm25": self.bm25_indexed}

counts property

What the restore wrote (documents, chunks, bm25).

Returns:

Type	Description
dict[str, int]	The row counts, plus the chunks the BM25 index now holds.

chunk_record(row)

Render one chunks row as its chunks.jsonl object.

Parameters:

Name	Type	Description	Default
row	CorpusChunkRow	The snapshot row.	required

Returns:

Type	Description
dict[str, Any]	Every column except the embedding (which rides embeddings.f32,
dict[str, Any]	at the same row index).

Source code in varagity/backup/rag.py

def chunk_record(row: CorpusChunkRow) -> dict[str, Any]:
    """Render one ``chunks`` row as its ``chunks.jsonl`` object.

    Args:
        row: The snapshot row.

    Returns:
        Every column except the embedding (which rides ``embeddings.f32``,
        at the same row index).
    """
    return {
        "chunk_id": row.chunk_id,
        "doc_id": row.doc_id,
        "original_index": row.original_index,
        "chunk_index": row.chunk_index,
        "content": row.content,
        "context": row.context,
        "contextualized_content": row.contextualized_content,
        "metadata": row.metadata,
        "created_at": row.created_at.isoformat(),
    }

derive_corpus_stale(manifest, current)

The corpus-stale flag a restored RAG bundle leaves.

Stale when the bundle was stale at export, or when the settings its rows were made under differ from this stack's — the same "the corpus may not match the current settings" the flag always means.

Parameters:

Name	Type	Description	Default
manifest	BundleManifest	The restored bundle's manifest.	required
current	Mapping[str, str]	This stack's reingest-affecting settings, in env-string form.	required

Returns:

Type	Description
bool	Whether the restored corpus is stale.

Source code in varagity/backup/rag.py

def derive_corpus_stale(manifest: BundleManifest, current: Mapping[str, str]) -> bool:
    """The corpus-stale flag a restored RAG bundle leaves.

    Stale when the bundle was stale at export, or when the settings its
    rows were made under differ from this stack's — the same "the corpus
    may not match the current settings" the flag always means.

    Args:
        manifest: The restored bundle's manifest.
        current: This stack's reingest-affecting settings, in env-string
            form.

    Returns:
        Whether the restored corpus is stale.
    """
    return manifest.stale or bool(ingest_drift(manifest, current))

document_record(row, docs_root)

Render one documents row as its documents.jsonl object.

Parameters:

Name	Type	Description	Default
row	CorpusDocumentRow	The snapshot row.	required
docs_root	Path	The resolved DOCS_PATH the relative path is taken against.	required

Returns:

Type	Description
dict[str, Any]	Every column, plus relative_path (None when the source lives
dict[str, Any]	outside the corpus directory).

Source code in varagity/backup/rag.py

def document_record(row: CorpusDocumentRow, docs_root: Path) -> dict[str, Any]:
    """Render one ``documents`` row as its ``documents.jsonl`` object.

    Args:
        row: The snapshot row.
        docs_root: The resolved ``DOCS_PATH`` the relative path is taken
            against.

    Returns:
        Every column, plus ``relative_path`` (``None`` when the source lives
        outside the corpus directory).
    """
    return {
        "doc_id": row.doc_id,
        "source": row.source,
        "relative_path": relative_to_root(row.source, docs_root),
        "file_type": row.file_type,
        "content_hash": row.content_hash,
        "n_chunks": row.n_chunks,
        "ingested_at": row.ingested_at.isoformat(),
    }

export_rag(dest, store, ingest_settings, *, stale, on_progress=None)

Export the whole chunk corpus into a new bundle in dest.

Parameters:

Name	Type	Description	Default
dest	Path	The exports folder (<DOCS_PATH>/varagity-exports).	required
store	ContextualVectorDB	The vector store to snapshot.	required
ingest_settings	Mapping[str, str]	The reingest-affecting settings at export, in env-string form — the runner reads them, so this package never imports the API layer.	required
stale	bool	The corpus-stale flag at export.	required
on_progress	ProgressCallback | None	Called with (stage, current, total) as the export moves through snapshot → documents → chunks → finalize.	None

Returns:

Type	Description
RagExport	The published bundle and how long it took.

Raises:

Type	Description
NothingToExport	If the snapshot holds no documents.
ValueError	If a stored vector's dimension isn't EMBEDDING_DIM.
OSError	If the bundle can't be written.

Source code in varagity/backup/rag.py

def export_rag(
    dest: Path,
    store: ContextualVectorDB,
    ingest_settings: Mapping[str, str],
    *,
    stale: bool,
    on_progress: ProgressCallback | None = None,
) -> RagExport:
    """Export the whole chunk corpus into a new bundle in ``dest``.

    Args:
        dest: The exports folder (``<DOCS_PATH>/varagity-exports``).
        store: The vector store to snapshot.
        ingest_settings: The reingest-affecting settings at export, in
            env-string form — the runner reads them, so this package never
            imports the API layer.
        stale: The corpus-stale flag at export.
        on_progress: Called with ``(stage, current, total)`` as the export
            moves through ``snapshot`` → ``documents`` → ``chunks`` →
            ``finalize``.

    Returns:
        The published bundle and how long it took.

    Raises:
        NothingToExport: If the snapshot holds no documents.
        ValueError: If a stored vector's dimension isn't ``EMBEDDING_DIM``.
        OSError: If the bundle can't be written.
    """
    settings = get_settings()
    report: ProgressCallback = on_progress if on_progress is not None else _ignore
    docs_root = Path(settings.DOCS_PATH).resolve()
    dim = settings.EMBEDDING_DIM
    started = time.perf_counter()
    created_at = datetime.now(UTC)

    report("snapshot", None, None)
    with store.iter_corpus() as snapshot:
        if snapshot.document_count == 0:
            raise NothingToExport("the corpus holds no documents — there is nothing to export")
        # File names only in these lines: they are relayed into the panel's
        # log, where an in-container path (/app/docs/…) would contradict the
        # checkout-relative "Saved to" path beside it.
        logger.info(
            "exporting %d document(s) and %d chunk(s) from one snapshot",
            snapshot.document_count,
            snapshot.chunk_count,
        )
        with write_bundle(dest, "rag", created_at=created_at) as bundle:
            documents = _write_documents(bundle, snapshot, docs_root, report)
            # The spool sits beside the bundle (same disk) and is hidden and
            # unnamed where the OS allows: it is never listed, and it
            # vanishes with the process if the export dies.
            with tempfile.TemporaryFile(dir=dest, prefix=".varagity-rag-spool-") as spool:
                chunks, models = _write_chunks(
                    bundle, snapshot, dim=dim, spool=spool, report=report
                )
                report("finalize", None, None)
                _write_embeddings(bundle, spool)
            bundle.seal(
                embedding_model=settings.EMBEDDING_MODEL,
                embedding_dim=dim,
                stale=stale,
                counts={"documents": documents, "chunks": chunks},
                ingest_settings=dict(ingest_settings),
            )
    _warn_on_model_drift(models, settings.EMBEDDING_MODEL)
    written = bundle.result
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "exported %d document(s) and %d chunk(s) to %s (%d bytes) in %.1fs",
        documents,
        chunks,
        written.name,
        written.size_bytes,
        wall_clock_s,
    )
    return RagExport(bundle=written, wall_clock_s=wall_clock_s)

ingest_drift(manifest, current)

Name the reingest-affecting settings a RAG bundle differs from this stack on.

Parameters:

Name	Type	Description	Default
manifest	BundleManifest	The bundle's manifest.	required
current	Mapping[str, str]	This stack's reingest-affecting settings, in env-string form (varagity.api.backup_runner.current_ingest_settings).	required

Returns:

Type	Description
list[str]	The differing setting names, sorted — a setting recorded on only one
list[str]	side counts as differing, and a bundle that recorded no settings
list[str]	differs on all of them.

Source code in varagity/backup/rag.py

def ingest_drift(manifest: BundleManifest, current: Mapping[str, str]) -> list[str]:
    """Name the reingest-affecting settings a RAG bundle differs from this stack on.

    Args:
        manifest: The bundle's manifest.
        current: This stack's reingest-affecting settings, in env-string
            form (``varagity.api.backup_runner.current_ingest_settings``).

    Returns:
        The differing setting names, sorted — a setting recorded on only one
        side counts as differing, and a bundle that recorded no settings
        differs on all of them.
    """
    recorded = manifest.ingest_settings
    if recorded is None:
        return sorted(current)
    return sorted(
        name for name in set(recorded) | set(current) if recorded.get(name) != current.get(name)
    )

restore_rag(bundle, store, bm25, docs_root, *, on_progress=None)

Replace the chunk corpus with a bundle's rows, then rebuild the BM25 index.

No model is called — nothing is parsed, chunked, contextualized or embedded. Under the corpus's exclusive advisory lock, held until the BM25 rebuild ends (so no ingest in any process can write rows the rebuild would miss, or touch the index while it is gone):

1. validate — the cheap verdict (the route checked it before the run; this guards a direct call and a file swapped since).
2. load — one transaction empties both tables and streams the bundle's lines into binary COPY, re-rooting each source; every member's sha256 is checked as it is read — the restore's one read of the bundle, and its only hash — so a damaged bundle (or one a line of which no row can be made from) can never commit.
3. index — the HNSW index is rebuilt and the transaction commits: the restore has taken effect. Any failure before this rolled back, leaving the previous corpus untouched.
4. bm25 — the index is recreated and refilled from the restored rows.

Parameters:

Name	Type	Description	Default
bundle	Path	The bundle file.	required
store	ContextualVectorDB	The vector store to replace (its connection holds the lock).	required
bm25	ElasticsearchBM25	The BM25 store to rebuild.	required
docs_root	Path	The restoring stack's DOCS_PATH (re-rooting target).	required
on_progress	RestoreProgressCallback | None	Called with (stage, current, total) per stage; the load and bm25 stages count chunks.	None

Returns:

Type	Description
RagRestore	What was restored, and how long it took.

Raises:

Type	Description
BundleInvalid	If the bundle is unsound or damaged (nothing was replaced — the transaction rolled back).
BundleIncompatible	If this stack must not restore it (nothing was replaced).
CorpusLocked	If an ingest holds the corpus lock (nothing was replaced).
BM25RebuildFailed	If the rows were restored but the BM25 index could not be rebuilt.

Source code in varagity/backup/rag.py

def restore_rag(
    bundle: Path,
    store: ContextualVectorDB,
    bm25: ElasticsearchBM25,
    docs_root: Path,
    *,
    on_progress: RestoreProgressCallback | None = None,
) -> RagRestore:
    """Replace the chunk corpus with a bundle's rows, then rebuild the BM25 index.

    No model is called — nothing is parsed, chunked, contextualized or
    embedded. Under the corpus's **exclusive** advisory lock, held until the
    BM25 rebuild ends (so no ingest in any process can write rows the
    rebuild would miss, or touch the index while it is gone):

    1. ``validate`` — the cheap verdict (the route checked it before the
       run; this guards a direct call and a file swapped since).
    2. ``load`` — one transaction empties both tables and streams the
       bundle's lines into binary ``COPY``, re-rooting each ``source``;
       every member's sha256 is checked as it is read — the restore's one
       read of the bundle, and its only hash — so a damaged bundle (or one
       a line of which no row can be made from) can never commit.
    3. ``index`` — the HNSW index is rebuilt and the transaction commits:
       the restore has taken effect. Any failure before this rolled back,
       leaving the previous corpus untouched.
    4. ``bm25`` — the index is recreated and refilled from the restored
       rows.

    Args:
        bundle: The bundle file.
        store: The vector store to replace (its connection holds the lock).
        bm25: The BM25 store to rebuild.
        docs_root: The restoring stack's ``DOCS_PATH`` (re-rooting target).
        on_progress: Called with ``(stage, current, total)`` per stage; the
            ``load`` and ``bm25`` stages count chunks.

    Returns:
        What was restored, and how long it took.

    Raises:
        BundleInvalid: If the bundle is unsound or damaged (nothing was
            replaced — the transaction rolled back).
        BundleIncompatible: If this stack must not restore it (nothing was
            replaced).
        varagity.stores.vector_store.CorpusLocked: If an ingest holds the
            corpus lock (nothing was replaced).
        BM25RebuildFailed: If the rows were restored but the BM25 index
            could not be rebuilt.
    """
    settings = get_settings()
    report: RestoreProgressCallback = on_progress if on_progress is not None else _ignore
    dim = settings.EMBEDDING_DIM
    root = docs_root.resolve()
    started = time.perf_counter()

    report("validate", None, None)
    manifest = inspect_bundle(bundle, "rag").manifest_or_raise()
    size_bytes = bundle.stat().st_size
    expected_documents = manifest.counts["documents"]
    expected_chunks = manifest.counts["chunks"]
    logger.info(
        "restoring %d document(s) and %d chunk(s) from %s",
        expected_documents,
        expected_chunks,
        bundle.name,
    )
    with store.corpus_lock(exclusive=True):
        sources: dict[str, str] = {}
        with open_bundle(bundle, manifest) as source:
            documents, chunks = store.replace_corpus(
                _document_rows(source, root, sources, expected_documents),
                _chunk_rows(
                    source,
                    sources,
                    dim=dim,
                    expected=expected_chunks,
                    progress=_Progress(report, "load", expected_chunks),
                ),
                on_indexing=lambda: report("index", None, None),
            )
        logger.info("loaded %d document(s) and %d chunk(s) — committed", documents, chunks)
        progress = _Progress(report, "bm25", chunks)
        try:
            with closing(store.iter_index_rows()) as rows:
                indexed = bm25.rebuild(rows, on_indexed=progress.tick)
        except Exception as error:
            raise BM25RebuildFailed(
                f"the rows are restored, but the BM25 index could not be rebuilt "
                f"({type(error).__name__}: {error}) — restore the same bundle again to "
                "repair it",
                manifest,
            ) from error
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "restored %d document(s) and %d chunk(s) from %s in %.1fs; BM25 holds %d",
        documents,
        chunks,
        bundle.name,
        wall_clock_s,
        indexed,
    )
    return RagRestore(
        bundle_name=bundle.name,
        size_bytes=size_bytes,
        manifest=manifest,
        documents=documents,
        chunks=chunks,
        bm25_indexed=indexed,
        wall_clock_s=wall_clock_s,
    )

restored_source(source, relative_path, docs_root)

Re-root one document's source onto the restoring stack's corpus directory.

doc_id already hashes the corpus-relative path, so the row's identity is portable; source (the absolute path the readers resolve — the document list, previews, "also remove the file") is rebuilt the way ingest builds it, (DOCS_PATH / relative_path).resolve(). A document exported from outside the corpus has no relative path and keeps its original absolute source.

Parameters:

Name	Type	Description	Default
source	str	The source recorded at export.	required
relative_path	str | None	Its path relative to the exporting DOCS_PATH, or None.	required
docs_root	Path	The resolved DOCS_PATH of the restoring stack.	required

Returns:

Type	Description
str	The source to restore.

Raises:

Type	Description
BundleInvalid	If the relative path is absolute, empty, or climbs out of the corpus (..) — no export writes one.

Source code in varagity/backup/rag.py

def restored_source(source: str, relative_path: str | None, docs_root: Path) -> str:
    """Re-root one document's ``source`` onto the restoring stack's corpus directory.

    ``doc_id`` already hashes the corpus-relative path, so the row's
    identity is portable; ``source`` (the absolute path the readers resolve
    — the document list, previews, "also remove the file") is rebuilt the
    way ingest builds it, ``(DOCS_PATH / relative_path).resolve()``. A
    document exported from outside the corpus has no relative path and
    keeps its original absolute ``source``.

    Args:
        source: The ``source`` recorded at export.
        relative_path: Its path relative to the exporting ``DOCS_PATH``, or
            ``None``.
        docs_root: The resolved ``DOCS_PATH`` of the restoring stack.

    Returns:
        The ``source`` to restore.

    Raises:
        BundleInvalid: If the relative path is absolute, empty, or climbs
            out of the corpus (``..``) — no export writes one.
    """
    if relative_path is None:
        return source
    relative = PurePosixPath(relative_path)
    if relative.is_absolute() or not relative.parts or ".." in relative.parts:
        raise BundleInvalid(
            f"{DOCUMENTS_MEMBER} records relative_path {relative_path!r}, which is not a path "
            "inside the corpus."
        )
    return str(docs_root.joinpath(*relative.parts).resolve())

vector_bytes(vector, dim, chunk_id)

Encode one vector as a little-endian float32 matrix row.

pgvector's binary form is a >HH header (dimensions, unused) and big-endian float4s; one byte swap turns that payload into the little-endian row the bundle stores, on any host, without a decimal round trip — the bits are the stored bits.

Parameters:

Name	Type	Description	Default
vector	Vector	The stored vector.	required
dim	int	The dimension the bundle declares (EMBEDDING_DIM).	required
chunk_id	str	The owning chunk (for the error message).	required

Returns:

Type	Description
bytes	dim × 4 bytes.

Raises:

Type	Description
ValueError	If the vector's dimension isn't dim — the matrix would no longer be rectangular, and the manifest would lie.

Source code in varagity/backup/rag.py

def vector_bytes(vector: Vector, dim: int, chunk_id: str) -> bytes:
    """Encode one vector as a little-endian float32 matrix row.

    pgvector's binary form is a ``>HH`` header (dimensions, unused) and
    big-endian float4s; one byte swap turns that payload into the
    little-endian row the bundle stores, on any host, without a decimal
    round trip — the bits are the stored bits.

    Args:
        vector: The stored vector.
        dim: The dimension the bundle declares (``EMBEDDING_DIM``).
        chunk_id: The owning chunk (for the error message).

    Returns:
        ``dim × 4`` bytes.

    Raises:
        ValueError: If the vector's dimension isn't ``dim`` — the matrix
            would no longer be rectangular, and the manifest would lie.
    """
    raw = vector.to_binary()
    dimensions = int.from_bytes(raw[:2], "big")
    if dimensions != dim:
        raise ValueError(
            f"chunk {chunk_id} has a {dimensions}-dimensional embedding, but EMBEDDING_DIM is {dim}"
        )
    values = array("f")
    values.frombytes(raw[4:])
    values.byteswap()
    return values.tobytes()

vector_from_bytes(raw, dim)

Decode one little-endian float32 matrix row back into a vector, bit-exact.

The inverse of :func:vector_bytes: one byte swap turns the row into pgvector's big-endian payload, on any host.

Parameters:

Name	Type	Description	Default
raw	bytes	dim × 4 bytes of embeddings.f32.	required
dim	int	The vector's dimension.	required

Returns:

Type	Description
Vector	The vector.

Source code in varagity/backup/rag.py

def vector_from_bytes(raw: bytes, dim: int) -> Vector:
    """Decode one little-endian float32 matrix row back into a vector, bit-exact.

    The inverse of :func:`vector_bytes`: one byte swap turns the row into
    pgvector's big-endian payload, on any host.

    Args:
        raw: ``dim × 4`` bytes of ``embeddings.f32``.
        dim: The vector's dimension.

    Returns:
        The vector.
    """
    values = array("f")
    values.frombytes(raw)
    values.byteswap()
    return Vector.from_binary(dim.to_bytes(2, "big") + b"\x00\x00" + values.tobytes())

varagity.backup.graph

The graph bundle: the engine's working directory, verbatim (ADR-019).

A graph backup is the smallest thing that restores the message graph with no model call — and for the graph that is the whole working directory. Its files are the extracted graph itself, its vector indexes, the extracted text and chunk tracking, the doc-status resume ledger, and the LLM response cache that a later incremental build rebuilds shared entities from; beside them sit the repo's two sidecars, the build manifest that keeps the next build a diffing upsert and the size summary. Nothing is converted or re-derived — every file goes into the bundle byte for byte — so a bundle is tied to the engine version that wrote it, which is why its manifest records that engine's file format (:attr:~varagity.backup.bundle.BundleManifest.engine).

The bundle's data members, under one folder named for the engine::

lightrag/                             the workdir verbatim, *.tmp* skipped   deflated
├── kv_store_*.json                   LightRAG's key-value stores (doc status, the LLM cache …)
├── vdb_*.json                        its vector indexes
├── graph_chunk_entity_relation.graphml
├── varagity_manifest.json            the build diff
└── varagity_graph_summary.json       the size cache — the bundle's counts

The write lock does the hard part. :func:export_graph copies the files under :meth:~varagity.graph.service.GraphService.writing, the lock every build, resume and delete takes, so no build can change the graph mid-copy — while reads take no lock, so chat and /graph keep answering throughout. A read does write one file: LightRAG flushes its LLM response cache to disk after every query. That is harmless, because the engine writes each file to a *.tmp* sibling and renames it into place: every member is a whole file, from before a query or after it, and a cache one query newer than the rest of the graph is still a valid cache. The *.tmp* siblings are never copied.

The stale flag is read last. Deleting a graph source flags the graph stale without taking any lock, so the flag is read once every file is copied, just before the manifest is sealed: a source deleted mid-export is then recorded. Only a completed rebuild clears the flag, and the corpus gate keeps rebuilds out for the whole export, so a late read is never less accurate than an early one.

A restore is extract, swap and open (:func:restore_graph), with no model call. The bundle's files are extracted — each member's sha256 checked again as it is read — into a staging folder beside the workdir, on the same volume, and :meth:~varagity.graph.service.GraphService.replace_workdir swaps it in under the service's write and open locks, opens it, and puts the previous graph back if it won't open. The graph-stale flag the restore leaves is the bundle's own (the graph-restore flow writes it).

GraphExportProgressCallback = Callable[[GraphExportStage, int | None, int | None], None] module-attribute

(stage, current, total) — how a graph export reports where it is (pack counts bytes).

GraphExportStage = Literal['snapshot', 'pack', 'finalize'] module-attribute

The graph export's progress stages, in order.

GraphRestoreProgressCallback = Callable[[GraphRestoreStage, int | None, int | None], None] module-attribute

(stage, current, total) — how a graph restore reports progress (extract counts bytes).

GraphRestoreStage = Literal['validate', 'extract', 'swap', 'open'] module-attribute

The graph restore's progress stages, in order.

GraphExportReport dataclass

What one graph export wrote.

Attributes:

Name	Type	Description
bundle	WrittenBundle	The published bundle (name, path, size, manifest).
files	int	How many workdir files it holds.
wall_clock_s	float	Seconds from the export's start (the write lock is taken at once) to the published bundle.

Source code in varagity/backup/graph.py

@dataclass(frozen=True)
class GraphExportReport:
    """What one graph export wrote.

    Attributes:
        bundle: The published bundle (name, path, size, manifest).
        files: How many workdir files it holds.
        wall_clock_s: Seconds from the export's start (the write lock is
            taken at once) to the published bundle.
    """

    bundle: WrittenBundle
    files: int
    wall_clock_s: float

    @property
    def counts(self) -> dict[str, int]:
        """What the bundle holds, by the graph's own summary (see :func:`graph_counts`).

        Returns:
            The manifest's counts.
        """
        return dict(self.bundle.manifest.counts)

counts property

What the bundle holds, by the graph's own summary (see :func:graph_counts).

Returns:

Type	Description
dict[str, int]	The manifest's counts.

GraphRestore dataclass

What one graph restore did.

Attributes:

Name	Type	Description
bundle_name	str	The restored bundle's file name.
size_bytes	int	The bundle's size on disk.
manifest	BundleManifest	Its manifest.
files	int	How many workdir files it restored.
stats	GraphStats	The restored graph's size, as its freshly opened session reported it.
wall_clock_s	float	Seconds from validation to an open graph.
stale	bool | None	The graph-stale flag the restore left — the bundle's own, written by the graph-restore flow; None until it has.

Source code in varagity/backup/graph.py

@dataclass(frozen=True)
class GraphRestore:
    """What one graph restore did.

    Attributes:
        bundle_name: The restored bundle's file name.
        size_bytes: The bundle's size on disk.
        manifest: Its manifest.
        files: How many workdir files it restored.
        stats: The restored graph's size, as its freshly opened session
            reported it.
        wall_clock_s: Seconds from validation to an open graph.
        stale: The graph-stale flag the restore left — the bundle's own,
            written by the ``graph-restore`` flow; ``None`` until it has.
    """

    bundle_name: str
    size_bytes: int
    manifest: BundleManifest
    files: int
    stats: GraphStats
    wall_clock_s: float
    stale: bool | None = None

    @property
    def counts(self) -> dict[str, int]:
        """What the restored graph holds, by the bundle's manifest (its size summary at export).

        Returns:
            ``thread_days``, ``messages`` and — when the graph reported them
            at export — ``entities`` and ``relations``.
        """
        return dict(self.manifest.counts)

counts property

What the restored graph holds, by the bundle's manifest (its size summary at export).

Returns:

Type	Description
dict[str, int]	thread_days, messages and — when the graph reported them
dict[str, int]	at export — entities and relations.

export_graph(service, dest, *, read_stale, on_progress=None)

Export the message graph's working directory into a new bundle in dest.

Holds the service's write lock for the whole copy — no build, resume or delete can start until the bundle is published, while reads (graph chat, /graph) are answered as usual. No model is called and no session is opened: the files are copied as they are on disk, which is exactly what an API restart would load.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (its workdir, write lock, and engine format).	required
dest	Path	The exports folder (<GRAPH_DOCS_PATH>/varagity-exports).	required
read_stale	Callable[[], bool]	Reads the graph-stale flag (the flow passes a Postgres read). Called once, after the last file is copied and before the manifest is sealed, so a source deleted mid-export is recorded; if it raises, the export fails and leaves no bundle.	required
on_progress	GraphExportProgressCallback | None	Called with (stage, current, total) as the export moves through snapshot (take the lock, list the files) → pack (bytes copied of the files' total) → finalize.	None

Returns:

Type	Description
GraphExportReport	The published bundle and how long it took.

Raises:

Type	Description
GraphBuildInProgress	If a build, resume or delete holds the write lock.
NothingToExport	If the workdir holds no file to copy (nothing has been built).
ValueError	If the engine's name could not be the bundle's folder.
OSError	If a file can't be read or the bundle can't be written.

Source code in varagity/backup/graph.py

def export_graph(
    service: GraphService,
    dest: Path,
    *,
    read_stale: Callable[[], bool],
    on_progress: GraphExportProgressCallback | None = None,
) -> GraphExportReport:
    """Export the message graph's working directory into a new bundle in ``dest``.

    Holds the service's write lock for the whole copy — no build, resume or
    delete can start until the bundle is published, while reads (graph chat,
    ``/graph``) are answered as usual. No model is called and no session is
    opened: the files are copied as they are on disk, which is exactly what
    an API restart would load.

    Args:
        service: The process's graph service (its workdir, write lock, and
            engine format).
        dest: The exports folder (``<GRAPH_DOCS_PATH>/varagity-exports``).
        read_stale: Reads the graph-stale flag (the flow passes a Postgres
            read). Called once, after the last file is copied and before the
            manifest is sealed, so a source deleted mid-export is recorded;
            if it raises, the export fails and leaves no bundle.
        on_progress: Called with ``(stage, current, total)`` as the export
            moves through ``snapshot`` (take the lock, list the files) →
            ``pack`` (bytes copied of the files' total) → ``finalize``.

    Returns:
        The published bundle and how long it took.

    Raises:
        varagity.graph.service.GraphBuildInProgress: If a build, resume or
            delete holds the write lock.
        NothingToExport: If the workdir holds no file to copy (nothing has
            been built).
        ValueError: If the engine's name could not be the bundle's folder.
        OSError: If a file can't be read or the bundle can't be written.
    """
    settings = get_settings()
    report: GraphExportProgressCallback = on_progress if on_progress is not None else _ignore
    started = time.perf_counter()
    created_at = datetime.now(UTC)

    report("snapshot", None, None)
    with service.writing():
        files, skipped = _workdir_entries(service.workdir)
        for name, why in skipped:
            logger.warning("not exporting %s from the graph workdir — %s", name, why)
        if not files:
            raise NothingToExport(
                "nothing has been built into the graph — there is nothing to export"
            )
        engine = service.storage_format()
        if not is_graph_file_name(engine.name):
            raise ValueError(f"the graph engine {engine.name!r} can't name a bundle folder")
        counts = graph_counts(service.workdir)
        total = sum(path.stat().st_size for path in files)
        # File names only in these lines: they are relayed into the panel's
        # log, where an in-container path (/app/graph-data/…) would read as
        # nonsense beside the checkout-relative "Saved to" path.
        logger.info(
            "exporting %d graph file(s), %s bytes, written by %s %s",
            len(files),
            f"{total:,}",
            engine.package,
            engine.version,
        )
        with write_bundle(dest, "graph", created_at=created_at) as bundle:
            progress = _ByteProgress(report, total)
            for path in files:
                with (
                    path.open("rb") as handle,
                    bundle.member(f"{engine.name}/{path.name}") as member,
                ):
                    while block := handle.read(_COPY_BYTES):
                        member.write(block)
                        progress.advance(len(block))
            progress.finish()
            report("finalize", None, None)
            # Last, not first: a source deleted during the copy has flagged
            # the graph stale by now (see the module docstring).
            stale = read_stale()
            bundle.seal(
                embedding_model=settings.EMBEDDING_MODEL,
                embedding_dim=settings.EMBEDDING_DIM,
                stale=stale,
                counts=counts,
                engine=engine,
            )
    written = bundle.result
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "exported %d graph file(s) to %s (%s bytes) in %.1fs",
        len(files),
        written.name,
        f"{written.size_bytes:,}",
        wall_clock_s,
    )
    return GraphExportReport(bundle=written, files=len(files), wall_clock_s=wall_clock_s)

graph_counts(workdir)

What a graph bundle holds, by the workdir's own size summary.

The summary sidecar is the graph's size cache (what the status route and the gauges read) — never the graphml, which a count would have to parse.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory.	required

Returns:

Type	Description
dict[str, int]	thread_days (transcript documents) and messages (distinct
dict[str, int]	guids), plus entities and relations when the graph reported
dict[str, int]	them — or nothing at all when no build has written a summary yet (a
dict[str, int]	first build killed partway leaves files, but no summary).

Source code in varagity/backup/graph.py

def graph_counts(workdir: Path) -> dict[str, int]:
    """What a graph bundle holds, by the workdir's own size summary.

    The summary sidecar is the graph's size cache (what the status route and
    the gauges read) — never the graphml, which a count would have to parse.

    Args:
        workdir: The engine's working directory.

    Returns:
        ``thread_days`` (transcript documents) and ``messages`` (distinct
        guids), plus ``entities`` and ``relations`` when the graph reported
        them — or nothing at all when no build has written a summary yet (a
        first build killed partway leaves files, but no summary).
    """
    summary = load_summary(workdir)
    if summary is None:
        return {}
    counts = {"thread_days": summary.docs, "messages": summary.message_guids}
    if summary.entities is not None:
        counts["entities"] = summary.entities
    if summary.relations is not None:
        counts["relations"] = summary.relations
    return counts

is_temp_file(name)

Report whether a workdir file is an engine's in-flight write.

LightRAG writes <file>.tmp.<pid>.<thread>.<ns> beside a file and renames it into place; the repo's sidecars use <file>.<pid>.tmp. A leftover one is a crash's debris, never part of the graph.

Parameters:

Name	Type	Description	Default
name	str	A file name in the workdir.	required

Returns:

Type	Description
bool	True for any name containing .tmp (the *.tmp* pattern).

Source code in varagity/backup/graph.py

def is_temp_file(name: str) -> bool:
    """Report whether a workdir file is an engine's in-flight write.

    LightRAG writes ``<file>.tmp.<pid>.<thread>.<ns>`` beside a file and
    renames it into place; the repo's sidecars use ``<file>.<pid>.tmp``. A
    leftover one is a crash's debris, never part of the graph.

    Args:
        name: A file name in the workdir.

    Returns:
        ``True`` for any name containing ``.tmp`` (the ``*.tmp*`` pattern).
    """
    return ".tmp" in name

restore_graph(service, bundle, *, run_id, on_progress=None)

Replace the message graph's working directory with a bundle's.

No model is called and nothing is re-extracted — the bundle is a workdir:

1. validate — the cheap verdict (the route checked it before the run; this guards a direct call and a file swapped since);
2. extract — every file into a fresh staging folder beside the workdir (:meth:~varagity.graph.service.GraphService.staging_dir, on the same volume), each member's sha256 checked as it is read — the restore's one read of the bundle, and its only hash — and each file synced to disk, so a damaged bundle can never go live;
3. swap and open — :meth:~varagity.graph.service.GraphService.replace_workdir closes the session, swaps the folders under the service's write and open locks, and opens the restored graph; if it won't open, the previous graph is put back.

Whatever is left of the staging folder is removed however the restore ends; until the swap, the live graph is untouched.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (the workdir's single writer).	required
bundle	Path	The bundle file.	required
run_id	str	The restore run's id (names the folders beside the workdir).	required
on_progress	GraphRestoreProgressCallback | None	Called with (stage, current, total) per stage; extract counts bytes.	None

Returns:

Type	Description
GraphRestore	What was restored, and how long it took.

Raises:

Type	Description
BundleInvalid	If the bundle is unsound or damaged (the live graph is untouched).
BundleIncompatible	If this stack must not restore it (untouched).
GraphBuildInProgress	If a write holds the graph's write lock at the swap (untouched).
GraphUnavailable	If the restored graph would not open (the previous graph is back in place).
OSError	If a file can't be extracted or a folder can't be moved (the previous graph stays, or is put back).

Source code in varagity/backup/graph.py

def restore_graph(
    service: GraphService,
    bundle: Path,
    *,
    run_id: str,
    on_progress: GraphRestoreProgressCallback | None = None,
) -> GraphRestore:
    """Replace the message graph's working directory with a bundle's.

    No model is called and nothing is re-extracted — the bundle *is* a
    workdir:

    1. ``validate`` — the cheap verdict (the route checked it before the
       run; this guards a direct call and a file swapped since);
    2. ``extract`` — every file into a fresh staging folder beside the
       workdir (:meth:`~varagity.graph.service.GraphService.staging_dir`, on
       the same volume), each member's sha256 checked as it is read — the
       restore's one read of the bundle, and its only hash — and each file
       synced to disk, so a damaged bundle can never go live;
    3. ``swap`` and ``open`` —
       :meth:`~varagity.graph.service.GraphService.replace_workdir` closes
       the session, swaps the folders under the service's write and open
       locks, and opens the restored graph; if it won't open, the previous
       graph is put back.

    Whatever is left of the staging folder is removed however the restore
    ends; until the swap, the live graph is untouched.

    Args:
        service: The process's graph service (the workdir's single writer).
        bundle: The bundle file.
        run_id: The restore run's id (names the folders beside the workdir).
        on_progress: Called with ``(stage, current, total)`` per stage;
            ``extract`` counts bytes.

    Returns:
        What was restored, and how long it took.

    Raises:
        BundleInvalid: If the bundle is unsound or damaged (the live graph
            is untouched).
        BundleIncompatible: If this stack must not restore it (untouched).
        varagity.graph.service.GraphBuildInProgress: If a write holds the
            graph's write lock at the swap (untouched).
        varagity.graph.service.GraphUnavailable: If the restored graph would
            not open (the previous graph is back in place).
        OSError: If a file can't be extracted or a folder can't be moved
            (the previous graph stays, or is put back).
    """
    report: GraphRestoreProgressCallback = on_progress if on_progress is not None else _ignore
    started = time.perf_counter()

    report("validate", None, None)
    manifest = inspect_bundle(bundle, "graph").manifest_or_raise()
    folder = f"{graph_folder(manifest)}/"
    engine = manifest.engine
    size_bytes = bundle.stat().st_size
    total = sum(member.size_bytes for member in manifest.members)
    # File names only, as the export logs: these lines are relayed into the
    # panel, where an in-container path would read as nonsense.
    logger.info(
        "restoring %d graph file(s), %s bytes, written by %s, from %s",
        len(manifest.members),
        f"{total:,}",
        f"{engine.package} {engine.version}" if engine is not None else "an unknown engine",
        bundle.name,
    )
    staged = service.staging_dir(run_id)
    staged.mkdir(parents=True)  # a fresh folder, never one that already holds files
    try:
        progress = _ByteProgress(report, total, stage="extract")
        with open_bundle(bundle, manifest) as source:
            for member in manifest.members:
                # The verdict held every member to "<folder><flat safe name>".
                target = staged / member.name.removeprefix(folder)
                with source.member(member.name) as stream, target.open("xb") as out:
                    while block := stream.read(_COPY_BYTES):
                        out.write(block)
                        progress.advance(len(block))
                    out.flush()
                    os.fsync(out.fileno())
        progress.finish()
        report("swap", None, None)
        try:
            stats = service.replace_workdir(
                staged, run_id, on_open=lambda: report("open", None, None)
            )
        except GraphUnavailable:
            logger.warning(
                "the restored graph would not open, so the restore was undone — the graph is "
                "as it was before"
            )
            raise
    finally:
        # Gone already after a swap; a failed restore's extracted files otherwise.
        shutil.rmtree(staged, ignore_errors=True)
    wall_clock_s = time.perf_counter() - started
    logger.info(
        "restored %d graph file(s) from %s in %.1fs",
        len(manifest.members),
        bundle.name,
        wall_clock_s,
    )
    return GraphRestore(
        bundle_name=bundle.name,
        size_bytes=size_bytes,
        manifest=manifest,
        files=len(manifest.members),
        stats=stats,
        wall_clock_s=wall_clock_s,
    )

workdir_files(workdir)

List the files a graph export would copy.

What the export route's nothing_to_export check asks — the same rule the export applies, so the route and the run can never disagree on whether there is anything to back up.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory (a missing one holds none).	required

Returns:

Type	Description
list[Path]	The workdir's own files, in name order, temp files and anything a
list[Path]	bundle can't carry left out.

Source code in varagity/backup/graph.py

def workdir_files(workdir: Path) -> list[Path]:
    """List the files a graph export would copy.

    What the export route's ``nothing_to_export`` check asks — the same
    rule the export applies, so the route and the run can never disagree on
    whether there is anything to back up.

    Args:
        workdir: The engine's working directory (a missing one holds none).

    Returns:
        The workdir's own files, in name order, temp files and anything a
        bundle can't carry left out.
    """
    return _workdir_entries(workdir)[0]