GraphRAG

The graph subsystem (spec_graphrag §5.2, §10; ADR-017) — a peer retrieval corpus beside chunk-RAG, not a sixth retriever: structured messages (an iMessage chat.db in v1) become a knowledge graph with per-message provenance.

Stage 1 shipped the engine-independent foundations plus the bake-off harness that decided the engine; stage 2 productionized the winner — the resumable build behind API routes, the graph query path, the web surfaces (Corpus tab, source selector, evidence, and the /graph view), and the workdir-derived gauges.

• Message sources are a registry family chosen by structural dispatch — a file goes to the first source whose matches() accepts it (the parser precedent), so there is no config.py vocabulary tuple to keep in lockstep. v1 registers only imessage.
• Graph engines are a registry family in the usual @register("name") shape, selected by GRAPH_ENGINE. ADR-017 picked lightrag and the losing adapters were deleted at stage-2 start; the seam is what stays, so the decision remains benchmark-revisitable. Importing the package must not import an engine library — the adapter imports lightrag inside session(), so the registry stays free even though the engine is a main dependency.
• Rendering is pure and shared: messages become thread-day transcripts, and identical thread-days always render to an identical doc_key and text — the property the build diff rests on.
• The workdir owns two sidecars. varagity_manifest.json records doc_key → {content hash, message guids, thread, span}, which makes a build a diffing upsert (LightRAG's enqueue dedup would otherwise keep a stale transcript forever) and keeps provenance durable across restarts. varagity_graph_summary.json caches the graph's size so status polls and metric scrapes never walk the graphml.
• One process is the single writer. varagity.graph.service holds the lone session, guards writes with a single-flight lock, and lets reads through unlocked — the session runs its event loop on its own thread, so a chat query is answered while a multi-day backfill is still extracting.
• Answer synthesis is the repo's, not the engine's (ADR-017 chose retrieval-only): the adapter returns entities, relations, and transcript excerpts, and varagity.graph.answer writes the grounded, capped, <think>-stripped answer over them.

Message sources

varagity.graph.sources.base

Message-source protocol and registry (spec_graphrag §5.2, §10.2; the spec §5.1 pattern).

Each message-source module defines one implementation decorated with @register("name"); unlike the chat/retriever registries (settings-field selection), sources are chosen by structural dispatch — the parser precedent (varagity.ingest.loader._BUCKET_PARSERS): a file is routed to the first source whose :meth:MessageSource.matches accepts it, so there is no config.py vocabulary tuple to keep in lockstep. v1 registers only imessage; adding a platform later is one new module plus its import line in :mod:varagity.graph.sources, zero caller edits.

A message source produces structured messages (speaker, timestamp, thread, per-message identity) — not prose for chunking — so the family lives under :mod:varagity.graph, not :mod:varagity.ingest. The models are pydantic (not dataclasses) because the fixture generator, eval results, and stage-2 wire payloads all serialize them.

MessageBatch

Bases: BaseModel

A parsed source file: its stable identity plus every message in it.

The identity recipe is the corpus recipe reused verbatim (:func:varagity.stores.records.derive_doc_id over the path relative to GRAPH_DOCS_PATH plus the file's byte hash), so a graph document's doc_id is portable across host/container/machines exactly as a RAG document's is.

Attributes:

Name	Type	Description
doc_id	str	Stable id of the source file (see :func:batch_for_path).
relative_path	str	POSIX path of the file relative to the graph root.
messages	list[SourceMessage]	Every parsed message, oldest first.

Source code in varagity/graph/sources/base.py

class MessageBatch(BaseModel):
    """A parsed source file: its stable identity plus every message in it.

    The identity recipe is the corpus recipe reused verbatim
    (:func:`varagity.stores.records.derive_doc_id` over the path relative to
    ``GRAPH_DOCS_PATH`` plus the file's byte hash), so a graph document's
    ``doc_id`` is portable across host/container/machines exactly as a RAG
    document's is.

    Attributes:
        doc_id: Stable id of the source file (see :func:`batch_for_path`).
        relative_path: POSIX path of the file relative to the graph root.
        messages: Every parsed message, oldest first.
    """

    doc_id: str
    relative_path: str
    messages: list[SourceMessage]

MessageSource

Bases: Protocol

Parses one file format into :class:SourceMessage objects.

runtime_checkable mirrors :class:~varagity.chat.base.ChatEngine: the protocol will appear in stage-2 Prefect flow signatures, and Prefect builds a pydantic parameter schema from the annotations at decoration time, which requires types usable with isinstance.

Source code in varagity/graph/sources/base.py

@runtime_checkable
class MessageSource(Protocol):
    """Parses one file format into :class:`SourceMessage` objects.

    ``runtime_checkable`` mirrors :class:`~varagity.chat.base.ChatEngine`:
    the protocol will appear in stage-2 Prefect flow signatures, and Prefect
    builds a pydantic parameter schema from the annotations at decoration
    time, which requires types usable with ``isinstance``.
    """

    def matches(self, path: Path) -> bool:
        """Report whether this source can parse ``path`` (a cheap probe).

        Args:
            path: Candidate file.

        Returns:
            ``True`` if the file is this source's format. Must never raise —
            a probe that can't read the file returns ``False``.
        """
        ...

    def parse(self, path: Path, *, verbose: int = 0) -> list[SourceMessage]:
        """Parse ``path`` into normalized messages, oldest first.

        Called only after :meth:`matches` has accepted ``path``.

        Args:
            path: File to parse (its format already confirmed by
                :meth:`matches`).
            verbose: Validated console verbosity (0–2).

        Returns:
            Every recoverable message, oldest first (unrecoverable rows are
            counted and skipped, never emitted with empty text).
        """
        ...

matches(path)

Report whether this source can parse path (a cheap probe).

Parameters:

Name	Type	Description	Default
path	Path	Candidate file.	required

Returns:

Type	Description
bool	True if the file is this source's format. Must never raise —
bool	a probe that can't read the file returns False.

Source code in varagity/graph/sources/base.py

def matches(self, path: Path) -> bool:
    """Report whether this source can parse ``path`` (a cheap probe).

    Args:
        path: Candidate file.

    Returns:
        ``True`` if the file is this source's format. Must never raise —
        a probe that can't read the file returns ``False``.
    """
    ...

parse(path, *, verbose=0)

Parse path into normalized messages, oldest first.

Called only after :meth:matches has accepted path.

Parameters:

Name	Type	Description	Default
path	Path	File to parse (its format already confirmed by :meth:matches).	required
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
list[SourceMessage]	Every recoverable message, oldest first (unrecoverable rows are
list[SourceMessage]	counted and skipped, never emitted with empty text).

Source code in varagity/graph/sources/base.py

def parse(self, path: Path, *, verbose: int = 0) -> list[SourceMessage]:
    """Parse ``path`` into normalized messages, oldest first.

    Called only after :meth:`matches` has accepted ``path``.

    Args:
        path: File to parse (its format already confirmed by
            :meth:`matches`).
        verbose: Validated console verbosity (0–2).

    Returns:
        Every recoverable message, oldest first (unrecoverable rows are
        counted and skipped, never emitted with empty text).
    """
    ...

SourceMessage

Bases: BaseModel

One normalized message from any message source (spec_graphrag §10.2).

Attributes:

Name	Type	Description
guid	str	Stable per-message identity — message.guid for iMessage, never the volatile ROWID (a rebuilt/merged database keeps the guid). Sources without native ids hash thread+sender+time+text.
thread_id	str	Stable thread identity (chat.guid for iMessage).
thread_name	str	Human-facing thread label: the chat's display_name when set, else the sorted participant names (group chats often have an empty display_name).
sender_handle	str	Raw handle (phone number / email); "" for the owner (whose handle is not modelled — ego is structural).
sender_name	str	Mapped display name (GRAPH_HANDLE_NAMES), falling back to the raw handle; the owner label when :attr:is_from_me.
is_from_me	bool	Whether the owner sent it (message.is_from_me — the structural ego marker that Q3-class attribution needs).
timestamp	datetime	Send time as a timezone-aware UTC datetime (both Apple epoch eras decoded to the same wall clock).
text	str	The decoded message body; never empty (rows with no recoverable text are skipped at parse time, not emitted blank).
tapbacks	list[Tapback]	Reactions folded onto this message (empty when none).

Source code in varagity/graph/sources/base.py

class SourceMessage(BaseModel):
    """One normalized message from any message source (spec_graphrag §10.2).

    Attributes:
        guid: Stable per-message identity — ``message.guid`` for iMessage,
            never the volatile ``ROWID`` (a rebuilt/merged database keeps the
            guid). Sources without native ids hash thread+sender+time+text.
        thread_id: Stable thread identity (``chat.guid`` for iMessage).
        thread_name: Human-facing thread label: the chat's ``display_name``
            when set, else the sorted participant names (group chats often
            have an empty ``display_name``).
        sender_handle: Raw handle (phone number / email); ``""`` for the
            owner (whose handle is not modelled — ego is structural).
        sender_name: Mapped display name (``GRAPH_HANDLE_NAMES``), falling
            back to the raw handle; the owner label when :attr:`is_from_me`.
        is_from_me: Whether the owner sent it (``message.is_from_me`` — the
            structural ego marker that Q3-class attribution needs).
        timestamp: Send time as a timezone-aware UTC datetime (both Apple
            epoch eras decoded to the same wall clock).
        text: The decoded message body; never empty (rows with no
            recoverable text are skipped at parse time, not emitted blank).
        tapbacks: Reactions folded onto this message (empty when none).
    """

    guid: str
    thread_id: str
    thread_name: str
    sender_handle: str
    sender_name: str
    is_from_me: bool
    timestamp: datetime
    text: str
    tapbacks: list[Tapback] = []

Tapback

Bases: BaseModel

One folded reaction on a message (a sentiment signal, not a message).

Tapback rows in chat.db never become standalone :class:SourceMessage records — they fold onto the message they react to (spec_graphrag §10.1).

Attributes:

Name	Type	Description
kind	str	The reaction, one of loved | liked | disliked | laughed | emphasized | questioned.
sender_name	str	Display name of whoever reacted (the owner label when the reaction is the owner's).

Source code in varagity/graph/sources/base.py

class Tapback(BaseModel):
    """One folded reaction on a message (a sentiment signal, not a message).

    Tapback rows in ``chat.db`` never become standalone
    :class:`SourceMessage` records — they fold onto the message they react to
    (spec_graphrag §10.1).

    Attributes:
        kind: The reaction, one of ``loved`` | ``liked`` | ``disliked`` |
            ``laughed`` | ``emphasized`` | ``questioned``.
        sender_name: Display name of whoever reacted (the owner label when
            the reaction is the owner's).
    """

    kind: str
    sender_name: str

batch_for_path(path, root, *, verbose=0)

Parse path into a :class:MessageBatch with a stable doc_id.

doc_id is derived from the path relative to root plus the file's byte hash — the same recipe as RAG documents (:func:varagity.stores.records.derive_doc_id) — so overlapping re-ingests of a growing database can upsert against a stable id. A path outside root (e.g. a symlink target) falls back to the file name, exactly as :func:varagity.ingest.loader._ingest_file does.

Parameters:

Name	Type	Description	Default
path	Path	File to parse (must be claimed by a registered source).	required
root	Path	Corpus root the doc_id is relativized against (call sites pass settings.GRAPH_DOCS_PATH).	required
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
MessageBatch	The parsed batch: identity plus its messages.

Raises:

Type	Description
ValueError	If no registered source matches path.

Source code in varagity/graph/sources/base.py

def batch_for_path(path: Path, root: Path, *, verbose: int = 0) -> MessageBatch:
    """Parse ``path`` into a :class:`MessageBatch` with a stable ``doc_id``.

    ``doc_id`` is derived from the path relative to ``root`` plus the file's
    byte hash — the same recipe as RAG documents
    (:func:`varagity.stores.records.derive_doc_id`) — so overlapping
    re-ingests of a growing database can upsert against a stable id. A
    ``path`` outside ``root`` (e.g. a symlink target) falls back to the file
    name, exactly as :func:`varagity.ingest.loader._ingest_file` does.

    Args:
        path: File to parse (must be claimed by a registered source).
        root: Corpus root the ``doc_id`` is relativized against (call sites
            pass ``settings.GRAPH_DOCS_PATH``).
        verbose: Validated console verbosity (0–2).

    Returns:
        The parsed batch: identity plus its messages.

    Raises:
        ValueError: If no registered source matches ``path``.
    """
    source = find_message_source(path)
    if source is None:
        raise ValueError(
            f"no message source matches {path} — registered: {list(MESSAGE_SOURCE_REGISTRY)}"
        )
    try:
        relative = path.resolve().relative_to(root.resolve()).as_posix()
    except ValueError:  # target outside the corpus root (e.g. a symlink)
        relative = path.name
    file_hash = content_hash(path.read_bytes())
    doc_id = derive_doc_id(relative, file_hash)
    messages = source.parse(path, verbose=verbose)
    return MessageBatch(doc_id=doc_id, relative_path=relative, messages=messages)

find_message_source(path)

Find the first registered source whose :meth:MessageSource.matches accepts path.

Structural dispatch (the _BUCKET_PARSERS precedent): a file with no matching source returns None rather than raising, so callers can count-and-skip an unsupported drop-in the way ingest discovery does.

Parameters:

Name	Type	Description	Default
path	Path	Candidate file.	required

Returns:

Type	Description
MessageSource | None	The matching source, or None when no registered source claims it.

Source code in varagity/graph/sources/base.py

def find_message_source(path: Path) -> MessageSource | None:
    """Find the first registered source whose :meth:`MessageSource.matches` accepts ``path``.

    Structural dispatch (the ``_BUCKET_PARSERS`` precedent): a file with no
    matching source returns ``None`` rather than raising, so callers can
    count-and-skip an unsupported drop-in the way ingest discovery does.

    Args:
        path: Candidate file.

    Returns:
        The matching source, or ``None`` when no registered source claims it.
    """
    for source in MESSAGE_SOURCE_REGISTRY.values():
        if source.matches(path):
            return source
    return None

get_message_source(name)

Look up a registered message source by name.

Parameters:

Name	Type	Description	Default
name	str	Registry key (e.g. "imessage").	required

Returns:

Type	Description
MessageSource	The registered source instance.

Raises:

Type	Description
KeyError	If no source is registered under name (the message lists the available ones).

Source code in varagity/graph/sources/base.py

def get_message_source(name: str) -> MessageSource:
    """Look up a registered message source by name.

    Args:
        name: Registry key (e.g. ``"imessage"``).

    Returns:
        The registered source instance.

    Raises:
        KeyError: If no source is registered under ``name`` (the message
            lists the available ones).
    """
    if name not in MESSAGE_SOURCE_REGISTRY:
        raise KeyError(
            f"Unknown message source {name!r}. Available: {list(MESSAGE_SOURCE_REGISTRY)}"
        )
    return MESSAGE_SOURCE_REGISTRY[name]

register(name)

Class decorator registering a message-source instance under name.

Parameters:

Name	Type	Description	Default
name	str	Registry key (also the source's stable identifier).	required

Returns:

Type	Description
Callable[[T], T]	The decorator, which instantiates and registers the class unchanged.

Source code in varagity/graph/sources/base.py

def register[T: type[Any]](name: str) -> Callable[[T], T]:
    """Class decorator registering a message-source instance under ``name``.

    Args:
        name: Registry key (also the source's stable identifier).

    Returns:
        The decorator, which instantiates and registers the class unchanged.
    """

    def deco(cls: T) -> T:
        MESSAGE_SOURCE_REGISTRY[name] = cls()
        return cls

    return deco

varagity.graph.sources.imessage

The iMessage chat.db message source (spec_graphrag §10.1).

Parses a copied iMessage SQLite store into :class:SourceMessage objects with full fidelity — the structure today's prose parsers throw away:

• Stable identity is message.guid (never the volatile ROWID), so an overlapping re-ingest of a growing database upserts rather than duplicates.
• Both Apple epoch eras are decoded per row by magnitude: message.date counts from 2001-01-01 UTC, in seconds pre-macOS 10.13/iOS 11 and nanoseconds after — a decade-spanning database contains both.
• attributedBody is decoded when the plain text column is NULL (the newer-OS default): the body lives in a serialized NSAttributedString typedstream blob, decoded here via pytypedstream.
• Tapbacks (associated_message_type 2000–2005 add / 3000–3005 remove) fold onto the message they react to; they never become standalone messages.
• Ego is structural: is_from_me marks the owner's messages, mapped to settings.graph_owner_label with an empty handle.
• Handles → names: participant handles (phone/email) are mapped through settings.graph_handle_name_map, falling back to the raw handle.

The database is the export: copy it with its -wal/-shm sidecars or recent messages are missing (SQLite honors the sidecars automatically when they sit beside the file). The parser opens the file read-only and never writes.

Running the module is the R3 verification tool (uv run python -m varagity.graph.sources.imessage /path/to/chat.db): it prints message/thread counts, the date range, tapback-fold count, and a few sample messages so a real export can be sanity-checked by hand.

IMessageSource

Parses an iMessage chat.db into normalized messages (spec_graphrag §10.1).

Source code in varagity/graph/sources/imessage.py

@register("imessage")
class IMessageSource:
    """Parses an iMessage ``chat.db`` into normalized messages (spec_graphrag §10.1)."""

    def matches(self, path: Path) -> bool:
        """Report whether ``path`` is an iMessage ``chat.db`` (a cheap probe).

        Three gates, cheapest first: a ``.db`` suffix, the 16-byte SQLite
        magic header, and a ``message`` table in ``sqlite_master``. Any read
        or SQLite error means "not my format" — the probe never raises.

        Args:
            path: Candidate file.

        Returns:
            ``True`` only if all three gates pass.
        """
        if path.suffix.lower() != ".db":
            return False
        try:
            with path.open("rb") as handle:
                magic = handle.read(len(_SQLITE_MAGIC))
        except OSError:
            return False
        if magic != _SQLITE_MAGIC:
            return False
        try:
            conn = _connect_ro(path)
            try:
                row = conn.execute(
                    "SELECT name FROM sqlite_master WHERE type='table' AND name='message'"
                ).fetchone()
            finally:
                conn.close()
        except sqlite3.Error:
            return False
        return row is not None

    def parse(self, path: Path, *, verbose: int = 0) -> list[SourceMessage]:
        """Parse the database into messages, oldest first (spec_graphrag §10.1).

        Reads handles, threads, and messages in three passes, folds tapbacks
        onto their targets, and skips rows with no recoverable body (counted
        in the summary log, never emitted blank). A message linked into more
        than one chat is emitted once (its first thread).

        Args:
            path: The ``chat.db`` to parse (already accepted by
                :meth:`matches`).
            verbose: Validated console verbosity (0–2); the parse summary is
                a ``logging`` line independent of it (the R3 tool prints the
                human-facing view).

        Returns:
            Every recoverable message, sorted by timestamp then guid.
        """
        settings = get_settings()
        owner_label = settings.graph_owner_label
        name_map = settings.graph_handle_name_map

        conn = _connect_ro(path)
        try:
            handle_by_rowid: dict[int, str] = {
                rowid: handle for rowid, handle in conn.execute("SELECT ROWID, id FROM handle")
            }

            def name_for_handle(handle_rowid: int | None) -> tuple[str, str]:
                """Map a handle ROWID to its ``(raw_handle, display_name)`` pair."""
                handle = handle_by_rowid.get(handle_rowid, "") if handle_rowid else ""
                return handle, name_map.get(handle, handle)

            chat_guid: dict[int, str] = {}
            chat_display: dict[int, str] = {}
            for chat_rowid, guid, display_name in conn.execute(
                "SELECT ROWID, guid, display_name FROM chat"
            ):
                chat_guid[chat_rowid] = guid
                chat_display[chat_rowid] = display_name or ""
            participants: dict[int, set[str]] = defaultdict(set)
            for chat_id, handle_id in conn.execute(
                "SELECT chat_id, handle_id FROM chat_handle_join"
            ):
                participants[chat_id].add(name_for_handle(handle_id)[1])
            # Group chats usually have an empty display_name — fall back to
            # the sorted participant names (spec_graphrag §10.1).
            thread_name_by_chat: dict[int, str] = {
                chat_rowid: (
                    display.strip()
                    or ", ".join(sorted(n for n in participants.get(chat_rowid, set()) if n))
                )
                for chat_rowid, display in chat_display.items()
            }

            rows = conn.execute(_MESSAGE_QUERY).fetchall()
        finally:
            conn.close()

        messages_by_guid: dict[str, SourceMessage] = {}
        tapback_net: dict[tuple[str, str, str], int] = defaultdict(int)
        seen_rowids: set[int] = set()
        skipped = 0

        for (
            rowid,
            guid,
            text,
            attributed_body,
            date,
            is_from_me,
            handle_id,
            assoc_type,
            assoc_guid,
            chat_id,
        ) in rows:
            if rowid in seen_rowids:  # a message linked into >1 chat — emit once
                continue
            seen_rowids.add(rowid)

            reaction = self._tapback_kind(assoc_type)
            if reaction is not None:
                kind, is_add = reaction
                target = self._strip_target_prefix(assoc_guid)
                if target is None:  # malformed association guid
                    continue
                reactor = owner_label if is_from_me else name_for_handle(handle_id)[1]
                tapback_net[(target, reactor, kind)] += 1 if is_add else -1
                continue

            body = (
                text if (text and text.strip()) else self._decode_attributed_body(attributed_body)
            )
            if not body:  # nothing recoverable — count and skip, never emit blank
                skipped += 1
                continue
            if is_from_me:
                sender_handle, sender_name = "", owner_label
            else:
                sender_handle, sender_name = name_for_handle(handle_id)
            messages_by_guid[guid] = SourceMessage(
                guid=guid,
                thread_id=chat_guid.get(chat_id, ""),
                thread_name=thread_name_by_chat.get(chat_id, ""),
                sender_handle=sender_handle,
                sender_name=sender_name,
                is_from_me=bool(is_from_me),
                timestamp=self._apple_timestamp(date),
                text=body,
            )

        folded = self._fold_tapbacks(tapback_net, messages_by_guid)
        messages = sorted(messages_by_guid.values(), key=lambda m: (m.timestamp, m.guid))
        logger.info(
            "parsed %s: %d message(s) decoded, %d skipped, %d tapback(s) folded",
            path.name,
            len(messages),
            skipped,
            folded,
        )
        return messages

    @staticmethod
    def _apple_timestamp(raw: int | None) -> datetime:
        """Decode a ``message.date`` into a tz-aware UTC datetime.

        Detects the epoch era per row by magnitude (:data:`_NS_THRESHOLD`);
        a ``0``/NULL date resolves to the epoch base rather than being
        dropped (dropping would silently lose rows, and goldens never depend
        on undated ones). Integer arithmetic keeps the decode exact.

        Args:
            raw: The raw ``message.date`` value (``None`` or ``0`` allowed).

        Returns:
            The send time as an aware UTC datetime.
        """
        if not raw:
            return _APPLE_EPOCH
        if raw > _NS_THRESHOLD:
            seconds, nanos = divmod(int(raw), _NANOS_PER_SECOND)
            return _APPLE_EPOCH + timedelta(seconds=seconds, microseconds=nanos // 1000)
        return _APPLE_EPOCH + timedelta(seconds=int(raw))

    @staticmethod
    def _tapback_kind(assoc_type: int | None) -> tuple[str, bool] | None:
        """Classify an ``associated_message_type`` as a tapback.

        Args:
            assoc_type: The row's ``associated_message_type`` (``0``/NULL for
                a normal message).

        Returns:
            A ``(kind, is_add)`` pair for a tapback add (2000–2005) or remove
            (3000–3005), or ``None`` when the row is not a tapback.
        """
        if assoc_type is None:
            return None
        span = len(_TAPBACK_KINDS)
        if _TAPBACK_ADD_BASE <= assoc_type < _TAPBACK_ADD_BASE + span:
            return _TAPBACK_KINDS[assoc_type - _TAPBACK_ADD_BASE], True
        if _TAPBACK_REMOVE_BASE <= assoc_type < _TAPBACK_REMOVE_BASE + span:
            return _TAPBACK_KINDS[assoc_type - _TAPBACK_REMOVE_BASE], False
        return None

    @staticmethod
    def _strip_target_prefix(assoc_guid: str | None) -> str | None:
        """Strip an ``associated_message_guid`` down to the bare target guid.

        iMessage prefixes the target with ``p:N/`` (message part N) or
        ``bp:`` (whole message); a bare guid is passed through defensively.

        Args:
            assoc_guid: The raw ``associated_message_guid``.

        Returns:
            The bare target guid, or ``None`` when the value is empty or a
            malformed ``p:`` prefix with no ``/``.
        """
        if not assoc_guid:
            return None
        if assoc_guid.startswith("bp:"):
            return assoc_guid[len("bp:") :]
        if assoc_guid.startswith("p:"):
            _, _, rest = assoc_guid.partition("/")
            return rest or None
        return assoc_guid

    @staticmethod
    def _fold_tapbacks(
        net: dict[tuple[str, str, str], int],
        messages_by_guid: dict[str, SourceMessage],
    ) -> int:
        """Fold net-positive reactions onto their target messages.

        A remove cancels a matching add per ``(target, reactor, kind)`` (the
        net count); a reaction whose target is absent from this parse is
        dropped silently (targets outside a partial export are expected).

        Args:
            net: Net add-minus-remove count per ``(target_guid, reactor,
                kind)``.
            messages_by_guid: The decoded messages, keyed by guid (mutated
                in place — reactions append to each target's ``tapbacks``).

        Returns:
            The number of reactions folded onto a present target.
        """
        folded = 0
        for (target_guid, reactor, kind), count in net.items():
            if count <= 0:
                continue
            target = messages_by_guid.get(target_guid)
            if target is None:  # reaction to a message outside this export
                continue
            target.tapbacks.append(Tapback(kind=kind, sender_name=reactor))
            folded += 1
        return folded

    @staticmethod
    def _decode_attributed_body(blob: bytes | None) -> str | None:
        """Decode a NULL-``text`` row's body from its ``attributedBody`` blob.

        Newer macOS/iOS store the body only in a serialized
        ``NSAttributedString`` typedstream (spec_graphrag §10.1). The archived
        string sits first in the object graph (attribute runs follow it), so
        a depth-first search for the first ``NSString`` value returns the
        message text. Any decode failure is a skip, not an abort (a bad blob
        must not fail the whole file); R3's real-export gate validates this
        path, with ``imessage-exporter`` as the documented contingency.

        Args:
            blob: The row's ``attributedBody`` bytes, or ``None``.

        Returns:
            The decoded body, or ``None`` when the blob is empty or
            undecodable.
        """
        if not blob:
            return None
        try:
            unarchived = typedstream.unarchive_from_data(blob)
        except Exception:  # any malformed blob skips the row (never aborts the file)
            logger.warning("attributedBody decode failed — skipping the row", exc_info=True)
            return None
        return _first_ns_string(unarchived)

matches(path)

Report whether path is an iMessage chat.db (a cheap probe).

Three gates, cheapest first: a .db suffix, the 16-byte SQLite magic header, and a message table in sqlite_master. Any read or SQLite error means "not my format" — the probe never raises.

Parameters:

Name	Type	Description	Default
path	Path	Candidate file.	required

Returns:

Type	Description
bool	True only if all three gates pass.

Source code in varagity/graph/sources/imessage.py

def matches(self, path: Path) -> bool:
    """Report whether ``path`` is an iMessage ``chat.db`` (a cheap probe).

    Three gates, cheapest first: a ``.db`` suffix, the 16-byte SQLite
    magic header, and a ``message`` table in ``sqlite_master``. Any read
    or SQLite error means "not my format" — the probe never raises.

    Args:
        path: Candidate file.

    Returns:
        ``True`` only if all three gates pass.
    """
    if path.suffix.lower() != ".db":
        return False
    try:
        with path.open("rb") as handle:
            magic = handle.read(len(_SQLITE_MAGIC))
    except OSError:
        return False
    if magic != _SQLITE_MAGIC:
        return False
    try:
        conn = _connect_ro(path)
        try:
            row = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='message'"
            ).fetchone()
        finally:
            conn.close()
    except sqlite3.Error:
        return False
    return row is not None

parse(path, *, verbose=0)

Parse the database into messages, oldest first (spec_graphrag §10.1).

Reads handles, threads, and messages in three passes, folds tapbacks onto their targets, and skips rows with no recoverable body (counted in the summary log, never emitted blank). A message linked into more than one chat is emitted once (its first thread).

Parameters:

Name	Type	Description	Default
path	Path	The chat.db to parse (already accepted by :meth:matches).	required
verbose	int	Validated console verbosity (0–2); the parse summary is a logging line independent of it (the R3 tool prints the human-facing view).	0

Returns:

Type	Description
list[SourceMessage]	Every recoverable message, sorted by timestamp then guid.

Source code in varagity/graph/sources/imessage.py

def parse(self, path: Path, *, verbose: int = 0) -> list[SourceMessage]:
    """Parse the database into messages, oldest first (spec_graphrag §10.1).

    Reads handles, threads, and messages in three passes, folds tapbacks
    onto their targets, and skips rows with no recoverable body (counted
    in the summary log, never emitted blank). A message linked into more
    than one chat is emitted once (its first thread).

    Args:
        path: The ``chat.db`` to parse (already accepted by
            :meth:`matches`).
        verbose: Validated console verbosity (0–2); the parse summary is
            a ``logging`` line independent of it (the R3 tool prints the
            human-facing view).

    Returns:
        Every recoverable message, sorted by timestamp then guid.
    """
    settings = get_settings()
    owner_label = settings.graph_owner_label
    name_map = settings.graph_handle_name_map

    conn = _connect_ro(path)
    try:
        handle_by_rowid: dict[int, str] = {
            rowid: handle for rowid, handle in conn.execute("SELECT ROWID, id FROM handle")
        }

        def name_for_handle(handle_rowid: int | None) -> tuple[str, str]:
            """Map a handle ROWID to its ``(raw_handle, display_name)`` pair."""
            handle = handle_by_rowid.get(handle_rowid, "") if handle_rowid else ""
            return handle, name_map.get(handle, handle)

        chat_guid: dict[int, str] = {}
        chat_display: dict[int, str] = {}
        for chat_rowid, guid, display_name in conn.execute(
            "SELECT ROWID, guid, display_name FROM chat"
        ):
            chat_guid[chat_rowid] = guid
            chat_display[chat_rowid] = display_name or ""
        participants: dict[int, set[str]] = defaultdict(set)
        for chat_id, handle_id in conn.execute(
            "SELECT chat_id, handle_id FROM chat_handle_join"
        ):
            participants[chat_id].add(name_for_handle(handle_id)[1])
        # Group chats usually have an empty display_name — fall back to
        # the sorted participant names (spec_graphrag §10.1).
        thread_name_by_chat: dict[int, str] = {
            chat_rowid: (
                display.strip()
                or ", ".join(sorted(n for n in participants.get(chat_rowid, set()) if n))
            )
            for chat_rowid, display in chat_display.items()
        }

        rows = conn.execute(_MESSAGE_QUERY).fetchall()
    finally:
        conn.close()

    messages_by_guid: dict[str, SourceMessage] = {}
    tapback_net: dict[tuple[str, str, str], int] = defaultdict(int)
    seen_rowids: set[int] = set()
    skipped = 0

    for (
        rowid,
        guid,
        text,
        attributed_body,
        date,
        is_from_me,
        handle_id,
        assoc_type,
        assoc_guid,
        chat_id,
    ) in rows:
        if rowid in seen_rowids:  # a message linked into >1 chat — emit once
            continue
        seen_rowids.add(rowid)

        reaction = self._tapback_kind(assoc_type)
        if reaction is not None:
            kind, is_add = reaction
            target = self._strip_target_prefix(assoc_guid)
            if target is None:  # malformed association guid
                continue
            reactor = owner_label if is_from_me else name_for_handle(handle_id)[1]
            tapback_net[(target, reactor, kind)] += 1 if is_add else -1
            continue

        body = (
            text if (text and text.strip()) else self._decode_attributed_body(attributed_body)
        )
        if not body:  # nothing recoverable — count and skip, never emit blank
            skipped += 1
            continue
        if is_from_me:
            sender_handle, sender_name = "", owner_label
        else:
            sender_handle, sender_name = name_for_handle(handle_id)
        messages_by_guid[guid] = SourceMessage(
            guid=guid,
            thread_id=chat_guid.get(chat_id, ""),
            thread_name=thread_name_by_chat.get(chat_id, ""),
            sender_handle=sender_handle,
            sender_name=sender_name,
            is_from_me=bool(is_from_me),
            timestamp=self._apple_timestamp(date),
            text=body,
        )

    folded = self._fold_tapbacks(tapback_net, messages_by_guid)
    messages = sorted(messages_by_guid.values(), key=lambda m: (m.timestamp, m.guid))
    logger.info(
        "parsed %s: %d message(s) decoded, %d skipped, %d tapback(s) folded",
        path.name,
        len(messages),
        skipped,
        folded,
    )
    return messages

Records, engine seam, rendering, and synthesis

varagity.graph.records

Engine-independent graph records (spec_graphrag §5.2, §10.2).

Every adapter normalizes its engine's native payloads into these models — which is what let the ADR-017 bake-off score three very different engines through one shape, and what keeps the seam benchmark-revisitable now that one of them won. They are pydantic (not dataclasses) for the same reason :class:~varagity.graph.sources.base.SourceMessage is: they go on the wire (graph evidence in the SSE stream, the graph-export endpoint) and into persisted eval results.

Two deliberate honesty rules run through the models:

• Optional counts mean "the engine cannot report this", not zero — :class:GraphStats fields are None when an engine exposes no way to ask (criterion §8.2#4 is measured honesty, not a hidden default).
• :attr:GraphEvidence.raw keeps the engine-native payload so the ADR autopsy can look at what an adapter couldn't map, rather than at a silently lossy projection.

BuildReport

Bases: BaseModel

What one :meth:~varagity.graph.base.GraphSession.build call did.

Attributes:

Name	Type	Description
messages_seen	int	Messages handed to the engine after guid-merging the batches (the upsert grain — a second build over an overlapping batch sees the union, not the sum).
wall_clock_s	float	Seconds the indexing took. The headline bake-off number (§8.2#8) and the input to Phase 5's scheduling.
failures	list[str]	Human-readable failures the adapter caught and continued past (a rejected document, an extraction call that never returned). Empty means a clean build; the count is criterion §8.2#2 data, so failures are collected rather than raised.

Source code in varagity/graph/records.py

class BuildReport(BaseModel):
    """What one :meth:`~varagity.graph.base.GraphSession.build` call did.

    Attributes:
        messages_seen: Messages handed to the engine after guid-merging the
            batches (the upsert grain — a second build over an overlapping
            batch sees the union, not the sum).
        wall_clock_s: Seconds the indexing took. The headline bake-off
            number (§8.2#8) and the input to Phase 5's scheduling.
        failures: Human-readable failures the adapter caught and continued
            past (a rejected document, an extraction call that never
            returned). Empty means a clean build; the count is criterion
            §8.2#2 data, so failures are collected rather than raised.
    """

    messages_seen: int
    wall_clock_s: float
    failures: list[str] = []

EngineFormat

Bases: BaseModel

Which engine wrote a graph's working directory, in which version of its files.

A graph backup is the workdir verbatim (ADR-019), so it is only as portable as the engine's on-disk format: a bundle records this beside its files, and a restore holds it to the running engine's. The engine reports its own format (:meth:~varagity.graph.base.GraphEngine.storage_format) — the registry is pluggable, so nothing outside an adapter names a library.

Attributes:

Name	Type	Description
name	str	The engine's registry name — what GRAPH_ENGINE selects, and the name of its working directory (lightrag).
package	str	The distribution whose code reads and writes the files (lightrag-hku).
version	str	That distribution's installed version (1.5.4).

Source code in varagity/graph/records.py

class EngineFormat(BaseModel):
    """Which engine wrote a graph's working directory, in which version of its files.

    A graph backup is the workdir verbatim (ADR-019), so it is only as
    portable as the engine's on-disk format: a bundle records this beside
    its files, and a restore holds it to the running engine's. The engine
    reports its own format (:meth:`~varagity.graph.base.GraphEngine.storage_format`)
    — the registry is pluggable, so nothing outside an adapter names a
    library.

    Attributes:
        name: The engine's registry name — what ``GRAPH_ENGINE`` selects, and
            the name of its working directory (``lightrag``).
        package: The distribution whose code reads and writes the files
            (``lightrag-hku``).
        version: That distribution's installed version (``1.5.4``).
    """

    name: str
    package: str
    version: str

GraphAnswer

Bases: BaseModel

One engine's answer to one question, with its evidence.

Attributes:

Name	Type	Description
answer	str	The generated answer text, <think>-stripped. On a +synthesis mode it is :mod:varagity.graph.answer's grounded answer over the retrieval (ADR-017's shipped design); on an unsuffixed mode it is the engine's own answer pipeline, kept so the bake-off's numbers stay reproducible.
evidence	GraphEvidence	What the engine retrieved to produce it.
mode	str	The engine query mode actually used (the --mode escape hatch records extra passes under their own mode name).
latency_s	float	Wall-clock seconds for the whole query, retrieval and generation together.

Source code in varagity/graph/records.py

class GraphAnswer(BaseModel):
    """One engine's answer to one question, with its evidence.

    Attributes:
        answer: The generated answer text, ``<think>``-stripped. On a
            ``+synthesis`` mode it is :mod:`varagity.graph.answer`'s
            grounded answer over the retrieval (ADR-017's shipped design);
            on an unsuffixed mode it is the engine's own answer pipeline,
            kept so the bake-off's numbers stay reproducible.
        evidence: What the engine retrieved to produce it.
        mode: The engine query mode actually used (the ``--mode`` escape
            hatch records extra passes under their own mode name).
        latency_s: Wall-clock seconds for the whole query, retrieval and
            generation together.
    """

    answer: str
    evidence: GraphEvidence
    mode: str
    latency_s: float

GraphCommunity

Bases: BaseModel

One community (cluster) summary the engine surfaced.

Only an engine with a community tier ever populates this, and the shipped one has none (LightRAG builds no community layer at all — R1, which is why the graph view clusters by entity_type instead). The model stays because the seam does: an engine that does summarize clusters slots in without a record change.

Attributes:

Name	Type	Description
id	str	The engine's community identifier.
title	str | None	Human-facing community label (None when unnamed).
summary	str	The community's summary text.

Source code in varagity/graph/records.py

class GraphCommunity(BaseModel):
    """One community (cluster) summary the engine surfaced.

    Only an engine with a community tier ever populates this, and the
    shipped one has none (LightRAG builds no community layer at all — R1,
    which is why the graph view clusters by ``entity_type`` instead). The
    model stays because the seam does: an engine that *does* summarize
    clusters slots in without a record change.

    Attributes:
        id: The engine's community identifier.
        title: Human-facing community label (``None`` when unnamed).
        summary: The community's summary text.
    """

    id: str
    title: str | None = None
    summary: str

GraphEntity

Bases: BaseModel

One node the engine surfaced as evidence for an answer.

Attributes:

Name	Type	Description
name	str	The entity's canonical name, as the engine resolved it.
type	str | None	The engine's entity type/category (None when the engine does not type its nodes).
summary	str | None	The engine's description of the entity (None when it surfaces bare names).

Source code in varagity/graph/records.py

class GraphEntity(BaseModel):
    """One node the engine surfaced as evidence for an answer.

    Attributes:
        name: The entity's canonical name, as the engine resolved it.
        type: The engine's entity type/category (``None`` when the engine
            does not type its nodes).
        summary: The engine's description of the entity (``None`` when it
            surfaces bare names).
    """

    name: str
    type: str | None = None
    summary: str | None = None

GraphEvidence

Bases: BaseModel

What the engine retrieved to support one answer (spec_graphrag §10.2).

Attributes:

Name	Type	Description
entities	list[GraphEntity]	Nodes cited by the retrieval.
relations	list[GraphRelation]	Edges cited by the retrieval.
communities	list[GraphCommunity]	Community summaries cited by the retrieval (empty for engines with no community layer).
message_guids	list[str]	Source-message provenance — [] when the engine surfaced nothing mappable back to a message. Empty is a result, not a failure: provenance recall is scored None for such engines rather than 0 (criterion §8.2#4).
raw	dict[str, Any] | None	The engine-native payload the mapping was derived from, kept verbatim for the ADR autopsy (None when the engine returned nothing structured).

Source code in varagity/graph/records.py

class GraphEvidence(BaseModel):
    """What the engine retrieved to support one answer (spec_graphrag §10.2).

    Attributes:
        entities: Nodes cited by the retrieval.
        relations: Edges cited by the retrieval.
        communities: Community summaries cited by the retrieval (empty for
            engines with no community layer).
        message_guids: Source-message provenance — ``[]`` when the engine
            surfaced nothing mappable back to a message. Empty is a
            *result*, not a failure: provenance recall is scored ``None``
            for such engines rather than 0 (criterion §8.2#4).
        raw: The engine-native payload the mapping was derived from, kept
            verbatim for the ADR autopsy (``None`` when the engine returned
            nothing structured).
    """

    entities: list[GraphEntity] = []
    relations: list[GraphRelation] = []
    communities: list[GraphCommunity] = []
    message_guids: list[str] = []
    raw: dict[str, Any] | None = None

GraphExport

Bases: BaseModel

A renderable slice of the whole graph (spec_graphrag §4.4).

Attributes:

Name	Type	Description
nodes	list[GraphExportNode]	The exported entities, engine order (degree-descending for a whole-graph export).
edges	list[GraphExportEdge]	Every edge between exported nodes.
truncated	bool	Whether the cap hid nodes the graph actually holds — the honesty flag the view must surface rather than silently pretending it drew everything.

Source code in varagity/graph/records.py

class GraphExport(BaseModel):
    """A renderable slice of the whole graph (spec_graphrag §4.4).

    Attributes:
        nodes: The exported entities, engine order (degree-descending for a
            whole-graph export).
        edges: Every edge between exported nodes.
        truncated: Whether the cap hid nodes the graph actually holds — the
            honesty flag the view must surface rather than silently
            pretending it drew everything.
    """

    nodes: list[GraphExportNode] = []
    edges: list[GraphExportEdge] = []
    truncated: bool = False

GraphExportEdge

Bases: BaseModel

One relation in a graph export, flattened for the wire.

Attributes:

Name	Type	Description
id	str	Stable edge identity within the export (source-target).
source	str	Name of the edge's source entity.
target	str	Name of the edge's target entity.
label	str | None	Short relation label/keywords (None when unlabelled).
description	str | None	The engine's longer description of the relation.

Source code in varagity/graph/records.py

class GraphExportEdge(BaseModel):
    """One relation in a graph export, flattened for the wire.

    Attributes:
        id: Stable edge identity within the export (``source-target``).
        source: Name of the edge's source entity.
        target: Name of the edge's target entity.
        label: Short relation label/keywords (``None`` when unlabelled).
        description: The engine's longer description of the relation.
    """

    id: str
    source: str
    target: str
    label: str | None = None
    description: str | None = None

GraphExportNode

Bases: BaseModel

One entity in a graph export, flattened for the wire.

Attributes:

Name	Type	Description
id	str	The entity's canonical name — LightRAG's node id is the name.
entity_type	str | None	The engine's type/category, which is what the view colors and clusters by (LightRAG builds no community tier, so type is the honest grouping — ADR-017).
description	str | None	The engine's merged description of the entity.
degree	int	How many of this export's edges touch the node. It is a subgraph degree, not a whole-graph one: an export is capped and degree-ordered, so the number describes the picture being drawn.
doc_keys	list[str]	The transcript documents the entity was extracted from — the drill-down join back to the corpus (a name in the graph view resolves to the days it was said on, resolved through :mod:varagity.graph.manifest). Empty when the engine does not record document provenance on its nodes. Deliberately not on the export wire schema: a whole-graph export would carry tens of keys per node for a picture that draws none of them, so only the entity-detail route projects it.

Source code in varagity/graph/records.py

class GraphExportNode(BaseModel):
    """One entity in a graph export, flattened for the wire.

    Attributes:
        id: The entity's canonical name — LightRAG's node id *is* the name.
        entity_type: The engine's type/category, which is what the view
            colors and clusters by (LightRAG builds no community tier, so
            type is the honest grouping — ADR-017).
        description: The engine's merged description of the entity.
        degree: How many of **this export's** edges touch the node. It is a
            subgraph degree, not a whole-graph one: an export is capped and
            degree-ordered, so the number describes the picture being drawn.
        doc_keys: The transcript documents the entity was extracted from —
            the drill-down join back to the corpus (a name in the graph view
            resolves to the days it was said on, resolved through
            :mod:`varagity.graph.manifest`). Empty when the engine does not
            record document provenance on its nodes. Deliberately *not* on
            the export wire schema: a whole-graph export would carry tens of
            keys per node for a picture that draws none of them, so only the
            entity-detail route projects it.
    """

    id: str
    entity_type: str | None = None
    description: str | None = None
    degree: int = 0
    doc_keys: list[str] = []

GraphRelation

Bases: BaseModel

One edge the engine surfaced as evidence for an answer.

Attributes:

Name	Type	Description
source	str	Name of the edge's source entity.
target	str	Name of the edge's target entity.
label	str | None	Short relation label/keywords (None when unlabelled).
description	str | None	The engine's longer description of the relation — the fact our synthesis grounds on (:func:varagity.graph.answer.facts_block).

Source code in varagity/graph/records.py

class GraphRelation(BaseModel):
    """One edge the engine surfaced as evidence for an answer.

    Attributes:
        source: Name of the edge's source entity.
        target: Name of the edge's target entity.
        label: Short relation label/keywords (``None`` when unlabelled).
        description: The engine's longer description of the relation — the
            *fact* our synthesis grounds on (:func:`varagity.graph.answer.facts_block`).
    """

    source: str
    target: str
    label: str | None = None
    description: str | None = None

GraphRetrieval

Bases: BaseModel

What one graph retrieval found, before any answer is written.

The unit ADR-017's retrieval-only design turns on: the engine finds entities, relations, and transcript passages, and this repo writes the answer over them — non-streaming for the harness (:func:varagity.graph.answer.synthesize), streamed for a chat turn (:func:varagity.pipeline.graph_flow.graph_query_stream_flow). Both paths therefore ground on the same payload, which is what keeps the eval graph harness a regression guard for the shipped query path.

Attributes:

Name	Type	Description
evidence	GraphEvidence	Entities and relations the retrieval surfaced (plus communities, for an engine that has them — LightRAG does not).
excerpts	list[TranscriptExcerpt]	The retrieved transcript passages, in the engine's own relevance order.
mode	str	The engine query mode that produced it, as the caller named it (a +synthesis suffix included — it records what ran).

Source code in varagity/graph/records.py

class GraphRetrieval(BaseModel):
    """What one graph retrieval found, before any answer is written.

    The unit ADR-017's retrieval-only design turns on: the engine finds
    entities, relations, and transcript passages, and *this repo* writes the
    answer over them — non-streaming for the harness
    (:func:`varagity.graph.answer.synthesize`), streamed for a chat turn
    (:func:`varagity.pipeline.graph_flow.graph_query_stream_flow`). Both
    paths therefore ground on the same payload, which is what keeps the
    ``eval graph`` harness a regression guard for the shipped query path.

    Attributes:
        evidence: Entities and relations the retrieval surfaced (plus
            communities, for an engine that has them — LightRAG does not).
        excerpts: The retrieved transcript passages, in the engine's own
            relevance order.
        mode: The engine query mode that produced it, as the caller named it
            (a ``+synthesis`` suffix included — it records what ran).
    """

    evidence: GraphEvidence
    excerpts: list[TranscriptExcerpt] = []
    mode: str

GraphStats

Bases: BaseModel

Size of the graph an engine holds, as far as it will say.

Every field is optional because "the engine exposes no count API" is a real and reportable state — the incremental-reindex check reads these before and after a delta build, and a None there is honest data for the ADR rather than a zero that reads like an empty graph.

Attributes:

Name	Type	Description
entities	int | None	Entity/node count, or None if unavailable.
relations	int | None	Relation/edge count, or None if unavailable.
communities	int | None	Community count, or None if unavailable — which includes engines with no community layer at all.

Source code in varagity/graph/records.py

class GraphStats(BaseModel):
    """Size of the graph an engine holds, as far as it will say.

    Every field is optional because "the engine exposes no count API" is a
    real and reportable state — the incremental-reindex check reads these
    before and after a delta build, and a ``None`` there is honest data for
    the ADR rather than a zero that reads like an empty graph.

    Attributes:
        entities: Entity/node count, or ``None`` if unavailable.
        relations: Relation/edge count, or ``None`` if unavailable.
        communities: Community count, or ``None`` if unavailable — which
            includes engines with no community layer at all.
    """

    entities: int | None = None
    relations: int | None = None
    communities: int | None = None

TranscriptExcerpt

Bases: BaseModel

One retrieved transcript passage, normalized across engines.

The document-grain unit of graph evidence: a document-shaped engine retrieves text alongside its entities and relations, and that text is what a grounded answer is actually written from (plan decision #6 — the shipped query path grounds on the full retrieval payload, so the gate measures the same diet).

Grain is the transcript day-span, not the message: that is what the engines can honestly attribute (ADR-017's priced regret), which is why :attr:message_guids is a span of guids rather than one.

Attributes:

Name	Type	Description
doc_key	str	The :attr:~varagity.graph.render.TranscriptDoc.doc_key the passage came from — the join key back to the corpus.
thread_name	str	Human-facing thread label, for the answer's citation — resolved from the workdir manifest when it knows the key, so a chunk-grain hit carries the same label as a doc-grain hit of the same document (falls back to the passage's transcript header, then to the thread id).
span	str	The document's day span (YYYY-MM-DD or first..last), parsed from the key.
text	str	The retrieved passage, verbatim.
message_guids	list[str]	Per-message provenance for the passage's document, or [] when the guid index does not know the key (a --skip-build re-score, until stage 2's workdir manifest makes the index durable).

Source code in varagity/graph/records.py

class TranscriptExcerpt(BaseModel):
    """One retrieved transcript passage, normalized across engines.

    The document-grain unit of graph evidence: a document-shaped engine
    retrieves *text* alongside its entities and relations, and that text is
    what a grounded answer is actually written from (plan decision #6 — the
    shipped query path grounds on the full retrieval payload, so the gate
    measures the same diet).

    Grain is the transcript day-span, not the message: that is what the
    engines can honestly attribute (ADR-017's priced regret), which is why
    :attr:`message_guids` is a *span* of guids rather than one.

    Attributes:
        doc_key: The :attr:`~varagity.graph.render.TranscriptDoc.doc_key`
            the passage came from — the join key back to the corpus.
        thread_name: Human-facing thread label, for the answer's citation —
            resolved from the workdir manifest when it knows the key, so a
            chunk-grain hit carries the same label as a doc-grain hit of the
            same document (falls back to the passage's transcript header,
            then to the thread id).
        span: The document's day span (``YYYY-MM-DD`` or ``first..last``),
            parsed from the key.
        text: The retrieved passage, verbatim.
        message_guids: Per-message provenance for the passage's document, or
            ``[]`` when the guid index does not know the key (a
            ``--skip-build`` re-score, until stage 2's workdir manifest makes
            the index durable).
    """

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

varagity.graph.base

Graph-engine protocol and registry (spec_graphrag §5.2, §8; the spec §5.1 pattern).

Each engine module defines one adapter decorated with @register("name"); callers resolve one by name from :attr:~varagity.config.Settings.GRAPH_ENGINE via :func:get_graph_engine. ADR-017 chose lightrag and the losing bake-off adapters were deleted at stage-2 start; the seam is what stays, so the decision remains benchmark-revisitable exactly as the retriever and chat-engine registries are.

The family is split in two on purpose:

• :class:GraphEngine is the registered singleton — stateless, zero-arg constructible, and cheap to import (adapters import their heavy engine libs inside :meth:GraphEngine.session, stage-1 decision #8).
• :class:GraphSession holds every piece of per-run state — the engine handle, its working directory, its event loop, the transcript→guid index — behind a context manager, because teardown is not optional.

:meth:GraphSession.build takes a sequence of batches and upserts at message grain: adapters guid-merge across source files (:func:varagity.graph.render.merge_batches) before rendering, so an overlapping chat.db upload or a re-export of a grown database never duplicates a message. A second build call with an overlapping batch must not duplicate either, and — the harder half — a changed document must not be silently kept: the shipped adapter diffs rendered content hashes against :mod:varagity.graph.manifest and deletes before re-inserting.

The rest of the protocol is what an application needs beyond a benchmark (stage-2 decision #11): :meth:GraphSession.resume to finish a killed build, :meth:GraphSession.retrieve to feed a streamed answer the app writes itself, :meth:GraphSession.export to draw the graph, :meth:GraphSession.delete_documents to retract sources, and :meth:GraphSession.document_statuses to report progress. Sessions are single-writer per working directory by engine invariant, which is why exactly one process (the API) ever opens one.

GraphEngine

Bases: Protocol

Opens working sessions against a graph engine (spec_graphrag §8).

It also names the format its working-directory files are in (:meth:storage_format), which is what lets a graph backup refuse a bundle another engine version wrote rather than hand it the wrong files.

runtime_checkable for the same reason :class:~varagity.chat.base.ChatEngine is: the protocol will appear in stage-2 Prefect flow signatures, and Prefect builds a pydantic parameter schema from the annotations at decoration time, which requires types usable with isinstance.

Source code in varagity/graph/base.py

@runtime_checkable
class GraphEngine(Protocol):
    """Opens working sessions against a graph engine (spec_graphrag §8).

    It also names the format its working-directory files are in
    (:meth:`storage_format`), which is what lets a graph backup refuse a
    bundle another engine version wrote rather than hand it the wrong files.

    ``runtime_checkable`` for the same reason
    :class:`~varagity.chat.base.ChatEngine` is: the protocol will appear in
    stage-2 Prefect flow signatures, and Prefect builds a pydantic parameter
    schema from the annotations at decoration time, which requires types
    usable with ``isinstance``.
    """

    def session(self, workdir: Path) -> AbstractContextManager[GraphSession]:
        """Open a session whose state lives under ``workdir``.

        Engines self-store (plan decision #9): every byte an engine writes —
        graph, vectors, KV, caches — belongs under ``workdir``, so wiping it
        is a clean slate and no compose service is involved.

        Args:
            workdir: The engine's working directory (created if absent).

        Returns:
            A context manager yielding the session; exiting it tears down
            whatever the engine started (subprocesses, event loops, stores).
        """
        ...

    def storage_format(self) -> EngineFormat:
        """Name the format this engine's working-directory files are in.

        What a graph backup records beside the workdir it copies (ADR-019),
        and what a restore holds a bundle to — the files are only as portable
        as the engine version that wrote them. Must stay **import-light**,
        like the rest of the registered surface: read the installed version
        from package metadata, never by importing the engine library.

        Returns:
            The engine's registry name, its distribution, and that
            distribution's installed version.
        """
        ...

session(workdir)

Open a session whose state lives under workdir.

Engines self-store (plan decision #9): every byte an engine writes — graph, vectors, KV, caches — belongs under workdir, so wiping it is a clean slate and no compose service is involved.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory (created if absent).	required

Returns:

Type	Description
AbstractContextManager[GraphSession]	A context manager yielding the session; exiting it tears down
AbstractContextManager[GraphSession]	whatever the engine started (subprocesses, event loops, stores).

Source code in varagity/graph/base.py

def session(self, workdir: Path) -> AbstractContextManager[GraphSession]:
    """Open a session whose state lives under ``workdir``.

    Engines self-store (plan decision #9): every byte an engine writes —
    graph, vectors, KV, caches — belongs under ``workdir``, so wiping it
    is a clean slate and no compose service is involved.

    Args:
        workdir: The engine's working directory (created if absent).

    Returns:
        A context manager yielding the session; exiting it tears down
        whatever the engine started (subprocesses, event loops, stores).
    """
    ...

storage_format()

Name the format this engine's working-directory files are in.

What a graph backup records beside the workdir it copies (ADR-019), and what a restore holds a bundle to — the files are only as portable as the engine version that wrote them. Must stay import-light, like the rest of the registered surface: read the installed version from package metadata, never by importing the engine library.

Returns:

Type	Description
EngineFormat	The engine's registry name, its distribution, and that
EngineFormat	distribution's installed version.

Source code in varagity/graph/base.py

def storage_format(self) -> EngineFormat:
    """Name the format this engine's working-directory files are in.

    What a graph backup records beside the workdir it copies (ADR-019),
    and what a restore holds a bundle to — the files are only as portable
    as the engine version that wrote them. Must stay **import-light**,
    like the rest of the registered surface: read the installed version
    from package metadata, never by importing the engine library.

    Returns:
        The engine's registry name, its distribution, and that
        distribution's installed version.
    """
    ...

GraphSession

Bases: Protocol

One engine's open working session over one working directory.

Sessions are handed out by :meth:GraphEngine.session and are valid only inside its with block.

Source code in varagity/graph/base.py

class GraphSession(Protocol):
    """One engine's open working session over one working directory.

    Sessions are handed out by :meth:`GraphEngine.session` and are valid only
    inside its ``with`` block.
    """

    def build(
        self,
        batches: Sequence[MessageBatch],
        *,
        verbose: int = 0,
        prune_removed: bool = True,
    ) -> BuildReport:
        """Index (or re-index) messages into the graph — an upsert, not an append.

        Re-callable by design: a build killed halfway through resumes by
        being called again, and one over an unchanged corpus costs no
        extraction at all.

        Args:
            batches: Parsed source files. Messages are guid-merged across
                them before rendering, so overlapping exports of the same
                conversation collapse to one message stream.
            verbose: Validated console verbosity (0–2).
            prune_removed: Whether ``batches`` render the **whole** corpus,
                making a document the graph holds but the render omits a
                deleted source. A bounded build (message cap, date floor)
                passes ``False``: its render is partial on purpose, so only
                removed keys whose every recorded message the render
                re-indexes under new spans (re-span casualties of the greedy
                day-span packing) may be pruned on its behalf.

        Returns:
            What the build did: messages seen, wall clock, and any failures
            the adapter caught and continued past.
        """
        ...

    def resume(self, *, verbose: int = 0) -> BuildReport:
        """Finish whatever a killed build left in flight, without re-rendering.

        Args:
            verbose: Validated console verbosity (0–2).

        Returns:
            What the pass did. ``messages_seen`` describes the corpus already
            indexed, not a fresh hand-off.
        """
        ...

    def retrieve(
        self, question: str, *, mode: str | None = None, verbose: int = 0
    ) -> GraphRetrieval:
        """Find the evidence for one question, writing no answer at all.

        The shipped query path's first half (ADR-017's retrieval-only
        decision): the app streams its own grounded answer over what this
        returns, so an engine must be able to retrieve *without* spending its
        own answer call. :meth:`query` is this composed with an answer stage,
        which is what keeps the harness measuring the shipped diet.

        Args:
            question: The search query, verbatim — already condensed by the
                chat engine when the turn needed it (the answer prompt still
                gets the user's own words; spec_v3 §4.2).
            mode: Engine query mode; ``None`` uses the adapter's primary.
            verbose: Validated console verbosity (0–2).

        Returns:
            The evidence and transcript passages, empty when the engine
            could not retrieve (a graph turn degrades to "no facts", it does
            not raise).
        """
        ...

    def query(self, question: str, *, mode: str | None = None, verbose: int = 0) -> GraphAnswer:
        """Answer one question from the graph.

        Args:
            question: The user's question, verbatim (golden ``kind`` metadata
                is never shown to an engine — stage-1 decision #10).
            mode: Engine query mode; ``None`` uses the adapter's primary mode
                (stage-1 decision #13 — the ``--mode`` escape hatch records
                extra passes).
            verbose: Validated console verbosity (0–2).

        Returns:
            The answer with its graph evidence and the mode actually used.
        """
        ...

    def stats(self) -> GraphStats:
        """Report the current graph size, as far as the engine will say.

        Returns:
            Entity/relation/community counts, each ``None`` where the engine
            exposes no way to ask (the incremental-reindex check reads this
            before and after a delta build).
        """
        ...

    def document_statuses(self) -> dict[str, int]:
        """Count the engine's documents by processing status.

        Returns:
            Status name → document count, in the engine's own vocabulary, or
            an empty mapping when it would not say (which a progress reader
            must treat as "no news", never as "zero documents").
        """
        ...

    def export(
        self,
        label: str = "*",
        *,
        max_depth: int = 3,
        max_nodes: int = 5000,
    ) -> GraphExport:
        """Read a renderable slice of the graph.

        Args:
            label: Entity name to centre the slice on; ``"*"`` takes the whole
                graph, ordered so the cap keeps the most connected nodes.
            max_depth: Hops to walk out from ``label``.
            max_nodes: Node cap; the engine may clamp it further and says so
                through :attr:`~varagity.graph.records.GraphExport.truncated`.

        Returns:
            The slice, empty when the graph cannot be read.
        """
        ...

    def delete_documents(self, doc_keys: Sequence[str]) -> int:
        """Remove documents, and whatever the graph derived from them.

        Args:
            doc_keys: Transcript document keys to retract.

        Returns:
            How many deletions the engine accepted.
        """
        ...

build(batches, *, verbose=0, prune_removed=True)

Index (or re-index) messages into the graph — an upsert, not an append.

Re-callable by design: a build killed halfway through resumes by being called again, and one over an unchanged corpus costs no extraction at all.

Parameters:

Name	Type	Description	Default
batches	Sequence[MessageBatch]	Parsed source files. Messages are guid-merged across them before rendering, so overlapping exports of the same conversation collapse to one message stream.	required
verbose	int	Validated console verbosity (0–2).	0
prune_removed	bool	Whether batches render the whole corpus, making a document the graph holds but the render omits a deleted source. A bounded build (message cap, date floor) passes False: its render is partial on purpose, so only removed keys whose every recorded message the render re-indexes under new spans (re-span casualties of the greedy day-span packing) may be pruned on its behalf.	True

Returns:

Type	Description
BuildReport	What the build did: messages seen, wall clock, and any failures
BuildReport	the adapter caught and continued past.

Source code in varagity/graph/base.py

def build(
    self,
    batches: Sequence[MessageBatch],
    *,
    verbose: int = 0,
    prune_removed: bool = True,
) -> BuildReport:
    """Index (or re-index) messages into the graph — an upsert, not an append.

    Re-callable by design: a build killed halfway through resumes by
    being called again, and one over an unchanged corpus costs no
    extraction at all.

    Args:
        batches: Parsed source files. Messages are guid-merged across
            them before rendering, so overlapping exports of the same
            conversation collapse to one message stream.
        verbose: Validated console verbosity (0–2).
        prune_removed: Whether ``batches`` render the **whole** corpus,
            making a document the graph holds but the render omits a
            deleted source. A bounded build (message cap, date floor)
            passes ``False``: its render is partial on purpose, so only
            removed keys whose every recorded message the render
            re-indexes under new spans (re-span casualties of the greedy
            day-span packing) may be pruned on its behalf.

    Returns:
        What the build did: messages seen, wall clock, and any failures
        the adapter caught and continued past.
    """
    ...

delete_documents(doc_keys)

Remove documents, and whatever the graph derived from them.

Parameters:

Name	Type	Description	Default
doc_keys	Sequence[str]	Transcript document keys to retract.	required

Returns:

Type	Description
int	How many deletions the engine accepted.

Source code in varagity/graph/base.py

def delete_documents(self, doc_keys: Sequence[str]) -> int:
    """Remove documents, and whatever the graph derived from them.

    Args:
        doc_keys: Transcript document keys to retract.

    Returns:
        How many deletions the engine accepted.
    """
    ...

document_statuses()

Count the engine's documents by processing status.

Returns:

Type	Description
dict[str, int]	Status name → document count, in the engine's own vocabulary, or
dict[str, int]	an empty mapping when it would not say (which a progress reader
dict[str, int]	must treat as "no news", never as "zero documents").

Source code in varagity/graph/base.py

def document_statuses(self) -> dict[str, int]:
    """Count the engine's documents by processing status.

    Returns:
        Status name → document count, in the engine's own vocabulary, or
        an empty mapping when it would not say (which a progress reader
        must treat as "no news", never as "zero documents").
    """
    ...

export(label='*', *, max_depth=3, max_nodes=5000)

Read a renderable slice of the graph.

Parameters:

Name	Type	Description	Default
label	str	Entity name to centre the slice on; "*" takes the whole graph, ordered so the cap keeps the most connected nodes.	'*'
max_depth	int	Hops to walk out from label.	3
max_nodes	int	Node cap; the engine may clamp it further and says so through :attr:~varagity.graph.records.GraphExport.truncated.	5000

Returns:

Type	Description
GraphExport	The slice, empty when the graph cannot be read.

Source code in varagity/graph/base.py

def export(
    self,
    label: str = "*",
    *,
    max_depth: int = 3,
    max_nodes: int = 5000,
) -> GraphExport:
    """Read a renderable slice of the graph.

    Args:
        label: Entity name to centre the slice on; ``"*"`` takes the whole
            graph, ordered so the cap keeps the most connected nodes.
        max_depth: Hops to walk out from ``label``.
        max_nodes: Node cap; the engine may clamp it further and says so
            through :attr:`~varagity.graph.records.GraphExport.truncated`.

    Returns:
        The slice, empty when the graph cannot be read.
    """
    ...

query(question, *, mode=None, verbose=0)

Answer one question from the graph.

Parameters:

Name	Type	Description	Default
question	str	The user's question, verbatim (golden kind metadata is never shown to an engine — stage-1 decision #10).	required
mode	str | None	Engine query mode; None uses the adapter's primary mode (stage-1 decision #13 — the --mode escape hatch records extra passes).	None
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
GraphAnswer	The answer with its graph evidence and the mode actually used.

Source code in varagity/graph/base.py

def query(self, question: str, *, mode: str | None = None, verbose: int = 0) -> GraphAnswer:
    """Answer one question from the graph.

    Args:
        question: The user's question, verbatim (golden ``kind`` metadata
            is never shown to an engine — stage-1 decision #10).
        mode: Engine query mode; ``None`` uses the adapter's primary mode
            (stage-1 decision #13 — the ``--mode`` escape hatch records
            extra passes).
        verbose: Validated console verbosity (0–2).

    Returns:
        The answer with its graph evidence and the mode actually used.
    """
    ...

resume(*, verbose=0)

Finish whatever a killed build left in flight, without re-rendering.

Parameters:

Name	Type	Description	Default
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
BuildReport	What the pass did. messages_seen describes the corpus already
BuildReport	indexed, not a fresh hand-off.

Source code in varagity/graph/base.py

def resume(self, *, verbose: int = 0) -> BuildReport:
    """Finish whatever a killed build left in flight, without re-rendering.

    Args:
        verbose: Validated console verbosity (0–2).

    Returns:
        What the pass did. ``messages_seen`` describes the corpus already
        indexed, not a fresh hand-off.
    """
    ...

retrieve(question, *, mode=None, verbose=0)

Find the evidence for one question, writing no answer at all.

The shipped query path's first half (ADR-017's retrieval-only decision): the app streams its own grounded answer over what this returns, so an engine must be able to retrieve without spending its own answer call. :meth:query is this composed with an answer stage, which is what keeps the harness measuring the shipped diet.

Parameters:

Name	Type	Description	Default
question	str	The search query, verbatim — already condensed by the chat engine when the turn needed it (the answer prompt still gets the user's own words; spec_v3 §4.2).	required
mode	str | None	Engine query mode; None uses the adapter's primary.	None
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
GraphRetrieval	The evidence and transcript passages, empty when the engine
GraphRetrieval	could not retrieve (a graph turn degrades to "no facts", it does
GraphRetrieval	not raise).

Source code in varagity/graph/base.py

def retrieve(
    self, question: str, *, mode: str | None = None, verbose: int = 0
) -> GraphRetrieval:
    """Find the evidence for one question, writing no answer at all.

    The shipped query path's first half (ADR-017's retrieval-only
    decision): the app streams its own grounded answer over what this
    returns, so an engine must be able to retrieve *without* spending its
    own answer call. :meth:`query` is this composed with an answer stage,
    which is what keeps the harness measuring the shipped diet.

    Args:
        question: The search query, verbatim — already condensed by the
            chat engine when the turn needed it (the answer prompt still
            gets the user's own words; spec_v3 §4.2).
        mode: Engine query mode; ``None`` uses the adapter's primary.
        verbose: Validated console verbosity (0–2).

    Returns:
        The evidence and transcript passages, empty when the engine
        could not retrieve (a graph turn degrades to "no facts", it does
        not raise).
    """
    ...

stats()

Report the current graph size, as far as the engine will say.

Returns:

Type	Description
GraphStats	Entity/relation/community counts, each None where the engine
GraphStats	exposes no way to ask (the incremental-reindex check reads this
GraphStats	before and after a delta build).

Source code in varagity/graph/base.py

def stats(self) -> GraphStats:
    """Report the current graph size, as far as the engine will say.

    Returns:
        Entity/relation/community counts, each ``None`` where the engine
        exposes no way to ask (the incremental-reindex check reads this
        before and after a delta build).
    """
    ...

get_graph_engine(name)

Look up a registered graph engine by name.

Parameters:

Name	Type	Description	Default
name	str	Registry key (e.g. "lightrag").	required

Returns:

Type	Description
GraphEngine	The registered engine instance.

Raises:

Type	Description
KeyError	If no engine is registered under name (the message lists the available ones).

Source code in varagity/graph/base.py

def get_graph_engine(name: str) -> GraphEngine:
    """Look up a registered graph engine by name.

    Args:
        name: Registry key (e.g. ``"lightrag"``).

    Returns:
        The registered engine instance.

    Raises:
        KeyError: If no engine is registered under ``name`` (the message
            lists the available ones).
    """
    if name not in GRAPH_ENGINE_REGISTRY:
        raise KeyError(f"Unknown graph engine {name!r}. Available: {list(GRAPH_ENGINE_REGISTRY)}")
    return GRAPH_ENGINE_REGISTRY[name]

register(name)

Class decorator registering a graph-engine instance under name.

Parameters:

Name	Type	Description	Default
name	str	Registry key (the --engine value the harness takes).	required

Returns:

Type	Description
Callable[[T], T]	The decorator, which instantiates and registers the class unchanged.

Source code in varagity/graph/base.py

def register[T: type[Any]](name: str) -> Callable[[T], T]:
    """Class decorator registering a graph-engine instance under ``name``.

    Args:
        name: Registry key (the ``--engine`` value the harness takes).

    Returns:
        The decorator, which instantiates and registers the class unchanged.
    """

    def deco(cls: T) -> T:
        GRAPH_ENGINE_REGISTRY[name] = cls()
        return cls

    return deco

varagity.graph.render

Rendering messages into what the graph engine eats (spec_graphrag §5.2, §10.2).

Rendering lives here rather than in the adapter — pure functions over :class:~varagity.graph.sources.base.SourceMessage are where most adapter correctness actually is, and they are testable without the engine installed. The shipped engine is document-shaped (ADR-017), so messages become thread transcripts split on day boundaries (:func:thread_transcripts). The episode-shaped rendering the bake-off's Graphiti seat ate went out with that adapter (stage-2 loser removal); git history has it if an episode-shaped engine ever re-enters the seam.

Upstream of it is :func:merge_batches, the upsert seam: source files overlap (a re-export of a grown chat.db is a superset of the older one), so messages are guid-deduped across batches before anything is rendered. Rendering is then a pure function of the merged messages: the same thread-days always produce the same :attr:TranscriptDoc.doc_key and the same text, so an unchanged conversation stays upsert-identical no matter how many times it is re-uploaded — which is exactly what :mod:varagity.graph.manifest's content hashes turn into a build diff.

Purity is why nothing here reads :func:~varagity.config.get_settings: the owner's display label rides in on :attr:~varagity.graph.sources.base.SourceMessage.sender_name, which the parser already resolved from GRAPH_OWNER_ALIASES. Re-deriving it at render time would let a settings change silently rewrite documents that the engines have already indexed under the same key.

TranscriptDoc

Bases: BaseModel

One rendered thread transcript, as the document-shaped engines see it.

Attributes:

Name	Type	Description
doc_key	str	Stable document identity, {thread_id}::{day-span} (one day, or first..last when a document packs several). It is deliberately not derived from a doc_id: the same thread-days rendered from an overlapping upload or a re-export of a grown database must land on the same key, and a file's doc_id changes with every byte added to it.
thread_id	str	The conversation this document came from.
thread_name	str	Human-facing thread label (the header's subject).
text	str	The rendered transcript — a Thread: header line followed by one [YYYY-MM-DD HH:MM] Sender: text line per message, with folded reactions indented beneath the message they land on.
message_guids	list[str]	Every message in the document, in transcript order — the map back from an engine's document-grain citation to per-message provenance.

Source code in varagity/graph/render.py

class TranscriptDoc(BaseModel):
    """One rendered thread transcript, as the document-shaped engines see it.

    Attributes:
        doc_key: Stable document identity, ``{thread_id}::{day-span}`` (one
            day, or ``first..last`` when a document packs several). It is
            deliberately **not** derived from a ``doc_id``: the same
            thread-days rendered from an overlapping upload or a re-export
            of a grown database must land on the same key, and a file's
            ``doc_id`` changes with every byte added to it.
        thread_id: The conversation this document came from.
        thread_name: Human-facing thread label (the header's subject).
        text: The rendered transcript — a ``Thread:`` header line followed by
            one ``[YYYY-MM-DD HH:MM] Sender: text`` line per message, with
            folded reactions indented beneath the message they land on.
        message_guids: Every message in the document, in transcript order —
            the map back from an engine's document-grain citation to
            per-message provenance.
    """

    doc_key: str
    thread_id: str
    thread_name: str
    text: str
    message_guids: list[str]

doc_guid_index(docs)

Index transcript documents by key, for mapping citations back to messages.

Parameters:

Name	Type	Description	Default
docs	Sequence[TranscriptDoc]	Rendered transcripts.	required

Returns:

Type	Description
dict[str, list[str]]	doc_key → its message guids, in transcript order.

Source code in varagity/graph/render.py

def doc_guid_index(docs: Sequence[TranscriptDoc]) -> dict[str, list[str]]:
    """Index transcript documents by key, for mapping citations back to messages.

    Args:
        docs: Rendered transcripts.

    Returns:
        ``doc_key`` → its message guids, in transcript order.
    """
    return {doc.doc_key: list(doc.message_guids) for doc in docs}

guids_in_payload(payload, index)

Recover message provenance from an engine-native payload.

Every adapter hands its engine document keys as the document ids/file names it indexes under, so an engine that cites its sources cites those keys — somewhere inside a shape this code deliberately does not model. The walk is therefore structural rather than schema-aware: every string in the payload (dict keys included, pydantic models dumped) is checked against the index, exactly first and as a substring second, so a bare key, a /path/to/key.txt, and a key<SEP>key join all resolve.

An engine that cites nothing simply yields [] — which is reported as "no provenance", never as a zero score (criterion §8.2#4).

Parameters:

Name	Type	Description	Default
payload	Any	Anything the engine returned (mapping, sequence, model, or string).	required
index	Mapping[str, Sequence[str]]	doc_key → message guids, from :func:doc_guid_index plus whatever aliases the adapter indexed its documents under.	required

Returns:

Type	Description
list[str]	The cited messages' guids, deduplicated, in the order the keys were
list[str]	encountered.

Source code in varagity/graph/render.py

def guids_in_payload(payload: Any, index: Mapping[str, Sequence[str]]) -> list[str]:
    """Recover message provenance from an engine-native payload.

    Every adapter hands its engine document keys as the document ids/file
    names it indexes under, so an engine that cites its sources cites those
    keys — somewhere inside a shape this code deliberately does not model.
    The walk is therefore structural rather than schema-aware: every string
    in the payload (dict keys included, pydantic models dumped) is checked
    against the index, exactly first and as a substring second, so a bare key,
    a ``/path/to/key.txt``, and a ``key<SEP>key`` join all resolve.

    An engine that cites nothing simply yields ``[]`` — which is reported as
    "no provenance", never as a zero score (criterion §8.2#4).

    Args:
        payload: Anything the engine returned (mapping, sequence, model, or
            string).
        index: ``doc_key`` → message guids, from :func:`doc_guid_index` plus
            whatever aliases the adapter indexed its documents under.

    Returns:
        The cited messages' guids, deduplicated, in the order the keys were
        encountered.
    """
    keys = list(index)
    matched: list[str] = []
    seen_keys: set[str] = set()
    for text in _iter_strings(payload):
        for key in _keys_in_text(text, keys):
            if key not in seen_keys:
                seen_keys.add(key)
                matched.append(key)
    guids: list[str] = []
    seen_guids: set[str] = set()
    for key in matched:
        for guid in index[key]:
            if guid not in seen_guids:
                seen_guids.add(guid)
                guids.append(guid)
    return guids

merge_batches(batches)

Guid-merge parsed source files into one ordered message stream.

The upsert seam (plan amendment, owner-approved 2026-07-25): overlapping chat.db uploads and re-exports of a growing database share message guids, so the union — not the concatenation — is what any engine should index. First occurrence wins: batches are processed in the order given, and a later batch's copy of an already-seen guid is dropped rather than overwriting (a re-export's rows are the same rows; preferring the first keeps a build reproducible regardless of upload order).

Parameters:

Name	Type	Description	Default
batches	Sequence[MessageBatch]	Parsed source files, in caller order.	required

Returns:

Type	Description
list[SourceMessage]	Every distinct message, sorted by (timestamp, guid) — the same
list[SourceMessage]	stable order the parser emits, so equal input renders equally.

Source code in varagity/graph/render.py

def merge_batches(batches: Sequence[MessageBatch]) -> list[SourceMessage]:
    """Guid-merge parsed source files into one ordered message stream.

    The upsert seam (plan amendment, owner-approved 2026-07-25): overlapping
    ``chat.db`` uploads and re-exports of a growing database share message
    guids, so the union — not the concatenation — is what any engine should
    index. First occurrence wins: batches are processed in the order given,
    and a later batch's copy of an already-seen guid is dropped rather than
    overwriting (a re-export's rows are the same rows; preferring the first
    keeps a build reproducible regardless of upload order).

    Args:
        batches: Parsed source files, in caller order.

    Returns:
        Every distinct message, sorted by ``(timestamp, guid)`` — the same
        stable order the parser emits, so equal input renders equally.
    """
    merged: dict[str, SourceMessage] = {}
    duplicates = 0
    for batch in batches:
        for message in batch.messages:
            if message.guid in merged:
                duplicates += 1
                continue
            merged[message.guid] = message
    if duplicates:
        logger.info(
            "merged %d batch(es): %d message(s), %d duplicate guid(s) dropped",
            len(batches),
            len(merged),
            duplicates,
        )
    return sorted(merged.values(), key=lambda message: (message.timestamp, message.guid))

thread_transcripts(messages, *, max_chars=DEFAULT_TRANSCRIPT_MAX_CHARS)

Render messages into per-thread transcript documents split on day boundaries.

Days are the atom: consecutive days of one thread are packed into a document until the next day would push it past max_chars, and a day is never split in two. That keeps :attr:TranscriptDoc.doc_key meaningful and stable, and costs nothing in practice because both document-shaped engines chunk documents themselves before extraction. A single day whose messages exceed the cap therefore becomes one oversized document, logged rather than silently truncated.

Parameters:

Name	Type	Description	Default
messages	Sequence[SourceMessage]	Merged messages (any order; they are re-sorted here).	required
max_chars	int	Size target per document, in characters, counted over the message lines (the header adds a short constant on top).	DEFAULT_TRANSCRIPT_MAX_CHARS

Returns:

Type	Description
list[TranscriptDoc]	One document per thread-day-span, threads in first-appearance order
list[TranscriptDoc]	and documents in chronological order within a thread.

Raises:

Type	Description
ValueError	If max_chars is not positive.

Source code in varagity/graph/render.py

def thread_transcripts(
    messages: Sequence[SourceMessage],
    *,
    max_chars: int = DEFAULT_TRANSCRIPT_MAX_CHARS,
) -> list[TranscriptDoc]:
    """Render messages into per-thread transcript documents split on day boundaries.

    Days are the atom: consecutive days of one thread are packed into a
    document until the next day would push it past ``max_chars``, and a day
    is never split in two. That keeps :attr:`TranscriptDoc.doc_key` meaningful
    and stable, and costs nothing in practice because both document-shaped
    engines chunk documents themselves before extraction. A single day whose
    messages exceed the cap therefore becomes one oversized document, logged
    rather than silently truncated.

    Args:
        messages: Merged messages (any order; they are re-sorted here).
        max_chars: Size target per document, in characters, counted over the
            message lines (the header adds a short constant on top).

    Returns:
        One document per thread-day-span, threads in first-appearance order
        and documents in chronological order within a thread.

    Raises:
        ValueError: If ``max_chars`` is not positive.
    """
    if max_chars <= 0:
        raise ValueError(f"max_chars must be positive; got {max_chars}")
    docs: list[TranscriptDoc] = []
    for thread_id, thread_messages in _by_thread(messages).items():
        packed: list[SourceMessage] = []
        size = 0
        for day, day_messages in _by_day(thread_messages).items():
            block = len(_render_lines(day_messages)) + 1  # +1 for the joining newline
            if packed and size + block > max_chars:
                docs.append(_transcript_doc(thread_id, packed))
                packed, size = [], 0
            if not packed and block > max_chars:
                logger.info(
                    "thread %s on %s renders to %d chars (max_chars=%d) — kept whole: a day is "
                    "the transcript atom and the engines chunk documents themselves",
                    thread_id,
                    day.isoformat(),
                    block,
                    max_chars,
                )
            packed.extend(day_messages)
            size += block
        if packed:
            docs.append(_transcript_doc(thread_id, packed))
    return docs

varagity.graph.answer

Grounded answer synthesis over graph evidence (spec_graphrag §7; ADR-017).

ADR-017 chose LightRAG retrieval-only: the engine finds the evidence and this repo writes the answer, so a graph answer obeys the same grounding discipline as a chunk-RAG one. This module is that seam, shared by every engine adapter rather than reimplemented per engine — it started as the Graphiti adapter's private synthesis (the fact-shaped candidate had no answer pipeline of its own) and was generalized when it became the shipped design.

Two renderers, one diet: :func:synthesis_context grounds the harness's non-streaming :func:synthesize (and the adapter's +synthesis modes), while :func:graph_answer_context renders the same facts and passages in the chunk-RAG [SOURCE] block shape the streamed API path answers from. Keeping the two payloads identical is what lets eval graph keep guarding the shipped query path.

Two rules run through it:

• The context is the whole retrieval payload, not just relation facts: entities and relations answer "who/what", but the transcript excerpts are where the archive actually says things (plan decision #6). An engine that retrieves both and is scored on facts alone under-measures itself.
• Nothing here raises. A failed synthesis returns "" — a scored miss in the harness and a degrade-to-honest turn in the app, never a dead run. An empty context short-circuits to :data:NO_EVIDENCE_ANSWER without spending an LLM call at all.

Sizing follows the same clamp discipline as :func:varagity.graph.engines.lightrag.fit_max_tokens: llama.cpp with context shift disabled hard-500s at the window instead of stopping gracefully, and a retrieval payload is bounded by the engine's token budget (LightRAG's max_total_tokens defaults far above a 16k window), so the cap has to be applied here — see :func:synthesis_max_chars.

answer_context_max_chars(context_tokens, generation_tokens)

Size a grounding context so prompt + answer fits the context window.

Parameters:

Name	Type	Description	Default
context_tokens	int	The served model's context window (LLM_CONTEXT_TOKENS).	required
generation_tokens	int	The generation cap the answer call will request.	required

Returns:

Type	Description
int	The character budget, never below :data:_MIN_CONTEXT_CHARS.

Source code in varagity/graph/answer.py

def answer_context_max_chars(context_tokens: int, generation_tokens: int) -> int:
    """Size a grounding context so prompt + answer fits the context window.

    Args:
        context_tokens: The served model's context window
            (``LLM_CONTEXT_TOKENS``).
        generation_tokens: The generation cap the answer call will request.

    Returns:
        The character budget, never below :data:`_MIN_CONTEXT_CHARS`.
    """
    budget = context_tokens - generation_tokens - _PROMPT_HEADROOM_TOKENS
    return max(_MIN_CONTEXT_CHARS, budget * _CHARS_PER_TOKEN)

excerpts_block(excerpts, *, max_chars)

Render retrieved transcript passages as labelled prompt blocks.

Whole excerpts are packed until the next one would not fit; a single excerpt larger than the whole budget is truncated rather than dropped (one oversized passage must not cost the answer its only grounding). The label is [{thread_name} ({span})] — the same :func:transcript_label the shipped query path cites with, so the model learns one source format.

Parameters:

Name	Type	Description	Default
excerpts	Sequence[TranscriptExcerpt]	Retrieved passages, most relevant first (the engine's order is preserved: packing drops from the tail).	required
max_chars	int	Character budget for the whole block.	required

Returns:

Type	Description
str	The blocks, blank-line separated, never longer than max_chars.

Source code in varagity/graph/answer.py

def excerpts_block(excerpts: Sequence[TranscriptExcerpt], *, max_chars: int) -> str:
    """Render retrieved transcript passages as labelled prompt blocks.

    Whole excerpts are packed until the next one would not fit; a single
    excerpt larger than the whole budget is truncated rather than dropped
    (one oversized passage must not cost the answer its only grounding). The
    label is ``[{thread_name} ({span})]`` — the same
    :func:`transcript_label` the shipped query path cites with, so the model
    learns one source format.

    Args:
        excerpts: Retrieved passages, most relevant first (the engine's
            order is preserved: packing drops from the tail).
        max_chars: Character budget for the whole block.

    Returns:
        The blocks, blank-line separated, never longer than ``max_chars``.
    """
    return _pack(
        excerpts,
        max_chars=max_chars,
        render=lambda excerpt, body: f"[{transcript_label(excerpt)}]\n{body}",
    )

facts_block(evidence)

Render retrieved relations and communities as prompt facts.

Parameters:

Name	Type	Description	Default
evidence	GraphEvidence	The normalized evidence from a graph retrieval.	required

Returns:

Type	Description
str	One - fact line per relation (community summaries appended after
str	them), or "" when the retrieval surfaced no describable edge —
str	which is what makes the synthesis call skippable.

Source code in varagity/graph/answer.py

def facts_block(evidence: GraphEvidence) -> str:
    """Render retrieved relations and communities as prompt facts.

    Args:
        evidence: The normalized evidence from a graph retrieval.

    Returns:
        One ``- fact`` line per relation (community summaries appended after
        them), or ``""`` when the retrieval surfaced no describable edge —
        which is what makes the synthesis call skippable.
    """
    lines = [
        f"- {relation.description or relation.label or ''}".rstrip()
        for relation in evidence.relations
        if relation.description or relation.label
    ]
    lines.extend(
        f"- community {community.title or community.id}: {community.summary}"
        for community in evidence.communities
        if community.summary
    )
    return "\n".join(lines)

graph_answer_context(retrieval, *, max_chars)

Render one graph retrieval into the chunk-RAG grounding context.

:func:synthesis_context's twin for the shipped streaming path (stage-2 decision #5): the same diet — relation facts first, then as many transcript passages as the budget holds — rendered in the shape :data:varagity.generation.answer.ANSWER_PROMPT expects, so a graph turn inherits the repo's grounding prompt, <think> splitting, citation contract, abort handling, and token accounting whole.

The diets are deliberately identical: the eval graph harness scores :func:synthesis_context, and it can only keep guarding this path while both ground on the same evidence.

Parameters:

Name	Type	Description	Default
retrieval	GraphRetrieval	What the engine found (entities and relations plus the transcript passages).	required
max_chars	int	Character budget for the whole context — size it with :func:answer_context_max_chars.	required

Returns:

Type	Description
str	The context block, never longer than max_chars, and "" when
str	the retrieval surfaced nothing at all (which the grounding prompt
str	turns into an honest "I don't know", not a fabrication).

Source code in varagity/graph/answer.py

def graph_answer_context(retrieval: GraphRetrieval, *, max_chars: int) -> str:
    """Render one graph retrieval into the chunk-RAG grounding context.

    :func:`synthesis_context`'s twin for the shipped streaming path
    (stage-2 decision #5): the same diet — relation facts first, then as
    many transcript passages as the budget holds — rendered in the shape
    :data:`varagity.generation.answer.ANSWER_PROMPT` expects, so a graph
    turn inherits the repo's grounding prompt, ``<think>`` splitting,
    citation contract, abort handling, and token accounting whole.

    The diets are deliberately identical: the ``eval graph`` harness scores
    :func:`synthesis_context`, and it can only keep guarding this path while
    both ground on the same evidence.

    Args:
        retrieval: What the engine found (entities and relations plus the
            transcript passages).
        max_chars: Character budget for the whole context — size it with
            :func:`answer_context_max_chars`.

    Returns:
        The context block, never longer than ``max_chars``, and ``""`` when
        the retrieval surfaced nothing at all (which the grounding prompt
        turns into an honest "I don't know", not a fabrication).
    """
    sections: list[str] = []
    used = 0
    facts = facts_block(retrieval.evidence)
    if facts:
        section = _truncate(f"{_GRAPH_FACTS_HEADER}\n{facts}", max_chars)
        if section:
            sections.append(section)
            used = len(section)
    room = max_chars - used - (len(_SECTION_SEPARATOR) if sections else 0)
    blocks = transcript_blocks(retrieval.excerpts, max_chars=max(0, room))
    if blocks:
        sections.append(blocks)
    return _SECTION_SEPARATOR.join(sections)

synthesis_context(evidence, excerpts=(), *, max_chars)

Render one retrieval into the grounding context of a synthesis prompt.

Facts come first (they are small and they name the people), then as many transcript excerpts as the remaining budget holds. Empty sections are omitted entirely, so a retrieval that surfaced nothing renders to "" and :func:synthesize can skip the model call.

Parameters:

Name	Type	Description	Default
evidence	GraphEvidence	Entities, relations, and communities the engine retrieved.	required
excerpts	Sequence[TranscriptExcerpt]	Transcript passages the engine retrieved (empty for fact-shaped engines, which have no document plane).	()
max_chars	int	Character budget for the whole context — size it with :func:synthesis_max_chars.	required

Returns:

Type	Description
str	The context, never longer than max_chars.

Source code in varagity/graph/answer.py

def synthesis_context(
    evidence: GraphEvidence,
    excerpts: Sequence[TranscriptExcerpt] = (),
    *,
    max_chars: int,
) -> str:
    """Render one retrieval into the grounding context of a synthesis prompt.

    Facts come first (they are small and they name the people), then as many
    transcript excerpts as the remaining budget holds. Empty sections are
    omitted entirely, so a retrieval that surfaced nothing renders to ``""``
    and :func:`synthesize` can skip the model call.

    Args:
        evidence: Entities, relations, and communities the engine retrieved.
        excerpts: Transcript passages the engine retrieved (empty for
            fact-shaped engines, which have no document plane).
        max_chars: Character budget for the whole context — size it with
            :func:`synthesis_max_chars`.

    Returns:
        The context, never longer than ``max_chars``.
    """
    sections: list[str] = []
    used = 0
    facts = facts_block(evidence)
    if facts:
        # A budget too small even for the header leaves no section at all —
        # appending "" here would put a stray separator in front of the
        # excerpts.
        section = _truncate(f"{_FACTS_HEADER}\n{facts}", max_chars)
        if section:
            sections.append(section)
            used = len(section)
    room = max_chars - used - len(_EXCERPTS_HEADER) - 1
    if sections:
        room -= len(_SECTION_SEPARATOR)
    rendered = excerpts_block(excerpts, max_chars=max(0, room))
    if rendered:
        sections.append(f"{_EXCERPTS_HEADER}\n{rendered}")
    return _SECTION_SEPARATOR.join(sections)

synthesis_max_chars(context_tokens)

Size the synthesis context so prompt + answer fits the context window.

Parameters:

Name	Type	Description	Default
context_tokens	int	The served model's context window (LLM_CONTEXT_TOKENS).	required

Returns:

Type	Description
int	The character budget for :func:synthesis_context, never below
int	data:_MIN_CONTEXT_CHARS.

Source code in varagity/graph/answer.py

def synthesis_max_chars(context_tokens: int) -> int:
    """Size the synthesis context so prompt + answer fits the context window.

    Args:
        context_tokens: The served model's context window
            (``LLM_CONTEXT_TOKENS``).

    Returns:
        The character budget for :func:`synthesis_context`, never below
        :data:`_MIN_CONTEXT_CHARS`.
    """
    return answer_context_max_chars(context_tokens, SYNTHESIS_MAX_TOKENS)

synthesize(llm, question, context)

Write a grounded answer over one retrieval's context.

Parameters:

Name	Type	Description	Default
llm	LLMClient	The chat client (:class:~varagity.models.llm.LLMClient, reused rather than a bespoke HTTP call, so the app's retry, clamp, and context-window discipline all apply).	required
question	str	The question, verbatim — never the condensed or otherwise rewritten form (the repo-wide rule: the original words drive the answer prompt).	required
context	str	The rendered grounding context, from :func:synthesis_context.	required

Returns:

Type	Description
str	The <think>-stripped answer; :data:NO_EVIDENCE_ANSWER when the
str	context is empty (no model call), or "" when the call failed —
str	a scored miss, never a raise.

Source code in varagity/graph/answer.py

def synthesize(llm: LLMClient, question: str, context: str) -> str:
    """Write a grounded answer over one retrieval's context.

    Args:
        llm: The chat client (:class:`~varagity.models.llm.LLMClient`, reused
            rather than a bespoke HTTP call, so the app's retry, clamp, and
            context-window discipline all apply).
        question: The question, verbatim — never the condensed or otherwise
            rewritten form (the repo-wide rule: the original words drive the
            answer prompt).
        context: The rendered grounding context, from
            :func:`synthesis_context`.

    Returns:
        The ``<think>``-stripped answer; :data:`NO_EVIDENCE_ANSWER` when the
        context is empty (no model call), or ``""`` when the call failed —
        a scored miss, never a raise.
    """
    if not context.strip():
        return NO_EVIDENCE_ANSWER
    prompt = SYNTHESIS_PROMPT.format(context=context, question=question)
    try:
        raw = llm.generate(
            [{"role": "user", "content": prompt}], max_tokens=SYNTHESIS_MAX_TOKENS, verbose=0
        )
    except Exception:  # a failed synthesis is a scored miss, not a dead run
        logger.warning("graph answer synthesis failed", exc_info=True)
        return ""
    # Mandatory: generate() returns reasoning stages verbatim (the condense and
    # HyDE precedents), and an unstripped block would be scored as the answer.
    return clean_response(raw)

transcript_blocks(excerpts, *, max_chars)

Render retrieved transcript passages as [SOURCE] provenance blocks.

:func:excerpts_block's twin for the streamed answer path, which is read under :data:varagity.generation.answer.ANSWER_PROMPT rather than :data:SYNTHESIS_PROMPT: same packing, same labels, but the chunk-RAG block shape, so a graph turn's citations, chips, and evidence-card matching all work through the machinery chunk turns already use.

Parameters:

Name	Type	Description	Default
excerpts	Sequence[TranscriptExcerpt]	Retrieved passages, most relevant first.	required
max_chars	int	Character budget for the whole block.	required

Returns:

Type	Description
str	The blocks, blank-line separated, never longer than max_chars.

Source code in varagity/graph/answer.py

def transcript_blocks(excerpts: Sequence[TranscriptExcerpt], *, max_chars: int) -> str:
    """Render retrieved transcript passages as ``[SOURCE]`` provenance blocks.

    :func:`excerpts_block`'s twin for the *streamed* answer path, which is
    read under :data:`varagity.generation.answer.ANSWER_PROMPT` rather than
    :data:`SYNTHESIS_PROMPT`: same packing, same labels, but the chunk-RAG
    block shape, so a graph turn's citations, chips, and evidence-card
    matching all work through the machinery chunk turns already use.

    Args:
        excerpts: Retrieved passages, most relevant first.
        max_chars: Character budget for the whole block.

    Returns:
        The blocks, blank-line separated, never longer than ``max_chars``.
    """
    return _pack(
        excerpts,
        max_chars=max_chars,
        render=lambda excerpt, body: format_source_block(
            source=transcript_label(excerpt), context="", content=body
        ),
    )

transcript_label(excerpt)

Render the citation label of one transcript passage.

The one owner of the label, because it is matched three ways: the model cites it out of the prompt, the evidence panel keys its cards by it, and the persisted message_sources snapshot stores it. A day-grain label is also the ADR-017 honesty line — the engines cannot attribute a single message, so nothing here pretends to.

Parameters:

Name	Type	Description	Default
excerpt	TranscriptExcerpt	The retrieved passage.	required

Returns:

Type	Description
str	"{thread_name} ({span})".

Source code in varagity/graph/answer.py

def transcript_label(excerpt: TranscriptExcerpt) -> str:
    """Render the citation label of one transcript passage.

    The one owner of the label, because it is matched three ways: the model
    cites it out of the prompt, the evidence panel keys its cards by it, and
    the persisted ``message_sources`` snapshot stores it. A day-grain label
    is also the ADR-017 honesty line — the engines cannot attribute a single
    message, so nothing here pretends to.

    Args:
        excerpt: The retrieved passage.

    Returns:
        ``"{thread_name} ({span})"``.
    """
    return f"{excerpt.thread_name} ({excerpt.span})"

Workdir state and the process-wide handle

varagity.graph.manifest

The graph workdir's sidecar files — build diff and durable provenance.

Two small JSON files live beside the engine's own storage inside the workdir, both owned by this module and both written atomically (tmp + :func:os.replace, the discipline the engine's own storages use):

• :data:MANIFEST_FILENAME — doc_key → {content_sha256, message_guids, thread_name, span} for every transcript document the graph holds. It earns its keep twice (stage-2 decision #9):

• It is the upsert diff LightRAG refuses to do. The engine's enqueue stage drops a re-submitted doc_id in any status, so a re-exported chat.db whose most recent thread-day gained messages would silently keep the stale transcript. Comparing rendered content hashes against the manifest is what turns that into an explicit delete-then-reinsert (:attr:ManifestDiff.changed).

• It is a durable provenance index. Stage 1 rebuilt doc_key → guids inside build() and lost it on every restart, so a re-opened session (or a --skip-build re-score) could not map a citation back to messages. The manifest survives both.

• :data:SUMMARY_FILENAME — the last known graph size, refreshed by build and delete. It is what :meth:~varagity.graph.base.GraphSession.stats and the Prometheus gauges read, so neither has to walk the graphml at request or scrape time.

Neither file is authoritative over the engine: both are caches of what the last write did. Anything unreadable, malformed, or written by a different schema version degrades to "no manifest" — i.e. every rendered document looks new, which is safe (the engine's own dedup absorbs the re-enqueue) where trusting a mis-shaped file would not be.

ManifestDiff

Bases: BaseModel

What one rendered corpus changes about an indexed workdir.

Attributes:

Name	Type	Description
new	list[str]	Document keys the graph has never seen.
changed	list[str]	Keys whose rendered content no longer hashes to what was indexed — the engine must delete these before re-inserting, since its enqueue stage drops an already-known doc_id.
unchanged	list[str]	Keys whose content is byte-identical to what is indexed; re-inserting them would cost extraction for nothing.
removed	list[str]	Keys the workdir holds that this render does not mention. A full-corpus render may delete them all; a bounded build (message_limit / since) is deliberately partial, so it may delete only the re-span casualties :meth:WorkdirManifest.respanned picks out — deleting the genuinely out-of-window rest on a partial render's say-so would erase the archive.

Source code in varagity/graph/manifest.py

class ManifestDiff(BaseModel):
    """What one rendered corpus changes about an indexed workdir.

    Attributes:
        new: Document keys the graph has never seen.
        changed: Keys whose rendered content no longer hashes to what was
            indexed — the engine must **delete** these before re-inserting,
            since its enqueue stage drops an already-known ``doc_id``.
        unchanged: Keys whose content is byte-identical to what is indexed;
            re-inserting them would cost extraction for nothing.
        removed: Keys the workdir holds that this render does not mention.
            A **full-corpus** render may delete them all; a bounded build
            (``message_limit`` / ``since``) is deliberately partial, so it
            may delete only the re-span casualties
            :meth:`WorkdirManifest.respanned` picks out — deleting the
            genuinely out-of-window rest on a partial render's say-so would
            erase the archive.
    """

    new: list[str] = []
    changed: list[str] = []
    unchanged: list[str] = []
    removed: list[str] = []

    @property
    def pending(self) -> list[str]:
        """Keys that must be (re-)indexed: the new ones and the changed ones.

        Returns:
            ``new + changed``, in that order.
        """
        return [*self.new, *self.changed]

pending property

Keys that must be (re-)indexed: the new ones and the changed ones.

Returns:

Type	Description
list[str]	new + changed, in that order.

ManifestDoc

Bases: BaseModel

One indexed transcript document, as the workdir remembers it.

Attributes:

Name	Type	Description
content_sha256	str	Hash of the rendered transcript text. Rendering is a pure function of the merged messages, so an unchanged conversation re-renders to an identical hash no matter how many times its source file is re-uploaded.
message_guids	list[str]	The document's messages, in transcript order — the provenance map behind a document-grain citation.
thread_name	str	Human-facing thread label at index time.
span	str	The document's day span (YYYY-MM-DD or first..last).

Source code in varagity/graph/manifest.py

class ManifestDoc(BaseModel):
    """One indexed transcript document, as the workdir remembers it.

    Attributes:
        content_sha256: Hash of the rendered transcript text. Rendering is a
            pure function of the merged messages, so an unchanged
            conversation re-renders to an identical hash no matter how many
            times its source file is re-uploaded.
        message_guids: The document's messages, in transcript order — the
            provenance map behind a document-grain citation.
        thread_name: Human-facing thread label at index time.
        span: The document's day span (``YYYY-MM-DD`` or ``first..last``).
    """

    content_sha256: str
    message_guids: list[str] = []
    thread_name: str = ""
    span: str = ""

WorkdirManifest

Bases: BaseModel

Every transcript document one workdir's graph was built from.

Attributes:

Name	Type	Description
version	int	Schema version (:data:MANIFEST_VERSION).
docs	dict[str, ManifestDoc]	doc_key → what was indexed under it.

Source code in varagity/graph/manifest.py

class WorkdirManifest(BaseModel):
    """Every transcript document one workdir's graph was built from.

    Attributes:
        version: Schema version (:data:`MANIFEST_VERSION`).
        docs: ``doc_key`` → what was indexed under it.
    """

    version: int = MANIFEST_VERSION
    docs: dict[str, ManifestDoc] = {}

    def guid_index(self) -> dict[str, list[str]]:
        """Project the manifest into the provenance index the adapters use.

        Returns:
            ``doc_key`` → its message guids, the same shape
            :func:`varagity.graph.render.doc_guid_index` produces from a
            fresh render.
        """
        return {key: list(doc.message_guids) for key, doc in self.docs.items()}

    def thread_name_index(self) -> dict[str, str]:
        """Project the manifest into the display-name map the adapters use.

        A retrieved chunk-grain passage carries no transcript header, so its
        excerpt would otherwise be labelled with the thread *id* while a
        doc-grain hit of the same document shows the rendered display name —
        two labels for one citation target. This index resolves both grains
        to the name the document was indexed under.

        Returns:
            ``doc_key`` → the thread's human-facing label at index time.
            Records without a name are omitted, so a miss falls through to
            the caller's next resolution step instead of blanking the label.
        """
        return {key: doc.thread_name for key, doc in self.docs.items() if doc.thread_name}

    def message_guid_count(self) -> int:
        """Count the distinct messages the manifest accounts for.

        Returns:
            The number of distinct guids across every document (documents
            never overlap in practice, but distinctness is what the number
            claims).
        """
        return len({guid for doc in self.docs.values() for guid in doc.message_guids})

    def diff(self, docs: Sequence[TranscriptDoc]) -> ManifestDiff:
        """Classify a freshly rendered corpus against what is indexed.

        Args:
            docs: The rendered transcripts (see
                :func:`varagity.graph.render.thread_transcripts`).

        Returns:
            The four-way split. Rendered order is preserved within each
            bucket; ``removed`` is sorted, having no render order to inherit.
        """
        diff = ManifestDiff()
        rendered: set[str] = set()
        for doc in docs:
            rendered.add(doc.doc_key)
            known = self.docs.get(doc.doc_key)
            if known is None:
                diff.new.append(doc.doc_key)
            elif known.content_sha256 == content_sha256(doc.text):
                diff.unchanged.append(doc.doc_key)
            else:
                diff.changed.append(doc.doc_key)
        diff.removed = sorted(key for key in self.docs if key not in rendered)
        return diff

    def respanned(self, removed: Sequence[str], docs: Sequence[TranscriptDoc]) -> list[str]:
        """Pick out the removed keys that are re-span casualties of one render.

        Day-span packing is greedy, which makes document keys
        window-relative: growing a bounded window backward shifts a thread's
        pack boundaries, re-keying downstream documents whose messages did
        not change. Their old keys then read as ``removed`` on every later
        build, and keeping them (the bounded default) piles duplicate
        transcripts — and inflated entity mentions — into the graph at each
        growth step.

        Full guid coverage is the proof that deleting one loses nothing:
        every message its record accounts for is in this render, being
        (re-)indexed under the new keys. Partial or unknown coverage — a
        record without guids proves nothing — reads as genuinely
        out-of-window content and is kept, per the bounded-build contract.

        Args:
            removed: Removed keys from :meth:`diff`, in diff order.
            docs: The same build's rendered transcripts.

        Returns:
            The removed keys a bounded build may safely prune, input order
            preserved.
        """
        rendered = {guid for doc in docs for guid in doc.message_guids}
        return [
            key
            for key in removed
            if (known := self.docs.get(key)) is not None
            and known.message_guids
            and rendered.issuperset(known.message_guids)
        ]

    def merged(
        self,
        docs: Sequence[TranscriptDoc],
        *,
        prune: bool,
        retain: Collection[str] = (),
    ) -> "WorkdirManifest":
        """Fold a rendered corpus into a new manifest.

        Args:
            docs: The rendered transcripts the build just indexed.
            prune: Whether the render was the **whole** corpus. ``True``
                drops documents this render did not mention (their source
                messages are gone); ``False`` keeps them, which is what a
                bounded build needs — its render is partial by construction.
            retain: Keys whose **existing** record must survive verbatim,
                for documents the engine refused to delete. Their old
                content is still in the graph, so writing the new hash would
                make the next build believe they are up to date and the
                stale transcript would become permanent — the exact failure
                this manifest exists to prevent.

        Returns:
            The new manifest (this one is left untouched).
        """
        entries = {} if prune else dict(self.docs)
        for doc in docs:
            if doc.doc_key in retain:
                continue
            entries[doc.doc_key] = ManifestDoc(
                content_sha256=content_sha256(doc.text),
                message_guids=list(doc.message_guids),
                thread_name=doc.thread_name,
                span=doc.doc_key.partition("::")[2],
            )
        for key in retain:
            if (known := self.docs.get(key)) is not None:
                entries[key] = known
        return WorkdirManifest(version=MANIFEST_VERSION, docs=entries)

    def without(self, doc_keys: Sequence[str]) -> "WorkdirManifest":
        """Drop documents that were deleted from the graph.

        Args:
            doc_keys: The keys removed from the engine.

        Returns:
            The new manifest (this one is left untouched).
        """
        dropped = set(doc_keys)
        return WorkdirManifest(
            version=MANIFEST_VERSION,
            docs={key: doc for key, doc in self.docs.items() if key not in dropped},
        )

diff(docs)

Classify a freshly rendered corpus against what is indexed.

Parameters:

Name	Type	Description	Default
docs	Sequence[TranscriptDoc]	The rendered transcripts (see :func:varagity.graph.render.thread_transcripts).	required

Returns:

Type	Description
ManifestDiff	The four-way split. Rendered order is preserved within each
ManifestDiff	bucket; removed is sorted, having no render order to inherit.

Source code in varagity/graph/manifest.py

def diff(self, docs: Sequence[TranscriptDoc]) -> ManifestDiff:
    """Classify a freshly rendered corpus against what is indexed.

    Args:
        docs: The rendered transcripts (see
            :func:`varagity.graph.render.thread_transcripts`).

    Returns:
        The four-way split. Rendered order is preserved within each
        bucket; ``removed`` is sorted, having no render order to inherit.
    """
    diff = ManifestDiff()
    rendered: set[str] = set()
    for doc in docs:
        rendered.add(doc.doc_key)
        known = self.docs.get(doc.doc_key)
        if known is None:
            diff.new.append(doc.doc_key)
        elif known.content_sha256 == content_sha256(doc.text):
            diff.unchanged.append(doc.doc_key)
        else:
            diff.changed.append(doc.doc_key)
    diff.removed = sorted(key for key in self.docs if key not in rendered)
    return diff

guid_index()

Project the manifest into the provenance index the adapters use.

Returns:

Type	Description
dict[str, list[str]]	doc_key → its message guids, the same shape
dict[str, list[str]]	func:varagity.graph.render.doc_guid_index produces from a
dict[str, list[str]]	fresh render.

Source code in varagity/graph/manifest.py

def guid_index(self) -> dict[str, list[str]]:
    """Project the manifest into the provenance index the adapters use.

    Returns:
        ``doc_key`` → its message guids, the same shape
        :func:`varagity.graph.render.doc_guid_index` produces from a
        fresh render.
    """
    return {key: list(doc.message_guids) for key, doc in self.docs.items()}

merged(docs, *, prune, retain=())

Fold a rendered corpus into a new manifest.

Parameters:

Name	Type	Description	Default
docs	Sequence[TranscriptDoc]	The rendered transcripts the build just indexed.	required
prune	bool	Whether the render was the whole corpus. True drops documents this render did not mention (their source messages are gone); False keeps them, which is what a bounded build needs — its render is partial by construction.	required
retain	Collection[str]	Keys whose existing record must survive verbatim, for documents the engine refused to delete. Their old content is still in the graph, so writing the new hash would make the next build believe they are up to date and the stale transcript would become permanent — the exact failure this manifest exists to prevent.	()

Returns:

Type	Description
WorkdirManifest	The new manifest (this one is left untouched).

Source code in varagity/graph/manifest.py

def merged(
    self,
    docs: Sequence[TranscriptDoc],
    *,
    prune: bool,
    retain: Collection[str] = (),
) -> "WorkdirManifest":
    """Fold a rendered corpus into a new manifest.

    Args:
        docs: The rendered transcripts the build just indexed.
        prune: Whether the render was the **whole** corpus. ``True``
            drops documents this render did not mention (their source
            messages are gone); ``False`` keeps them, which is what a
            bounded build needs — its render is partial by construction.
        retain: Keys whose **existing** record must survive verbatim,
            for documents the engine refused to delete. Their old
            content is still in the graph, so writing the new hash would
            make the next build believe they are up to date and the
            stale transcript would become permanent — the exact failure
            this manifest exists to prevent.

    Returns:
        The new manifest (this one is left untouched).
    """
    entries = {} if prune else dict(self.docs)
    for doc in docs:
        if doc.doc_key in retain:
            continue
        entries[doc.doc_key] = ManifestDoc(
            content_sha256=content_sha256(doc.text),
            message_guids=list(doc.message_guids),
            thread_name=doc.thread_name,
            span=doc.doc_key.partition("::")[2],
        )
    for key in retain:
        if (known := self.docs.get(key)) is not None:
            entries[key] = known
    return WorkdirManifest(version=MANIFEST_VERSION, docs=entries)

message_guid_count()

Count the distinct messages the manifest accounts for.

Returns:

Type	Description
int	The number of distinct guids across every document (documents
int	never overlap in practice, but distinctness is what the number
int	claims).

Source code in varagity/graph/manifest.py

def message_guid_count(self) -> int:
    """Count the distinct messages the manifest accounts for.

    Returns:
        The number of distinct guids across every document (documents
        never overlap in practice, but distinctness is what the number
        claims).
    """
    return len({guid for doc in self.docs.values() for guid in doc.message_guids})

respanned(removed, docs)

Pick out the removed keys that are re-span casualties of one render.

Day-span packing is greedy, which makes document keys window-relative: growing a bounded window backward shifts a thread's pack boundaries, re-keying downstream documents whose messages did not change. Their old keys then read as removed on every later build, and keeping them (the bounded default) piles duplicate transcripts — and inflated entity mentions — into the graph at each growth step.

Full guid coverage is the proof that deleting one loses nothing: every message its record accounts for is in this render, being (re-)indexed under the new keys. Partial or unknown coverage — a record without guids proves nothing — reads as genuinely out-of-window content and is kept, per the bounded-build contract.

Parameters:

Name	Type	Description	Default
removed	Sequence[str]	Removed keys from :meth:diff, in diff order.	required
docs	Sequence[TranscriptDoc]	The same build's rendered transcripts.	required

Returns:

Type	Description
list[str]	The removed keys a bounded build may safely prune, input order
list[str]	preserved.

Source code in varagity/graph/manifest.py

def respanned(self, removed: Sequence[str], docs: Sequence[TranscriptDoc]) -> list[str]:
    """Pick out the removed keys that are re-span casualties of one render.

    Day-span packing is greedy, which makes document keys
    window-relative: growing a bounded window backward shifts a thread's
    pack boundaries, re-keying downstream documents whose messages did
    not change. Their old keys then read as ``removed`` on every later
    build, and keeping them (the bounded default) piles duplicate
    transcripts — and inflated entity mentions — into the graph at each
    growth step.

    Full guid coverage is the proof that deleting one loses nothing:
    every message its record accounts for is in this render, being
    (re-)indexed under the new keys. Partial or unknown coverage — a
    record without guids proves nothing — reads as genuinely
    out-of-window content and is kept, per the bounded-build contract.

    Args:
        removed: Removed keys from :meth:`diff`, in diff order.
        docs: The same build's rendered transcripts.

    Returns:
        The removed keys a bounded build may safely prune, input order
        preserved.
    """
    rendered = {guid for doc in docs for guid in doc.message_guids}
    return [
        key
        for key in removed
        if (known := self.docs.get(key)) is not None
        and known.message_guids
        and rendered.issuperset(known.message_guids)
    ]

thread_name_index()

Project the manifest into the display-name map the adapters use.

A retrieved chunk-grain passage carries no transcript header, so its excerpt would otherwise be labelled with the thread id while a doc-grain hit of the same document shows the rendered display name — two labels for one citation target. This index resolves both grains to the name the document was indexed under.

Returns:

Type	Description
dict[str, str]	doc_key → the thread's human-facing label at index time.
dict[str, str]	Records without a name are omitted, so a miss falls through to
dict[str, str]	the caller's next resolution step instead of blanking the label.

Source code in varagity/graph/manifest.py

def thread_name_index(self) -> dict[str, str]:
    """Project the manifest into the display-name map the adapters use.

    A retrieved chunk-grain passage carries no transcript header, so its
    excerpt would otherwise be labelled with the thread *id* while a
    doc-grain hit of the same document shows the rendered display name —
    two labels for one citation target. This index resolves both grains
    to the name the document was indexed under.

    Returns:
        ``doc_key`` → the thread's human-facing label at index time.
        Records without a name are omitted, so a miss falls through to
        the caller's next resolution step instead of blanking the label.
    """
    return {key: doc.thread_name for key, doc in self.docs.items() if doc.thread_name}

without(doc_keys)

Drop documents that were deleted from the graph.

Parameters:

Name	Type	Description	Default
doc_keys	Sequence[str]	The keys removed from the engine.	required

Returns:

Type	Description
WorkdirManifest	The new manifest (this one is left untouched).

Source code in varagity/graph/manifest.py

def without(self, doc_keys: Sequence[str]) -> "WorkdirManifest":
    """Drop documents that were deleted from the graph.

    Args:
        doc_keys: The keys removed from the engine.

    Returns:
        The new manifest (this one is left untouched).
    """
    dropped = set(doc_keys)
    return WorkdirManifest(
        version=MANIFEST_VERSION,
        docs={key: doc for key, doc in self.docs.items() if key not in dropped},
    )

WorkdirSummary

Bases: BaseModel

The last known size of one workdir's graph.

A cache, not a source of truth: it is written by whatever last changed the graph, so a workdir mutated behind the adapter's back reports what the adapter last saw. That is the deliberate trade — Prometheus scrapes and status polls must never parse a multi-megabyte graphml.

Attributes:

Name	Type	Description
entities	int | None	Node count, or None when the graph would not say.
relations	int | None	Edge count, or None when the graph would not say.
message_guids	int	Messages the manifest accounts for.
docs	int	Transcript documents the manifest accounts for.
refreshed_at	datetime	When this summary was written (aware UTC) — the staleness signal for anything reading it.

Source code in varagity/graph/manifest.py

class WorkdirSummary(BaseModel):
    """The last known size of one workdir's graph.

    A cache, not a source of truth: it is written by whatever last changed
    the graph, so a workdir mutated behind the adapter's back reports what
    the adapter last saw. That is the deliberate trade — Prometheus scrapes
    and status polls must never parse a multi-megabyte graphml.

    Attributes:
        entities: Node count, or ``None`` when the graph would not say.
        relations: Edge count, or ``None`` when the graph would not say.
        message_guids: Messages the manifest accounts for.
        docs: Transcript documents the manifest accounts for.
        refreshed_at: When this summary was written (aware UTC) — the
            staleness signal for anything reading it.
    """

    entities: int | None = None
    relations: int | None = None
    message_guids: int = 0
    docs: int = 0
    refreshed_at: datetime

content_sha256(text)

Hash a rendered transcript for the build diff.

Parameters:

Name	Type	Description	Default
text	str	The rendered document text.	required

Returns:

Type	Description
str	The hex digest of its UTF-8 bytes.

Source code in varagity/graph/manifest.py

def content_sha256(text: str) -> str:
    """Hash a rendered transcript for the build diff.

    Args:
        text: The rendered document text.

    Returns:
        The hex digest of its UTF-8 bytes.
    """
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

load_manifest(workdir)

Read a workdir's manifest, degrading to an empty one.

A missing, unreadable, malformed, or differently-versioned file yields an empty manifest: every rendered document then looks new, the engine's own dedup absorbs the re-enqueue, and the next build writes a correct file. The alternative — trusting a file whose fields may mean something else — would corrupt the graph silently.

Parameters:

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

Returns:

Type	Description
WorkdirManifest	The manifest, or an empty one.

Source code in varagity/graph/manifest.py

def load_manifest(workdir: Path) -> WorkdirManifest:
    """Read a workdir's manifest, degrading to an empty one.

    A missing, unreadable, malformed, or differently-versioned file yields
    an empty manifest: every rendered document then looks new, the engine's
    own dedup absorbs the re-enqueue, and the next build writes a correct
    file. The alternative — trusting a file whose fields may mean something
    else — would corrupt the graph silently.

    Args:
        workdir: The engine's working directory.

    Returns:
        The manifest, or an empty one.
    """
    path = manifest_path(workdir)
    try:
        manifest = WorkdirManifest.model_validate_json(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return WorkdirManifest()
    except (OSError, ValueError):
        logger.warning("graph manifest at %s is unreadable — treating it as empty", path)
        return WorkdirManifest()
    if manifest.version != MANIFEST_VERSION:
        logger.warning(
            "graph manifest at %s is version %d, expected %d — treating it as empty",
            path,
            manifest.version,
            MANIFEST_VERSION,
        )
        return WorkdirManifest()
    return manifest

load_summary(workdir)

Read a workdir's summary sidecar.

Parameters:

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

Returns:

Type	Description
WorkdirSummary | None	The summary, or None when there is none to read (which callers
WorkdirSummary | None	answer by asking the graph itself).

Source code in varagity/graph/manifest.py

def load_summary(workdir: Path) -> WorkdirSummary | None:
    """Read a workdir's summary sidecar.

    Args:
        workdir: The engine's working directory.

    Returns:
        The summary, or ``None`` when there is none to read (which callers
        answer by asking the graph itself).
    """
    path = summary_path(workdir)
    try:
        return WorkdirSummary.model_validate_json(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return None
    except (OSError, ValueError):
        logger.warning("graph summary at %s is unreadable — ignoring it", path)
        return None

manifest_path(workdir)

Locate the manifest inside a workdir.

Parameters:

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

Returns:

Type	Description
Path	The manifest's path (which may not exist yet).

Source code in varagity/graph/manifest.py

def manifest_path(workdir: Path) -> Path:
    """Locate the manifest inside a workdir.

    Args:
        workdir: The engine's working directory.

    Returns:
        The manifest's path (which may not exist yet).
    """
    return workdir / MANIFEST_FILENAME

save_manifest(workdir, manifest)

Write a workdir's manifest atomically.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory (created if absent).	required
manifest	WorkdirManifest	The manifest to persist.	required

Source code in varagity/graph/manifest.py

def save_manifest(workdir: Path, manifest: WorkdirManifest) -> None:
    """Write a workdir's manifest atomically.

    Args:
        workdir: The engine's working directory (created if absent).
        manifest: The manifest to persist.
    """
    _write_atomic(manifest_path(workdir), manifest.model_dump_json(indent=2) + "\n")

save_summary(workdir, manifest, *, entities, relations)

Refresh a workdir's summary sidecar atomically.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory (created if absent).	required
manifest	WorkdirManifest	The manifest the document/message counts come from.	required
entities	int | None	Node count from the graph, or None if it would not say.	required
relations	int | None	Edge count from the graph, or None if it would not say.	required

Returns:

Type	Description
WorkdirSummary	The summary that was written.

Source code in varagity/graph/manifest.py

def save_summary(
    workdir: Path,
    manifest: WorkdirManifest,
    *,
    entities: int | None,
    relations: int | None,
) -> WorkdirSummary:
    """Refresh a workdir's summary sidecar atomically.

    Args:
        workdir: The engine's working directory (created if absent).
        manifest: The manifest the document/message counts come from.
        entities: Node count from the graph, or ``None`` if it would not say.
        relations: Edge count from the graph, or ``None`` if it would not say.

    Returns:
        The summary that was written.
    """
    summary = WorkdirSummary(
        entities=entities,
        relations=relations,
        message_guids=manifest.message_guid_count(),
        docs=len(manifest.docs),
        refreshed_at=datetime.now(UTC),
    )
    _write_atomic(summary_path(workdir), summary.model_dump_json(indent=2) + "\n")
    return summary

summary_path(workdir)

Locate the summary sidecar inside a workdir.

Parameters:

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

Returns:

Type	Description
Path	The sidecar's path (which may not exist yet).

Source code in varagity/graph/manifest.py

def summary_path(workdir: Path) -> Path:
    """Locate the summary sidecar inside a workdir.

    Args:
        workdir: The engine's working directory.

    Returns:
        The sidecar's path (which may not exist yet).
    """
    return workdir / SUMMARY_FILENAME

varagity.graph.service

The process-wide graph engine handle (spec_graphrag §5.2; ADR-017).

One process, one session, one writer. The engine's storages are single-writer per working directory by explicit invariant, so the graph corpus can only stay coherent if exactly one process ever opens it — which is why builds are API-only and there is deliberately no CLI graph build (stage-2 decision #4). This module is that single owner: a lazily-opened session behind a process singleton, in the shape :func:varagity.api.ingest_runner.get_ingest_runner established.

Two locks, doing different jobs:

• A build lock, non-reentrant and single-flight — the write lock. A second build attempt raises :class:GraphBuildInProgress rather than queueing, because the caller (an HTTP route) wants a 409, not a request that blocks for a day. Deletes take it too — they mutate the same graph — and so do a backup export and a backup restore (ADR-019), through :meth:GraphService.writing: holding it is what guarantees no file changes while the workdir is copied, and no build starts while it is swapped.
• An open lock, so two concurrent first-callers cannot open two sessions over the same directory.

Reads deliberately take neither. The session multiplexes its engine calls onto a loop thread, so queries, exports, and status polls are answered while a multi-day backfill is still extracting (stage-2 decision #10). That is the whole reason the session's threading model changed at stage 2, and the reason the graph corpus is usable during its own backfill.

A backup restore swaps the whole workdir (ADR-019): :meth:GraphService.replace_workdir moves a restored copy into place under both locks — the write lock, so no build can start, and the open lock, so no lazy open (a status poll, a /graph read) can land on a half-swapped folder — and puts the previous graph back if the new one won't open. Constructing the service finishes a swap that a crash interrupted, before anything in the process can read the workdir.

Nothing here raises on a broken engine except :class:GraphUnavailable, which callers turn into a structured degrade (ADR-017's degrade semantics) — a graph the app cannot open must cost the graph turn, never the process.

GraphBuildInProgress

Bases: RuntimeError

Raised when a build is requested while one is already running.

Source code in varagity/graph/service.py

class GraphBuildInProgress(RuntimeError):
    """Raised when a build is requested while one is already running."""

GraphService

Owner of the process's single graph session (see the module docstring).

Source code in varagity/graph/service.py

class GraphService:
    """Owner of the process's single graph session (see the module docstring)."""

    def __init__(self, engine_name: str | None = None, *, workdir: Path | None = None) -> None:
        """Resolve the engine and its working directory (both injectable).

        Nothing is opened here: a session that opened at import time would
        make an unbuilt graph an API startup failure, and would open the
        workdir in every process that merely imports this module. The one
        thing construction does on disk is finish a backup restore's swap
        that a crash interrupted (:meth:`_heal_interrupted_swap`) — first,
        before anything in this process can read, open, wipe or restore the
        workdir.

        Args:
            engine_name: Registry name of the engine; ``None`` reads
                ``GRAPH_ENGINE``.
            workdir: The engine's working directory; ``None`` derives
                :func:`graph_workdir` (``GRAPH_STORAGE_PATH/<engine>``,
                resolved absolute).

        Raises:
            KeyError: If ``engine_name`` names no registered engine.
        """
        self._engine_name = engine_name if engine_name is not None else get_settings().GRAPH_ENGINE
        self._engine = get_graph_engine(self._engine_name)
        self._workdir = workdir if workdir is not None else graph_workdir(self._engine_name)
        self._stack: ExitStack | None = None
        self._session: GraphSession | None = None
        self._open_lock = threading.Lock()
        self._build_lock = threading.Lock()
        self._heal_interrupted_swap()

    @property
    def engine_name(self) -> str:
        """Name of the engine this service resolved.

        Returns:
            The registry key (``GRAPH_ENGINE``).
        """
        return self._engine_name

    @property
    def workdir(self) -> Path:
        """Where the engine stores this process's graph.

        Returns:
            The absolute working directory.
        """
        return self._workdir

    @property
    def building(self) -> bool:
        """Whether a write — a build, resume, delete, or backup export or restore — holds it.

        Returns:
            ``True`` while one is in flight.
        """
        return self._build_lock.locked()

    def storage_format(self) -> EngineFormat:
        """Name the format the engine's workdir files are in (what a backup records).

        Returns:
            The engine's :meth:`~varagity.graph.base.GraphEngine.storage_format`
            — read from package metadata, so this opens no session.
        """
        return self._engine.storage_format()

    def session(self) -> GraphSession:
        """Return the open session, opening it on first use.

        Returns:
            The process's one session.

        Raises:
            GraphUnavailable: If the engine's session cannot be opened. The
                failure is not cached — the next call tries again, so a
                transient one (a model service still warming up) does not
                need a restart to clear.
        """
        with self._open_lock:
            if self._session is not None:
                return self._session
            logger.info("opening the %s graph session in %s", self._engine_name, self._workdir)
            stack = ExitStack()
            try:
                session = stack.enter_context(self._engine.session(self._workdir))
            except Exception as exc:
                stack.close()
                raise GraphUnavailable(
                    f"could not open the {self._engine_name!r} graph session "
                    f"in {self._workdir}: {exc}"
                ) from exc
            self._stack, self._session = stack, session
            return session

    def build(
        self,
        batches: Sequence[MessageBatch],
        *,
        verbose: int = 0,
        prune_removed: bool = True,
    ) -> BuildReport:
        """Upsert a rendered corpus into the graph, one build at a time.

        Args:
            batches: Parsed source files.
            verbose: Validated console verbosity (0–2).
            prune_removed: Whether ``batches`` render the whole corpus (see
                :meth:`varagity.graph.base.GraphSession.build`).

        Returns:
            What the build did.

        Raises:
            GraphBuildInProgress: If another build or delete is running.
            GraphUnavailable: If the session cannot be opened.
        """
        with self.writing():
            return self.session().build(batches, verbose=verbose, prune_removed=prune_removed)

    def resume(self, *, verbose: int = 0) -> BuildReport:
        """Finish whatever a killed build left in flight.

        Args:
            verbose: Validated console verbosity (0–2).

        Returns:
            What the pass did.

        Raises:
            GraphBuildInProgress: If another build or delete is running.
            GraphUnavailable: If the session cannot be opened.
        """
        with self.writing():
            return self.session().resume(verbose=verbose)

    def delete_documents(self, doc_keys: Sequence[str]) -> int:
        """Retract documents from the graph.

        Args:
            doc_keys: Transcript document keys to remove.

        Returns:
            How many deletions the engine accepted.

        Raises:
            GraphBuildInProgress: If a build is running — deletion rebuilds
                the entities a document partly supported, which may not race
                extraction.
            GraphUnavailable: If the session cannot be opened.
        """
        with self.writing():
            return self.session().delete_documents(doc_keys)

    def retrieve(
        self, question: str, *, mode: str | None = None, verbose: int = 0
    ) -> GraphRetrieval:
        """Find evidence for one question — never blocked by a build.

        The chat path's read: the API streams its own grounded answer over
        what this returns, so no engine answer call is spent here.

        Args:
            question: The search query, verbatim.
            mode: Engine query mode; ``None`` uses the adapter's primary.
            verbose: Validated console verbosity (0–2).

        Returns:
            The evidence and transcript excerpts.

        Raises:
            GraphUnavailable: If the session cannot be opened (the caller
                turns that into ADR-017's per-turn degrade).
        """
        return self.session().retrieve(question, mode=mode, verbose=verbose)

    def query(self, question: str, *, mode: str | None = None, verbose: int = 0) -> GraphAnswer:
        """Answer one question from the graph — never blocked by a build.

        Args:
            question: The question, verbatim.
            mode: Engine query mode; ``None`` uses the adapter's primary.
            verbose: Validated console verbosity (0–2).

        Returns:
            The answer with its evidence.

        Raises:
            GraphUnavailable: If the session cannot be opened.
        """
        return self.session().query(question, mode=mode, verbose=verbose)

    def stats(self) -> GraphStats:
        """Report the graph's size.

        Returns:
            Entity/relation/community counts, each ``None`` where unknown.

        Raises:
            GraphUnavailable: If the session cannot be opened.
        """
        return self.session().stats()

    def document_statuses(self) -> dict[str, int]:
        """Count the engine's documents by processing status.

        Returns:
            Status name → document count, empty when the engine would not say.

        Raises:
            GraphUnavailable: If the session cannot be opened.
        """
        return self.session().document_statuses()

    def export(
        self,
        label: str = "*",
        *,
        max_depth: int = 3,
        max_nodes: int = 5000,
    ) -> GraphExport:
        """Read a renderable slice of the graph.

        Args:
            label: Entity name to centre on; ``"*"`` is the whole graph.
            max_depth: Hops to walk out from ``label``.
            max_nodes: Node cap.

        Returns:
            The slice.

        Raises:
            GraphUnavailable: If the session cannot be opened.
        """
        return self.session().export(label, max_depth=max_depth, max_nodes=max_nodes)

    def close(self) -> None:
        """Tear the session down (API lifespan shutdown); idempotent.

        A teardown that raised would take an orderly shutdown with it, so
        failures are logged and swallowed — the session is dropped either
        way, and the next call opens a fresh one. The teardown runs under
        the open lock, so a concurrent first caller waits for it instead of
        opening a second session over files the first is still flushing.
        """
        with self._open_lock:
            self._close_locked()

    def _close_locked(self) -> None:
        """Tear the session down; the caller holds the open lock.

        Closing finalizes the engine's storages (flushing what they buffer
        into the workdir) and resets the engine library's process-global
        state, so the next open reads the workdir from disk rather than
        re-attaching to the old session's memory.
        """
        stack, self._stack, self._session = self._stack, None, None
        if stack is None:
            return
        try:
            stack.close()
        except Exception:
            logger.warning("graph session teardown failed", exc_info=True)

    def staging_dir(self, run_id: str) -> Path:
        """Name the folder a restore extracts a bundle into before the swap.

        ``<workdir>.restore-<run>``: beside the workdir, so on the same
        filesystem — every rename of :meth:`replace_workdir` stays atomic.
        Nothing is created here.

        Args:
            run_id: The restore run's id.

        Returns:
            The staging folder's path.

        Raises:
            ValueError: If ``run_id`` isn't a safe folder-name suffix.
        """
        return self._swap_dir(_STAGED, run_id)

    def replace_workdir(
        self,
        staged: Path,
        run_id: str,
        *,
        on_open: Callable[[], None] | None = None,
    ) -> GraphStats:
        """Swap a restored working directory in for the live one (a graph restore, ADR-019).

        ``_reset_workdir``'s close → change → reopen move, made safe to lose.
        Under the **write lock** throughout (no build, resume, delete or
        export can start), and under the **open lock** for the renames (no
        lazy open — a status poll, a ``/graph`` read — can land on a
        half-swapped folder):

        1. close the session — the engine flushes what it buffers into the
           *old* files, and its process-global state is reset so the next
           open reads the new ones from disk;
        2. move the live workdir aside to ``<workdir>.previous-<run>`` (when
           there is one), then ``staged`` into its place;
        3. open the restored graph and read its size — opening is what
           checks the files (LightRAG asserts its vectors' ``embedding_dim``);
        4. rename the previous graph to ``<workdir>.replaced-<run>`` — the
           **commit point**: until it, any failure (or a crash, which the
           next process's construction heals) brings the previous graph
           back; after it, the restored graph stands and the replaced one is
           deleted.

        A failure before the commit undoes the swap: the restored folder
        goes back to ``staged`` (its owner removes it), the previous graph
        returns and its session is reopened, and the error is re-raised.

        Reads stay unlocked outside the renames, so a graph turn already
        streaming when the session closes can end in an error (chat refuses
        *new* graph turns for the whole restore, before they open one).

        Args:
            staged: The restored workdir, complete — at :meth:`staging_dir`,
                so the renames stay on one filesystem.
            run_id: The restore run's id (names the folders beside the
                workdir).
            on_open: Called once the restored folder is in place, just before
                it is opened — the restore's ``open`` stage.

        Returns:
            The restored graph's size, as its freshly opened session reports
            it.

        Raises:
            GraphBuildInProgress: If a write holds the write lock (nothing
                was moved).
            GraphUnavailable: If the restored graph would not open (the
                previous one is back in place).
            OSError: If a rename fails (whatever had moved is moved back
                first).
        """
        previous = self._swap_dir(_PREVIOUS, run_id)
        replaced = self._swap_dir(_REPLACED, run_id)
        with self.writing():
            had_previous = self._swap_in(staged, previous)
            try:
                if on_open is not None:
                    on_open()
                stats = self.session().stats()
                if had_previous:
                    os.replace(previous, replaced)  # the commit point
            except Exception:
                self._swap_back(staged, previous, had_previous=had_previous)
                raise
            fsync_directory(self._workdir.parent)
            shutil.rmtree(replaced, ignore_errors=True)
            if replaced.exists(follow_symlinks=False):
                logger.warning(
                    "could not delete the replaced graph (%s) — the next API start removes it",
                    replaced.name,
                )
        logger.info("swapped a restored graph into %s", self._workdir.name)
        return stats

    def _swap_dir(self, kind: str, run_id: str) -> Path:
        """Name one of a restore's folders beside the workdir.

        Args:
            kind: ``restore`` | ``previous`` | ``replaced``.
            run_id: The restore run's id.

        Returns:
            ``<workdir>.<kind>-<run_id>``.

        Raises:
            ValueError: If ``run_id`` isn't a safe folder-name suffix.
        """
        if _RUN_ID.fullmatch(run_id) is None:
            raise ValueError(f"{run_id!r} can't name a folder beside the graph workdir")
        return self._workdir.with_name(f"{self._workdir.name}.{kind}-{run_id}")

    def _swap_in(self, staged: Path, previous: Path) -> bool:
        """Close the session and move the restored folder into the workdir's place.

        Both renames run under the open lock. A failed second rename puts
        the live workdir back before re-raising.

        Args:
            staged: The restored workdir.
            previous: Where the live workdir moves aside to.

        Returns:
            Whether there was a live workdir to move aside.

        Raises:
            OSError: If a rename fails.
        """
        fsync_directory(staged)  # the extracted names are durable before they go live
        with self._open_lock:
            self._close_locked()
            had_previous = self._workdir.exists(follow_symlinks=False)
            if had_previous:
                os.replace(self._workdir, previous)
            try:
                os.replace(staged, self._workdir)
            except OSError:
                if had_previous:
                    os.replace(previous, self._workdir)
                raise
        fsync_directory(self._workdir.parent)
        return had_previous

    def _swap_back(self, staged: Path, previous: Path, *, had_previous: bool) -> None:
        """Undo a swap whose restored graph would not open (or not commit).

        The restored folder returns to ``staged`` (its owner deletes it),
        the previous graph returns to the workdir, and — when there was one
        — its session is reopened. Never raises: the caller is already
        re-raising the failure that got here, and a rename that fails now
        leaves ``<workdir>.previous-<run>`` for the next process's
        construction to move back.

        Args:
            staged: Where the restored folder goes back to.
            previous: Where the previous graph waits.
            had_previous: Whether there was a previous graph at all (a
                restore onto a wiped stack has none to reopen).
        """
        with self._open_lock:
            self._close_locked()
            try:
                if self._workdir.exists(follow_symlinks=False):
                    os.replace(self._workdir, staged)
                if had_previous:
                    os.replace(previous, self._workdir)
            except OSError:
                logger.exception(
                    "could not put the previous graph back after a failed restore — the next "
                    "API start moves %s back into place",
                    previous.name,
                )
                return
        if not had_previous:
            logger.warning("the restored graph would not open — the graph is left unbuilt")
            return
        logger.warning("the restored graph would not open — the previous graph is back in place")
        try:
            self.session()
        except GraphUnavailable as error:
            logger.warning("the previous graph would not reopen either: %s", error)

    def _heal_interrupted_swap(self) -> None:
        """Finish a restore's swap that a crash interrupted (best-effort; at construction).

        A restore parks folders beside the workdir while it swaps (see
        :meth:`replace_workdir`), and a crash can leave any of them behind:

        * **a previous graph** (``.previous-<run>``) — the swap never reached
          its commit point, so the restore did not happen: whatever stands
          at the workdir (an unconfirmed restored graph, or nothing — the
          crash came between the renames) is removed and the previous graph
          moves back;
        * **a replaced graph** (``.replaced-<run>``) — the swap committed and
          the crash came mid-delete: it is removed, and the restored graph
          stands;
        * **a staging folder** (``.restore-<run>``) — a bundle extracted
          (perhaps partly) for a swap that never committed: it is removed.

        Run at construction — before anything in this process can read,
        open, wipe or restore the workdir — so a missing workdir can only
        mean a crash (never this process's own reingest wipe), and no
        restore of this process can have a folder in flight yet. Failures
        are logged, never raised: a graph that can't be healed must not cost
        the API its graph service; the next start tries again.
        """
        try:
            previous = self._swap_leftovers(_PREVIOUS)
            if previous:
                if self._workdir.exists(follow_symlinks=False):
                    shutil.rmtree(self._workdir)
                os.replace(previous[0], self._workdir)
                logger.warning(
                    "a graph restore was interrupted before it committed — %s is back in place",
                    self._workdir.name,
                )
            leftovers = [
                *previous[1:],
                *self._swap_leftovers(_REPLACED),
                *self._swap_leftovers(_STAGED),
            ]
            for leftover in leftovers:
                shutil.rmtree(leftover, ignore_errors=True)
                logger.info("removed %s, left by an interrupted graph restore", leftover.name)
        except OSError:
            logger.warning(
                "could not finish an interrupted graph restore beside %s",
                self._workdir.name,
                exc_info=True,
            )

    def _swap_leftovers(self, kind: str) -> list[Path]:
        """Find the folders of one kind a restore left beside the workdir.

        Args:
            kind: ``restore`` | ``previous`` | ``replaced``.

        Returns:
            The folders, most recently modified first (empty when the
            workdir's parent doesn't exist yet).

        Raises:
            OSError: If the parent can't be listed.
        """
        parent = self._workdir.parent
        if not parent.is_dir():
            return []
        prefix = f"{self._workdir.name}.{kind}-"
        found = [
            entry
            for entry in parent.iterdir()
            if entry.name.startswith(prefix)
            and _RUN_ID.fullmatch(entry.name[len(prefix) :]) is not None
            and entry.is_dir()
            and not entry.is_symlink()
        ]
        return sorted(found, key=lambda entry: entry.stat().st_mtime_ns, reverse=True)

    def writing(self) -> "_WriteLock":
        """Take the single-flight write lock for the duration of a block.

        The lock :meth:`build`, :meth:`resume` and :meth:`delete_documents`
        take, public so a caller that must see the workdir hold still can
        take it too — a backup export copies the files under it (ADR-019),
        so no build, resume or delete can start mid-copy, while reads stay
        unlocked and chat keeps answering (:meth:`replace_workdir` swaps a
        restored workdir in under it). Never blocks: a held lock is an
        answer (the caller's ``409``), not a wait.

        Returns:
            A context manager holding the lock (released on block exit,
            however the block ends).

        Raises:
            GraphBuildInProgress: If it is already held.
        """
        if not self._build_lock.acquire(blocking=False):
            raise GraphBuildInProgress(
                "the graph is being written — a build, a delete, or a backup export or restore "
                "holds its write lock"
            )
        return _WriteLock(self._build_lock)

building property

Whether a write — a build, resume, delete, or backup export or restore — holds it.

Returns:

Type	Description
bool	True while one is in flight.

engine_name property

Name of the engine this service resolved.

Returns:

Type	Description
str	The registry key (GRAPH_ENGINE).

workdir property

Where the engine stores this process's graph.

Returns:

Type	Description
Path	The absolute working directory.

__init__(engine_name=None, *, workdir=None)

Resolve the engine and its working directory (both injectable).

Nothing is opened here: a session that opened at import time would make an unbuilt graph an API startup failure, and would open the workdir in every process that merely imports this module. The one thing construction does on disk is finish a backup restore's swap that a crash interrupted (:meth:_heal_interrupted_swap) — first, before anything in this process can read, open, wipe or restore the workdir.

Parameters:

Name	Type	Description	Default
engine_name	str | None	Registry name of the engine; None reads GRAPH_ENGINE.	None
workdir	Path | None	The engine's working directory; None derives :func:graph_workdir (GRAPH_STORAGE_PATH/<engine>, resolved absolute).	None

Raises:

Type	Description
KeyError	If engine_name names no registered engine.

Source code in varagity/graph/service.py

def __init__(self, engine_name: str | None = None, *, workdir: Path | None = None) -> None:
    """Resolve the engine and its working directory (both injectable).

    Nothing is opened here: a session that opened at import time would
    make an unbuilt graph an API startup failure, and would open the
    workdir in every process that merely imports this module. The one
    thing construction does on disk is finish a backup restore's swap
    that a crash interrupted (:meth:`_heal_interrupted_swap`) — first,
    before anything in this process can read, open, wipe or restore the
    workdir.

    Args:
        engine_name: Registry name of the engine; ``None`` reads
            ``GRAPH_ENGINE``.
        workdir: The engine's working directory; ``None`` derives
            :func:`graph_workdir` (``GRAPH_STORAGE_PATH/<engine>``,
            resolved absolute).

    Raises:
        KeyError: If ``engine_name`` names no registered engine.
    """
    self._engine_name = engine_name if engine_name is not None else get_settings().GRAPH_ENGINE
    self._engine = get_graph_engine(self._engine_name)
    self._workdir = workdir if workdir is not None else graph_workdir(self._engine_name)
    self._stack: ExitStack | None = None
    self._session: GraphSession | None = None
    self._open_lock = threading.Lock()
    self._build_lock = threading.Lock()
    self._heal_interrupted_swap()

build(batches, *, verbose=0, prune_removed=True)

Upsert a rendered corpus into the graph, one build at a time.

Parameters:

Name	Type	Description	Default
batches	Sequence[MessageBatch]	Parsed source files.	required
verbose	int	Validated console verbosity (0–2).	0
prune_removed	bool	Whether batches render the whole corpus (see :meth:varagity.graph.base.GraphSession.build).	True

Returns:

Type	Description
BuildReport	What the build did.

Raises:

Type	Description
GraphBuildInProgress	If another build or delete is running.
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def build(
    self,
    batches: Sequence[MessageBatch],
    *,
    verbose: int = 0,
    prune_removed: bool = True,
) -> BuildReport:
    """Upsert a rendered corpus into the graph, one build at a time.

    Args:
        batches: Parsed source files.
        verbose: Validated console verbosity (0–2).
        prune_removed: Whether ``batches`` render the whole corpus (see
            :meth:`varagity.graph.base.GraphSession.build`).

    Returns:
        What the build did.

    Raises:
        GraphBuildInProgress: If another build or delete is running.
        GraphUnavailable: If the session cannot be opened.
    """
    with self.writing():
        return self.session().build(batches, verbose=verbose, prune_removed=prune_removed)

close()

Tear the session down (API lifespan shutdown); idempotent.

A teardown that raised would take an orderly shutdown with it, so failures are logged and swallowed — the session is dropped either way, and the next call opens a fresh one. The teardown runs under the open lock, so a concurrent first caller waits for it instead of opening a second session over files the first is still flushing.

Source code in varagity/graph/service.py

def close(self) -> None:
    """Tear the session down (API lifespan shutdown); idempotent.

    A teardown that raised would take an orderly shutdown with it, so
    failures are logged and swallowed — the session is dropped either
    way, and the next call opens a fresh one. The teardown runs under
    the open lock, so a concurrent first caller waits for it instead of
    opening a second session over files the first is still flushing.
    """
    with self._open_lock:
        self._close_locked()

delete_documents(doc_keys)

Retract documents from the graph.

Parameters:

Name	Type	Description	Default
doc_keys	Sequence[str]	Transcript document keys to remove.	required

Returns:

Type	Description
int	How many deletions the engine accepted.

Raises:

Type	Description
GraphBuildInProgress	If a build is running — deletion rebuilds the entities a document partly supported, which may not race extraction.
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def delete_documents(self, doc_keys: Sequence[str]) -> int:
    """Retract documents from the graph.

    Args:
        doc_keys: Transcript document keys to remove.

    Returns:
        How many deletions the engine accepted.

    Raises:
        GraphBuildInProgress: If a build is running — deletion rebuilds
            the entities a document partly supported, which may not race
            extraction.
        GraphUnavailable: If the session cannot be opened.
    """
    with self.writing():
        return self.session().delete_documents(doc_keys)

document_statuses()

Count the engine's documents by processing status.

Returns:

Type	Description
dict[str, int]	Status name → document count, empty when the engine would not say.

Raises:

Type	Description
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def document_statuses(self) -> dict[str, int]:
    """Count the engine's documents by processing status.

    Returns:
        Status name → document count, empty when the engine would not say.

    Raises:
        GraphUnavailable: If the session cannot be opened.
    """
    return self.session().document_statuses()

export(label='*', *, max_depth=3, max_nodes=5000)

Read a renderable slice of the graph.

Parameters:

Name	Type	Description	Default
label	str	Entity name to centre on; "*" is the whole graph.	'*'
max_depth	int	Hops to walk out from label.	3
max_nodes	int	Node cap.	5000

Returns:

Type	Description
GraphExport	The slice.

Raises:

Type	Description
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def export(
    self,
    label: str = "*",
    *,
    max_depth: int = 3,
    max_nodes: int = 5000,
) -> GraphExport:
    """Read a renderable slice of the graph.

    Args:
        label: Entity name to centre on; ``"*"`` is the whole graph.
        max_depth: Hops to walk out from ``label``.
        max_nodes: Node cap.

    Returns:
        The slice.

    Raises:
        GraphUnavailable: If the session cannot be opened.
    """
    return self.session().export(label, max_depth=max_depth, max_nodes=max_nodes)

query(question, *, mode=None, verbose=0)

Answer one question from the graph — never blocked by a build.

Parameters:

Name	Type	Description	Default
question	str	The question, verbatim.	required
mode	str | None	Engine query mode; None uses the adapter's primary.	None
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
GraphAnswer	The answer with its evidence.

Raises:

Type	Description
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def query(self, question: str, *, mode: str | None = None, verbose: int = 0) -> GraphAnswer:
    """Answer one question from the graph — never blocked by a build.

    Args:
        question: The question, verbatim.
        mode: Engine query mode; ``None`` uses the adapter's primary.
        verbose: Validated console verbosity (0–2).

    Returns:
        The answer with its evidence.

    Raises:
        GraphUnavailable: If the session cannot be opened.
    """
    return self.session().query(question, mode=mode, verbose=verbose)

replace_workdir(staged, run_id, *, on_open=None)

Swap a restored working directory in for the live one (a graph restore, ADR-019).

_reset_workdir's close → change → reopen move, made safe to lose. Under the write lock throughout (no build, resume, delete or export can start), and under the open lock for the renames (no lazy open — a status poll, a /graph read — can land on a half-swapped folder):

1. close the session — the engine flushes what it buffers into the old files, and its process-global state is reset so the next open reads the new ones from disk;
2. move the live workdir aside to <workdir>.previous-<run> (when there is one), then staged into its place;
3. open the restored graph and read its size — opening is what checks the files (LightRAG asserts its vectors' embedding_dim);
4. rename the previous graph to <workdir>.replaced-<run> — the commit point: until it, any failure (or a crash, which the next process's construction heals) brings the previous graph back; after it, the restored graph stands and the replaced one is deleted.

A failure before the commit undoes the swap: the restored folder goes back to staged (its owner removes it), the previous graph returns and its session is reopened, and the error is re-raised.

Reads stay unlocked outside the renames, so a graph turn already streaming when the session closes can end in an error (chat refuses new graph turns for the whole restore, before they open one).

Parameters:

Name	Type	Description	Default
staged	Path	The restored workdir, complete — at :meth:staging_dir, so the renames stay on one filesystem.	required
run_id	str	The restore run's id (names the folders beside the workdir).	required
on_open	Callable[[], None] | None	Called once the restored folder is in place, just before it is opened — the restore's open stage.	None

Returns:

Type	Description
GraphStats	The restored graph's size, as its freshly opened session reports
GraphStats	it.

Raises:

Type	Description
GraphBuildInProgress	If a write holds the write lock (nothing was moved).
GraphUnavailable	If the restored graph would not open (the previous one is back in place).
OSError	If a rename fails (whatever had moved is moved back first).

Source code in varagity/graph/service.py

def replace_workdir(
    self,
    staged: Path,
    run_id: str,
    *,
    on_open: Callable[[], None] | None = None,
) -> GraphStats:
    """Swap a restored working directory in for the live one (a graph restore, ADR-019).

    ``_reset_workdir``'s close → change → reopen move, made safe to lose.
    Under the **write lock** throughout (no build, resume, delete or
    export can start), and under the **open lock** for the renames (no
    lazy open — a status poll, a ``/graph`` read — can land on a
    half-swapped folder):

    1. close the session — the engine flushes what it buffers into the
       *old* files, and its process-global state is reset so the next
       open reads the new ones from disk;
    2. move the live workdir aside to ``<workdir>.previous-<run>`` (when
       there is one), then ``staged`` into its place;
    3. open the restored graph and read its size — opening is what
       checks the files (LightRAG asserts its vectors' ``embedding_dim``);
    4. rename the previous graph to ``<workdir>.replaced-<run>`` — the
       **commit point**: until it, any failure (or a crash, which the
       next process's construction heals) brings the previous graph
       back; after it, the restored graph stands and the replaced one is
       deleted.

    A failure before the commit undoes the swap: the restored folder
    goes back to ``staged`` (its owner removes it), the previous graph
    returns and its session is reopened, and the error is re-raised.

    Reads stay unlocked outside the renames, so a graph turn already
    streaming when the session closes can end in an error (chat refuses
    *new* graph turns for the whole restore, before they open one).

    Args:
        staged: The restored workdir, complete — at :meth:`staging_dir`,
            so the renames stay on one filesystem.
        run_id: The restore run's id (names the folders beside the
            workdir).
        on_open: Called once the restored folder is in place, just before
            it is opened — the restore's ``open`` stage.

    Returns:
        The restored graph's size, as its freshly opened session reports
        it.

    Raises:
        GraphBuildInProgress: If a write holds the write lock (nothing
            was moved).
        GraphUnavailable: If the restored graph would not open (the
            previous one is back in place).
        OSError: If a rename fails (whatever had moved is moved back
            first).
    """
    previous = self._swap_dir(_PREVIOUS, run_id)
    replaced = self._swap_dir(_REPLACED, run_id)
    with self.writing():
        had_previous = self._swap_in(staged, previous)
        try:
            if on_open is not None:
                on_open()
            stats = self.session().stats()
            if had_previous:
                os.replace(previous, replaced)  # the commit point
        except Exception:
            self._swap_back(staged, previous, had_previous=had_previous)
            raise
        fsync_directory(self._workdir.parent)
        shutil.rmtree(replaced, ignore_errors=True)
        if replaced.exists(follow_symlinks=False):
            logger.warning(
                "could not delete the replaced graph (%s) — the next API start removes it",
                replaced.name,
            )
    logger.info("swapped a restored graph into %s", self._workdir.name)
    return stats

resume(*, verbose=0)

Finish whatever a killed build left in flight.

Parameters:

Name	Type	Description	Default
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
BuildReport	What the pass did.

Raises:

Type	Description
GraphBuildInProgress	If another build or delete is running.
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def resume(self, *, verbose: int = 0) -> BuildReport:
    """Finish whatever a killed build left in flight.

    Args:
        verbose: Validated console verbosity (0–2).

    Returns:
        What the pass did.

    Raises:
        GraphBuildInProgress: If another build or delete is running.
        GraphUnavailable: If the session cannot be opened.
    """
    with self.writing():
        return self.session().resume(verbose=verbose)

retrieve(question, *, mode=None, verbose=0)

Find evidence for one question — never blocked by a build.

The chat path's read: the API streams its own grounded answer over what this returns, so no engine answer call is spent here.

Parameters:

Name	Type	Description	Default
question	str	The search query, verbatim.	required
mode	str | None	Engine query mode; None uses the adapter's primary.	None
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
GraphRetrieval	The evidence and transcript excerpts.

Raises:

Type	Description
GraphUnavailable	If the session cannot be opened (the caller turns that into ADR-017's per-turn degrade).

Source code in varagity/graph/service.py

def retrieve(
    self, question: str, *, mode: str | None = None, verbose: int = 0
) -> GraphRetrieval:
    """Find evidence for one question — never blocked by a build.

    The chat path's read: the API streams its own grounded answer over
    what this returns, so no engine answer call is spent here.

    Args:
        question: The search query, verbatim.
        mode: Engine query mode; ``None`` uses the adapter's primary.
        verbose: Validated console verbosity (0–2).

    Returns:
        The evidence and transcript excerpts.

    Raises:
        GraphUnavailable: If the session cannot be opened (the caller
            turns that into ADR-017's per-turn degrade).
    """
    return self.session().retrieve(question, mode=mode, verbose=verbose)

session()

Return the open session, opening it on first use.

Returns:

Type	Description
GraphSession	The process's one session.

Raises:

Type	Description
GraphUnavailable	If the engine's session cannot be opened. The failure is not cached — the next call tries again, so a transient one (a model service still warming up) does not need a restart to clear.

Source code in varagity/graph/service.py

def session(self) -> GraphSession:
    """Return the open session, opening it on first use.

    Returns:
        The process's one session.

    Raises:
        GraphUnavailable: If the engine's session cannot be opened. The
            failure is not cached — the next call tries again, so a
            transient one (a model service still warming up) does not
            need a restart to clear.
    """
    with self._open_lock:
        if self._session is not None:
            return self._session
        logger.info("opening the %s graph session in %s", self._engine_name, self._workdir)
        stack = ExitStack()
        try:
            session = stack.enter_context(self._engine.session(self._workdir))
        except Exception as exc:
            stack.close()
            raise GraphUnavailable(
                f"could not open the {self._engine_name!r} graph session "
                f"in {self._workdir}: {exc}"
            ) from exc
        self._stack, self._session = stack, session
        return session

staging_dir(run_id)

Name the folder a restore extracts a bundle into before the swap.

<workdir>.restore-<run>: beside the workdir, so on the same filesystem — every rename of :meth:replace_workdir stays atomic. Nothing is created here.

Parameters:

Name	Type	Description	Default
run_id	str	The restore run's id.	required

Returns:

Type	Description
Path	The staging folder's path.

Raises:

Type	Description
ValueError	If run_id isn't a safe folder-name suffix.

Source code in varagity/graph/service.py

def staging_dir(self, run_id: str) -> Path:
    """Name the folder a restore extracts a bundle into before the swap.

    ``<workdir>.restore-<run>``: beside the workdir, so on the same
    filesystem — every rename of :meth:`replace_workdir` stays atomic.
    Nothing is created here.

    Args:
        run_id: The restore run's id.

    Returns:
        The staging folder's path.

    Raises:
        ValueError: If ``run_id`` isn't a safe folder-name suffix.
    """
    return self._swap_dir(_STAGED, run_id)

stats()

Report the graph's size.

Returns:

Type	Description
GraphStats	Entity/relation/community counts, each None where unknown.

Raises:

Type	Description
GraphUnavailable	If the session cannot be opened.

Source code in varagity/graph/service.py

def stats(self) -> GraphStats:
    """Report the graph's size.

    Returns:
        Entity/relation/community counts, each ``None`` where unknown.

    Raises:
        GraphUnavailable: If the session cannot be opened.
    """
    return self.session().stats()

storage_format()

Name the format the engine's workdir files are in (what a backup records).

Returns:

Type	Description
EngineFormat	The engine's :meth:~varagity.graph.base.GraphEngine.storage_format
EngineFormat	— read from package metadata, so this opens no session.

Source code in varagity/graph/service.py

def storage_format(self) -> EngineFormat:
    """Name the format the engine's workdir files are in (what a backup records).

    Returns:
        The engine's :meth:`~varagity.graph.base.GraphEngine.storage_format`
        — read from package metadata, so this opens no session.
    """
    return self._engine.storage_format()

writing()

Take the single-flight write lock for the duration of a block.

The lock :meth:build, :meth:resume and :meth:delete_documents take, public so a caller that must see the workdir hold still can take it too — a backup export copies the files under it (ADR-019), so no build, resume or delete can start mid-copy, while reads stay unlocked and chat keeps answering (:meth:replace_workdir swaps a restored workdir in under it). Never blocks: a held lock is an answer (the caller's 409), not a wait.

Returns:

Type	Description
_WriteLock	A context manager holding the lock (released on block exit,
_WriteLock	however the block ends).

Raises:

Type	Description
GraphBuildInProgress	If it is already held.

Source code in varagity/graph/service.py

def writing(self) -> "_WriteLock":
    """Take the single-flight write lock for the duration of a block.

    The lock :meth:`build`, :meth:`resume` and :meth:`delete_documents`
    take, public so a caller that must see the workdir hold still can
    take it too — a backup export copies the files under it (ADR-019),
    so no build, resume or delete can start mid-copy, while reads stay
    unlocked and chat keeps answering (:meth:`replace_workdir` swaps a
    restored workdir in under it). Never blocks: a held lock is an
    answer (the caller's ``409``), not a wait.

    Returns:
        A context manager holding the lock (released on block exit,
        however the block ends).

    Raises:
        GraphBuildInProgress: If it is already held.
    """
    if not self._build_lock.acquire(blocking=False):
        raise GraphBuildInProgress(
            "the graph is being written — a build, a delete, or a backup export or restore "
            "holds its write lock"
        )
    return _WriteLock(self._build_lock)

GraphUnavailable

Bases: RuntimeError

Raised when the graph engine's session cannot be opened.

Source code in varagity/graph/service.py

class GraphUnavailable(RuntimeError):
    """Raised when the graph engine's session cannot be opened."""

get_graph_service()

Provide the process-wide graph service (FastAPI dependency, override seam).

Returns:

Type	Description
GraphService	The lazily created singleton. Construction only resolves settings and
GraphService	the registry — the engine session opens on first real use.

Source code in varagity/graph/service.py

def get_graph_service() -> GraphService:
    """Provide the process-wide graph service (FastAPI dependency, override seam).

    Returns:
        The lazily created singleton. Construction only resolves settings and
        the registry — the engine session opens on first real use.
    """
    global _default_service
    with _default_service_lock:
        if _default_service is None:
            _default_service = GraphService()
        return _default_service

graph_workdir(engine_name=None)

Locate the working directory an engine stores this graph in.

The one derivation of "where the graph lives", shared by the service and by anything that must read the workdir without going through the engine — the Prometheus gauges, which resolve a path at scrape time and must not depend on the engine registry being warm (:mod:varagity.observability.graph).

Parameters:

Name	Type	Description	Default
engine_name	str | None	Registry name of the engine; None reads GRAPH_ENGINE.	None

Returns:

Type	Description
Path	The absolute directory (relative workdirs have bitten this repo
Path	before — an engine consumes the path verbatim, from whatever cwd it
Path	happens to run in). It may not exist yet: nothing is created here.

Source code in varagity/graph/service.py

def graph_workdir(engine_name: str | None = None) -> Path:
    """Locate the working directory an engine stores this graph in.

    The one derivation of "where the graph lives", shared by the service
    and by anything that must read the workdir *without* going through the
    engine — the Prometheus gauges, which resolve a path at scrape time and
    must not depend on the engine registry being warm
    (:mod:`varagity.observability.graph`).

    Args:
        engine_name: Registry name of the engine; ``None`` reads
            ``GRAPH_ENGINE``.

    Returns:
        The absolute directory (relative workdirs have bitten this repo
        before — an engine consumes the path verbatim, from whatever cwd it
        happens to run in). It may not exist yet: nothing is created here.
    """
    settings = get_settings()
    name = engine_name if engine_name is not None else settings.GRAPH_ENGINE
    return (Path(settings.GRAPH_STORAGE_PATH) / name).resolve()

Engine adapter

varagity.graph.engines.lightrag

The LightRAG adapter — the shipped graph engine (spec_graphrag §8; ADR-017).

LightRAG is the document-shaped engine with the cleanest injection points: a custom async llm_model_func and an EmbeddingFunc(embedding_dim=1024) are first-class constructor arguments, so this is an engine where the <think>-strip template applies fully — every extraction, gleaning, and answer completion passes through :func:~varagity.models.llm.clean_response before LightRAG parses it. A reasoning model's unstripped <think> block breaks its delimiter grammar silently, which is exactly the trap the condense and HyDE features already paid for.

Storage is the file-based default (NetworkXStorage + NanoVectorDBStorage + JSON KV/doc-status) inside the session's working directory (ADR-017 pinned all four classes): LightRAG's Postgres graph plane needs the Apache AGE extension, which is a different image than this stack runs.

One process, one writer, queries live throughout. The session drives a private event loop on a daemon thread and submits every engine call with asyncio.run_coroutine_threadsafe, so a chat query is answered while a multi-day backfill is still extracting — which is how LightRAG's own server behaves, and why :class:~varagity.graph.service.GraphService keeps exactly one session per process (stage-2 decisions #10/#11). The engine's storages are single-writer per workspace by explicit invariant, so nothing outside this process may write the workdir; that is the reason there is no CLI graph build.

Builds are diffing upserts, not appends. LightRAG's enqueue stage drops a re-submitted doc_id in any status, so a re-exported chat.db whose newest thread-day grew would silently keep the stale transcript. :mod:varagity.graph.manifest records what was indexed; :meth:_LightRAGSession.build deletes the genuinely changed documents before re-enqueueing them and pays extraction for nothing else. Enqueue and process are separate engine calls, and process re-selects every in-flight/failed document at the top of each batch — so a killed build resumes by calling :meth:_LightRAGSession.resume.

Two more things this adapter owns, both recorded rather than hidden (spec §6):

• e5's asymmetric prefix rides GRAPH_QUERY_PREFIX. The bake-off ran unprefixed (a recorded stage-1 deviation) because LightRAG appeared to call one embedding function for passages and queries alike. It does not: EmbeddingFunc(supports_asymmetric=True) forwards a context="document"|"query" keyword to the hook (lightrag/utils.py:596-602), and the pinned vector store passes it on every search (kg/nano_vector_db_impl.py:429-431; also operate.py:4382 for the dual-level keyword embeddings). Passages stay unprefixed either way — which is correct under e5 discipline — so the setting was measurable on the already-built bake-off graph without re-embedding anything, and the acceptance gate shipped it on. :func:varagity.models.embeddings.format_query stays the single owner of the query format.
• Concurrency knobs are pinned through os.environ. They are read at import time by LightRAG's own settings, so they must be set before the lazy import — the one place in this repo where writing environment variables is the mechanism rather than a smell (varagity's own config is still read only through :func:~varagity.config.get_settings).

Answer composition is ours, not the engine's (ADR-017's retrieval-only decision). :meth:_LightRAGSession.retrieve is that half on its own: one aquery_data call returning structured entities/relations/chunks with no LLM answer stage. The app streams its answer over it (:func:varagity.pipeline.graph_flow.graph_query_stream_flow), and a <base>+synthesis mode composes the same retrieval with :mod:varagity.graph.answer's non-streaming writer — which is what the eval graph harness scores, so the harness keeps guarding the shipped path. Unsuffixed modes keep the bake-off's engine-composed path verbatim, so ADR-017's numbers stay reproducible.

LightRAGEngine

LightRAG behind the :class:~varagity.graph.base.GraphEngine protocol.

Source code in varagity/graph/engines/lightrag.py

@register(ENGINE_NAME)
class LightRAGEngine:
    """LightRAG behind the :class:`~varagity.graph.base.GraphEngine` protocol."""

    def storage_format(self) -> EngineFormat:
        """Name the file format this engine's workdir is in: the installed ``lightrag-hku``.

        Every file in the workdir is LightRAG's own JSON/GraphML (plus the
        repo's two sidecars), so the distribution's version *is* the
        format's version. Read through :mod:`importlib.metadata` — the
        registry must stay free of the engine library, which only
        :meth:`session` may import (stage-1 decision #8).

        Returns:
            ``lightrag`` / ``lightrag-hku`` / the installed version.

        Raises:
            importlib.metadata.PackageNotFoundError: If ``lightrag-hku`` is
                not installed (it is a main dependency, so only a broken
                environment gets here — where no session could open either).
        """
        return EngineFormat(
            name=ENGINE_NAME,
            package=ENGINE_PACKAGE,
            version=importlib.metadata.version(ENGINE_PACKAGE),
        )

    @contextmanager
    def session(self, workdir: Path) -> Iterator[GraphSession]:
        """Open a LightRAG session storing everything under ``workdir``.

        Args:
            workdir: The engine's working directory (created if absent).

        Yields:
            The open session; storages are initialized on entry and finalized
            on exit.
        """
        settings = get_settings()
        _pin_env()
        # Lazy even though lightrag is a main dependency now (stage-1
        # decision #8): the registry stays free to import, so unit tests and
        # CI never pay for the engine — numpy included, since LightRAG's
        # EmbeddingFunc expects an array and numpy is not a declared
        # dependency of this project, only a transitive one.
        import numpy as np
        from lightrag import LightRAG, QueryParam
        from lightrag.kg.shared_storage import finalize_share_data, initialize_pipeline_status
        from lightrag.utils import EmbeddingFunc

        workdir.mkdir(parents=True, exist_ok=True)
        chat_client = AsyncOpenAI(
            base_url=settings.BASE_MODEL_API_URL,
            api_key=settings.BASE_MODEL_API_KEY,
            timeout=_LLM_TIMEOUT_S,
        )
        embedding_client = AsyncOpenAI(
            base_url=settings.EMBEDDING_API_URL,
            api_key=settings.EMBEDDING_API_KEY,
            timeout=_EMBEDDING_TIMEOUT_S,
        )
        asymmetric = settings.GRAPH_QUERY_PREFIX
        if asymmetric:
            logger.info(
                "GRAPH_QUERY_PREFIX on: queries are e5-instruction-wrapped "
                "(supports_asymmetric=True); passages stay unprefixed"
            )
        embed = make_embedding_func(
            embedding_client, model=settings.EMBEDDING_MODEL, query_prefix=asymmetric
        )

        async def embedding_func(texts: Sequence[str], context: str | None = None) -> Any:
            return np.array(await embed(texts, context=context), dtype=np.float32)

        rag = LightRAG(
            working_dir=str(workdir),
            llm_model_func=make_llm_func(
                chat_client,
                model=settings.BASE_MODEL,
                temperature=settings.LLM_TEMPERATURE,
                context_tokens=settings.LLM_CONTEXT_TOKENS,
            ),
            embedding_func=EmbeddingFunc(
                embedding_dim=_EMBEDDING_DIM,
                func=embedding_func,
                # Without this LightRAG strips `context` before calling the
                # hook above, and the prefix setting is a dead knob.
                supports_asymmetric=asymmetric,
            ),
        )
        session = _LightRAGSession(
            rag, QueryParam, workdir=workdir, share_finalizer=finalize_share_data
        )
        try:
            session.run(rag.initialize_storages())
            session.run(initialize_pipeline_status())
            yield session
        finally:
            session.close()

session(workdir)

Open a LightRAG session storing everything under workdir.

Parameters:

Name	Type	Description	Default
workdir	Path	The engine's working directory (created if absent).	required

Yields:

Type	Description
GraphSession	The open session; storages are initialized on entry and finalized
GraphSession	on exit.

Source code in varagity/graph/engines/lightrag.py

@contextmanager
def session(self, workdir: Path) -> Iterator[GraphSession]:
    """Open a LightRAG session storing everything under ``workdir``.

    Args:
        workdir: The engine's working directory (created if absent).

    Yields:
        The open session; storages are initialized on entry and finalized
        on exit.
    """
    settings = get_settings()
    _pin_env()
    # Lazy even though lightrag is a main dependency now (stage-1
    # decision #8): the registry stays free to import, so unit tests and
    # CI never pay for the engine — numpy included, since LightRAG's
    # EmbeddingFunc expects an array and numpy is not a declared
    # dependency of this project, only a transitive one.
    import numpy as np
    from lightrag import LightRAG, QueryParam
    from lightrag.kg.shared_storage import finalize_share_data, initialize_pipeline_status
    from lightrag.utils import EmbeddingFunc

    workdir.mkdir(parents=True, exist_ok=True)
    chat_client = AsyncOpenAI(
        base_url=settings.BASE_MODEL_API_URL,
        api_key=settings.BASE_MODEL_API_KEY,
        timeout=_LLM_TIMEOUT_S,
    )
    embedding_client = AsyncOpenAI(
        base_url=settings.EMBEDDING_API_URL,
        api_key=settings.EMBEDDING_API_KEY,
        timeout=_EMBEDDING_TIMEOUT_S,
    )
    asymmetric = settings.GRAPH_QUERY_PREFIX
    if asymmetric:
        logger.info(
            "GRAPH_QUERY_PREFIX on: queries are e5-instruction-wrapped "
            "(supports_asymmetric=True); passages stay unprefixed"
        )
    embed = make_embedding_func(
        embedding_client, model=settings.EMBEDDING_MODEL, query_prefix=asymmetric
    )

    async def embedding_func(texts: Sequence[str], context: str | None = None) -> Any:
        return np.array(await embed(texts, context=context), dtype=np.float32)

    rag = LightRAG(
        working_dir=str(workdir),
        llm_model_func=make_llm_func(
            chat_client,
            model=settings.BASE_MODEL,
            temperature=settings.LLM_TEMPERATURE,
            context_tokens=settings.LLM_CONTEXT_TOKENS,
        ),
        embedding_func=EmbeddingFunc(
            embedding_dim=_EMBEDDING_DIM,
            func=embedding_func,
            # Without this LightRAG strips `context` before calling the
            # hook above, and the prefix setting is a dead knob.
            supports_asymmetric=asymmetric,
        ),
    )
    session = _LightRAGSession(
        rag, QueryParam, workdir=workdir, share_finalizer=finalize_share_data
    )
    try:
        session.run(rag.initialize_storages())
        session.run(initialize_pipeline_status())
        yield session
    finally:
        session.close()

storage_format()

Name the file format this engine's workdir is in: the installed lightrag-hku.

Every file in the workdir is LightRAG's own JSON/GraphML (plus the repo's two sidecars), so the distribution's version is the format's version. Read through :mod:importlib.metadata — the registry must stay free of the engine library, which only :meth:session may import (stage-1 decision #8).

Returns:

Type	Description
EngineFormat	lightrag / lightrag-hku / the installed version.

Raises:

Type	Description
PackageNotFoundError	If lightrag-hku is not installed (it is a main dependency, so only a broken environment gets here — where no session could open either).

Source code in varagity/graph/engines/lightrag.py

def storage_format(self) -> EngineFormat:
    """Name the file format this engine's workdir is in: the installed ``lightrag-hku``.

    Every file in the workdir is LightRAG's own JSON/GraphML (plus the
    repo's two sidecars), so the distribution's version *is* the
    format's version. Read through :mod:`importlib.metadata` — the
    registry must stay free of the engine library, which only
    :meth:`session` may import (stage-1 decision #8).

    Returns:
        ``lightrag`` / ``lightrag-hku`` / the installed version.

    Raises:
        importlib.metadata.PackageNotFoundError: If ``lightrag-hku`` is
            not installed (it is a main dependency, so only a broken
            environment gets here — where no session could open either).
    """
    return EngineFormat(
        name=ENGINE_NAME,
        package=ENGINE_PACKAGE,
        version=importlib.metadata.version(ENGINE_PACKAGE),
    )

evidence_from_context(context, index)

Normalize LightRAG's retrieval context into engine-independent evidence.

LightRAG returns its context as JSON (or as a formatted string, depending on version and mode), so the mapping is alias-tolerant and every unknown shape degrades to raw plus empty lists rather than raising — a normalizer that throws would turn an answered question into a harness crash.

Parameters:

Name	Type	Description	Default
context	Any	Whatever aquery(..., only_need_context=True) returned.	required
index	Mapping[str, Sequence[str]]	doc_key → message guids, for provenance recovery.	required

Returns:

Type	Description
GraphEvidence	The normalized evidence. LightRAG has no community layer, so
GraphEvidence	communities is always empty.

Source code in varagity/graph/engines/lightrag.py

def evidence_from_context(context: Any, index: Mapping[str, Sequence[str]]) -> GraphEvidence:
    """Normalize LightRAG's retrieval context into engine-independent evidence.

    LightRAG returns its context as JSON (or as a formatted string, depending
    on version and mode), so the mapping is alias-tolerant and every unknown
    shape degrades to ``raw`` plus empty lists rather than raising — a
    normalizer that throws would turn an answered question into a harness
    crash.

    Args:
        context: Whatever ``aquery(..., only_need_context=True)`` returned.
        index: ``doc_key`` → message guids, for provenance recovery.

    Returns:
        The normalized evidence. LightRAG has no community layer, so
        ``communities`` is always empty.
    """
    raw = _as_mapping(context)
    walked = context if raw is None else raw
    return GraphEvidence(
        entities=[
            entity for item in _section(raw, "entities") if (entity := _entity(item)) is not None
        ],
        relations=[
            relation
            for item in _section(raw, "relationships", "relations")
            if (relation := _relation(item)) is not None
        ],
        communities=[],
        message_guids=guids_in_payload(walked, index),
        raw=raw,
    )

export_from_knowledge_graph(graph)

Flatten LightRAG's KnowledgeGraph into the wire-ready export.

The engine's own shape keeps everything it knows in a free-form properties dict and reports no degree at all, so this projects the two fields the view actually draws with (type, description), unpacks the file_path provenance the drill-down joins on (:func:_doc_keys), and counts each node's edges within the returned slice — an export is capped and degree-ordered, so a slice-local degree is the honest number for the picture being drawn.

Parameters:

Name	Type	Description	Default
graph	Any	Whatever get_knowledge_graph returned (or None).	required

Returns:

Type	Description
GraphExport	The export. An unknown shape degrades to an empty one rather than
GraphExport	raising — the same rule the evidence normalizers follow.

Source code in varagity/graph/engines/lightrag.py

def export_from_knowledge_graph(graph: Any) -> GraphExport:
    """Flatten LightRAG's ``KnowledgeGraph`` into the wire-ready export.

    The engine's own shape keeps everything it knows in a free-form
    ``properties`` dict and reports no degree at all, so this projects the
    two fields the view actually draws with (type, description), unpacks the
    ``file_path`` provenance the drill-down joins on
    (:func:`_doc_keys`), and counts each node's edges **within the returned
    slice** — an export is capped and degree-ordered, so a slice-local degree
    is the honest number for the picture being drawn.

    Args:
        graph: Whatever ``get_knowledge_graph`` returned (or ``None``).

    Returns:
        The export. An unknown shape degrades to an empty one rather than
        raising — the same rule the evidence normalizers follow.
    """
    nodes = _attr(graph, "nodes") or []
    edges = _attr(graph, "edges") or []
    exported_edges: list[GraphExportEdge] = []
    degrees: dict[str, int] = {}
    for edge in edges:
        source, target = _attr(edge, "source"), _attr(edge, "target")
        if not isinstance(source, str) or not isinstance(target, str):
            continue
        properties = _attr(edge, "properties")
        properties = properties if isinstance(properties, Mapping) else {}
        exported_edges.append(
            GraphExportEdge(
                id=str(_attr(edge, "id") or f"{source}-{target}"),
                source=source,
                target=target,
                label=_text(properties, "keywords", "label", "relation"),
                description=_text(properties, "description", "summary", "content"),
            )
        )
        degrees[source] = degrees.get(source, 0) + 1
        degrees[target] = degrees.get(target, 0) + 1
    exported_nodes: list[GraphExportNode] = []
    for node in nodes:
        identifier = _attr(node, "id")
        if not isinstance(identifier, str) or not identifier:
            continue
        properties = _attr(node, "properties")
        properties = properties if isinstance(properties, Mapping) else {}
        exported_nodes.append(
            GraphExportNode(
                id=identifier,
                entity_type=_text(properties, "entity_type", "type", "category"),
                description=_text(properties, "description", "summary", "content"),
                degree=degrees.get(identifier, 0),
                doc_keys=_doc_keys(properties),
            )
        )
    return GraphExport(
        nodes=exported_nodes,
        edges=exported_edges,
        truncated=bool(_attr(graph, "is_truncated")),
    )

fit_max_tokens(messages, cap, context_tokens)

Clamp a generation cap so prompt + generation fits the context window.

The same discipline :func:varagity.models.llm._fit_max_tokens applies to the app's own client, re-implemented here because LightRAG is handed a raw async callable rather than an :class:~varagity.models.llm.LLMClient: llama.cpp with context shift disabled hard-500s mid-decode at the window instead of stopping gracefully, and LightRAG's extraction prompts are large.

Parameters:

Name	Type	Description	Default
messages	Sequence[Mapping[str, Any]]	The chat messages about to be sent (string contents are counted).	required
cap	int	The requested generation cap.	required
context_tokens	int	The served model's context window (LLM_CONTEXT_TOKENS).	required

Returns:

Name	Type	Description
	int	The cap, reduced to fit. A prompt that leaves no room at all still
gets	int	data:_MIN_TOKENS: that request is already lost, and a floor
	int	keeps the failure legible in the engine's own error rather than in a
	int	zero-token request. A caller asking for less than the floor still
	int	gets what it asked for.

Source code in varagity/graph/engines/lightrag.py

def fit_max_tokens(messages: Sequence[Mapping[str, Any]], cap: int, context_tokens: int) -> int:
    """Clamp a generation cap so prompt + generation fits the context window.

    The same discipline :func:`varagity.models.llm._fit_max_tokens` applies to
    the app's own client, re-implemented here because LightRAG is handed a raw
    async callable rather than an :class:`~varagity.models.llm.LLMClient`:
    llama.cpp with context shift disabled hard-500s mid-decode at the window
    instead of stopping gracefully, and LightRAG's extraction prompts are
    large.

    Args:
        messages: The chat messages about to be sent (string contents are
            counted).
        cap: The requested generation cap.
        context_tokens: The served model's context window
            (``LLM_CONTEXT_TOKENS``).

    Returns:
        The cap, reduced to fit. A prompt that leaves no room at all still
        gets :data:`_MIN_TOKENS`: that request is already lost, and a floor
        keeps the failure legible in the engine's own error rather than in a
        zero-token request. A caller asking for less than the floor still
        gets what it asked for.
    """
    used = sum(
        count_tokens(content)
        for message in messages
        if isinstance(content := message.get("content"), str)
    )
    available = context_tokens - used - _CTX_HEADROOM_TOKENS
    if available < cap:
        logger.debug("clamping LightRAG generation cap %d → %d (prompt ~%d)", cap, available, used)
    return min(cap, max(_MIN_TOKENS, available))

make_embedding_func(client, *, model, query_prefix=False)

Build the async embedding callable LightRAG embeds documents and queries with.

Passages are always sent unprefixed, which is what e5 discipline requires of documents. Queries are wrapped with :func:~varagity.models.embeddings.format_query only when query_prefix is on and the caller declared the call's side — which LightRAG does through the context keyword its EmbeddingFunc(supports_asymmetric=True) wrapper forwards. Both gates matter: with supports_asymmetric off the wrapper strips context before this function ever sees it, so the flag has to reach both.

Parameters:

Name	Type	Description	Default
client	AsyncOpenAI	An async OpenAI-compatible client pointed at infinity.	required
model	str	Served embedding model name.	required
query_prefix	bool	Whether to apply e5's query-side instruction wrapper (GRAPH_QUERY_PREFIX).	False

Returns:

Type	Description
Callable[..., Awaitable[list[list[float]]]]	A coroutine function mapping texts to their 1024-dim vectors, in
Callable[..., Awaitable[list[list[float]]]]	order, accepting LightRAG's optional context keyword.

Source code in varagity/graph/engines/lightrag.py

def make_embedding_func(
    client: AsyncOpenAI, *, model: str, query_prefix: bool = False
) -> Callable[..., Awaitable[list[list[float]]]]:
    """Build the async embedding callable LightRAG embeds documents and queries with.

    Passages are always sent unprefixed, which is what e5 discipline requires
    of documents. Queries are wrapped with
    :func:`~varagity.models.embeddings.format_query` only when
    ``query_prefix`` is on *and* the caller declared the call's side — which
    LightRAG does through the ``context`` keyword its
    ``EmbeddingFunc(supports_asymmetric=True)`` wrapper forwards. Both gates
    matter: with ``supports_asymmetric`` off the wrapper strips ``context``
    before this function ever sees it, so the flag has to reach both.

    Args:
        client: An async OpenAI-compatible client pointed at infinity.
        model: Served embedding model name.
        query_prefix: Whether to apply e5's query-side instruction wrapper
            (``GRAPH_QUERY_PREFIX``).

    Returns:
        A coroutine function mapping texts to their 1024-dim vectors, in
        order, accepting LightRAG's optional ``context`` keyword.
    """

    async def embed(texts: Sequence[str], context: str | None = None) -> list[list[float]]:
        payload = list(texts)
        if query_prefix and context == _QUERY_CONTEXT:
            payload = [format_query(text) for text in payload]
        response = await client.embeddings.create(model=model, input=payload)
        return [item.embedding for item in response.data]

    return embed

make_llm_func(client, *, model, temperature, context_tokens)

Build the async completion callable LightRAG drives every LLM stage with.

Parameters:

Name	Type	Description	Default
client	AsyncOpenAI	An async OpenAI-compatible client pointed at llama.cpp.	required
model	str	Model name to send (llama.cpp echoes it — provenance, not routing).	required
temperature	float	Sampling temperature.	required
context_tokens	int	The served model's context window, for the clamp.	required

Returns:

Type	Description
Callable[..., Awaitable[str]]	A coroutine function with LightRAG's expected signature
Callable[..., Awaitable[str]]	(prompt, optional system_prompt/history_messages, plus
Callable[..., Awaitable[str]]	whatever keyword arguments the calling stage passes) returning
Callable[..., Awaitable[str]]	<think>-stripped text.

Source code in varagity/graph/engines/lightrag.py

def make_llm_func(
    client: AsyncOpenAI,
    *,
    model: str,
    temperature: float,
    context_tokens: int,
) -> Callable[..., Awaitable[str]]:
    """Build the async completion callable LightRAG drives every LLM stage with.

    Args:
        client: An async OpenAI-compatible client pointed at llama.cpp.
        model: Model name to send (llama.cpp echoes it — provenance, not
            routing).
        temperature: Sampling temperature.
        context_tokens: The served model's context window, for the clamp.

    Returns:
        A coroutine function with LightRAG's expected signature
        (``prompt``, optional ``system_prompt``/``history_messages``, plus
        whatever keyword arguments the calling stage passes) returning
        ``<think>``-stripped text.
    """

    async def llm_model_func(
        prompt: str,
        system_prompt: str | None = None,
        history_messages: Sequence[Mapping[str, Any]] | None = None,
        **_kwargs: Any,
    ) -> str:
        messages: list[ChatCompletionMessageParam] = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        # LightRAG's history entries are already OpenAI-shaped dicts; the cast
        # is the same duck-typing concession the chat engines make at the LLM
        # seam (varagity/chat/condense.py).
        messages.extend(
            cast("ChatCompletionMessageParam", dict(turn)) for turn in history_messages or ()
        )
        messages.append({"role": "user", "content": prompt})
        response = await client.chat.completions.create(
            model=model,
            messages=messages,
            max_tokens=fit_max_tokens(messages, _MAX_TOKENS, context_tokens),
            temperature=temperature,
        )
        # Mandatory: LightRAG parses delimiter-grammar records straight out of
        # this string, and an unstripped reasoning stage breaks that silently
        # (the condense/HyDE precedent, spec_graphrag §7).
        return clean_response(response.choices[0].message.content or "")

    return llm_model_func

retrieval_from_query_data(payload, index, names=None)

Normalize an aquery_data payload into evidence plus transcript excerpts.

aquery_data is the structured retrieval API: it forces only_need_context internally and returns {"status", "data": {"entities", "relationships", "chunks", "references"}, "metadata"} with no answer call, which is exactly what ADR-017's retrieval-only design needs. A failure payload carries an empty data, and — like every normalizer here — an unknown shape degrades to empty lists rather than raising.

Parameters:

Name	Type	Description	Default
payload	Any	Whatever aquery_data returned (or None).	required
index	Mapping[str, Sequence[str]]	doc_key → message guids, for provenance recovery.	required
names	Mapping[str, str] | None	doc_key → display name, for labelling chunk-grain hits with the same thread name doc-grain hits render (None degrades to the header-parse fallback inside :func:_excerpt).	None

Returns:

Type	Description
GraphEvidence	The evidence (entities/relations; LightRAG has no community layer)
list[TranscriptExcerpt]	and the retrieved transcript passages, in the engine's own relevance
tuple[GraphEvidence, list[TranscriptExcerpt]]	order — one per transcript document: mix mode reaches the same
tuple[GraphEvidence, list[TranscriptExcerpt]]	document through more than one arm, and the best-ranked hit wins
tuple[GraphEvidence, list[TranscriptExcerpt]]	(:func:_dedupe_excerpts).

Source code in varagity/graph/engines/lightrag.py

def retrieval_from_query_data(
    payload: Any,
    index: Mapping[str, Sequence[str]],
    names: Mapping[str, str] | None = None,
) -> tuple[GraphEvidence, list[TranscriptExcerpt]]:
    """Normalize an ``aquery_data`` payload into evidence plus transcript excerpts.

    ``aquery_data`` is the structured retrieval API: it forces
    ``only_need_context`` internally and returns
    ``{"status", "data": {"entities", "relationships", "chunks",
    "references"}, "metadata"}`` with **no answer call**, which is exactly
    what ADR-017's retrieval-only design needs. A failure payload carries an
    empty ``data``, and — like every normalizer here — an unknown shape
    degrades to empty lists rather than raising.

    Args:
        payload: Whatever ``aquery_data`` returned (or ``None``).
        index: ``doc_key`` → message guids, for provenance recovery.
        names: ``doc_key`` → display name, for labelling chunk-grain hits
            with the same thread name doc-grain hits render (``None``
            degrades to the header-parse fallback inside :func:`_excerpt`).

    Returns:
        The evidence (entities/relations; LightRAG has no community layer)
        and the retrieved transcript passages, in the engine's own relevance
        order — one per transcript document: ``mix`` mode reaches the same
        document through more than one arm, and the best-ranked hit wins
        (:func:`_dedupe_excerpts`).
    """
    raw = _as_mapping(payload)
    data = _data_section(raw)
    evidence = GraphEvidence(
        entities=[
            entity for item in _section(data, "entities") if (entity := _entity(item)) is not None
        ],
        relations=[
            relation
            for item in _section(data, "relationships", "relations")
            if (relation := _relation(item)) is not None
        ],
        communities=[],
        message_guids=guids_in_payload(payload if raw is None else raw, index),
        raw=raw,
    )
    excerpts = _dedupe_excerpts(
        excerpt
        for item in _section(data, "chunks")
        if (excerpt := _excerpt(item, index, names or {})) is not None
    )
    return evidence, excerpts