Observability

The Prometheus metric catalog (spec_v2 §6.2) and its recording helpers. The probe points live in the pipeline flows, the reranked retriever, and the API health probes; the catalog is exposed at GET /metrics and charted by the provisioned Grafana dashboards under observability/.

Two kinds of instrument, and the difference decides which panel can answer which question:

	Source	Survives an API restart?	Sees CLI ingests?
varagity_corpus_* gauges	pgvector, at scrape time	yes	yes
varagity_graph_* gauges	the graph workdir's files, at scrape time	yes	n/a — graph builds are API-only
counters / histograms	this process	no	no (ADR-007)

Ingestion is rare and bursty, so increase()/rate() over an ingest counter reads 0 over any window — the counter's series is born at its full value after a restart, so Prometheus never observes the rise. Corpus size is therefore a gauge question, not a counter question (spec_v3 §6.1; ADR-013 is the full record); tests/unit/test_dashboards.py fails the build if a panel regresses to the counter form.

Both gauge families emit no samples at all rather than zeros when their store has never answered — an unbuilt graph is not a graph with zero entities — and serve the last good snapshot through an outage, because a scrape that raises takes /metrics down including the metrics that are fine.

varagity.observability

In-app observability: the Prometheus metric catalog (spec_v2 §6).

:mod:varagity.observability.metrics holds the spec_v2 §6.2 catalog plus the recording helpers the pipeline probe points call — instruments that record what this process did.

:mod:varagity.observability.corpus adds the complement: gauges read from pgvector at scrape time, answering what is in the store. Process history resets when the API restarts and misses CLI ingests entirely, which is what left the Ingestion dashboard reading zero through v2 (spec_v3 §6.1).

:mod:varagity.observability.graph applies that same shape to the graph corpus, reading the engine's working directory (never the engine) at scrape time — a backfill that ran for hours yesterday is a state question too.

The API's GET /metrics route exposes the same process-wide registry.

varagity.observability.metrics

Prometheus instrumentation: the spec_v2 §6.2 metric catalog + recording helpers.

The catalog registers in prometheus_client's default process-wide registry and is exposed by the API's GET /metrics (spec_v2 §6.1 source 1 — direct in-app instrumentation, the primary metrics source; plan decision #10). Probe points call the observe_*/count_*/set_* helpers below instead of touching the metric objects, so the label vocabulary stays in one place:

• the Prefect flows (:mod:varagity.pipeline.query_flow, :mod:varagity.pipeline.ingest_flow) — per-stage query latency, retrieval scores + rerank deltas, query/ingest counters, LLM tokens;
• the reranked retriever (:mod:varagity.retrieval.reranked) — the rerank stage's latency;
• the API health probes (:mod:varagity.api.deps) — per-dependency reachability.

Metrics accumulate per process: the API (a single uvicorn worker — plan decision #11, no multiprocess registry) is what Prometheus scrapes. A CLI run records into its own short-lived process instead — harmless, and it keeps the instrumented code path identical for both front-ends.

Recording is unconditional; settings.METRICS_ENABLED gates only the /metrics endpoint (the collectors are cheap in-memory counters).

catalog()

Map every metric this app exposes to its label names.

The dashboard guard test (spec_v3 §6.4) checks each panel expression against this, so a renamed metric or an undeclared sum by (…) label fails the unit suite instead of silently emptying a panel.

Names are prometheus_client's internal base names, which are not always what a dashboard queries: the client strips _total from a Counter (varagity_ingest_docs_total is catalogued as varagity_ingest_docs) and appends _bucket/_sum/_count to a Histogram at exposition time. Reconciling those suffixes is the caller's job.

Returns:

Type	Description
dict[str, tuple[str, ...]]	Metric name → declared label names, covering the module-level
dict[str, tuple[str, ...]]	catalog, the store-derived corpus gauges
dict[str, tuple[str, ...]]	(:mod:varagity.observability.corpus), and the workdir-derived
dict[str, tuple[str, ...]]	graph gauges (:mod:varagity.observability.graph).

Source code in varagity/observability/metrics.py

def catalog() -> dict[str, tuple[str, ...]]:
    """Map every metric this app exposes to its label names.

    The dashboard guard test (spec_v3 §6.4) checks each panel expression
    against this, so a renamed metric or an undeclared ``sum by (…)`` label
    fails the unit suite instead of silently emptying a panel.

    Names are ``prometheus_client``'s *internal* base names, which are not
    always what a dashboard queries: the client strips ``_total`` from a
    Counter (``varagity_ingest_docs_total`` is catalogued as
    ``varagity_ingest_docs``) and appends ``_bucket``/``_sum``/``_count``
    to a Histogram at exposition time. Reconciling those suffixes is the
    caller's job.

    Returns:
        Metric name → declared label names, covering the module-level
        catalog, the store-derived corpus gauges
        (:mod:`varagity.observability.corpus`), and the workdir-derived
        graph gauges (:mod:`varagity.observability.graph`).
    """
    from varagity.observability.corpus import CORPUS_GAUGES
    from varagity.observability.graph import GRAPH_GAUGES

    declared: dict[str, tuple[str, ...]] = {}
    for metric in _CATALOG:
        # prometheus_client exposes neither publicly; the alternative is
        # restating every name and label here, which is exactly the drift
        # this function exists to catch.
        declared[metric._name] = tuple(metric._labelnames)
    return declared | CORPUS_GAUGES | GRAPH_GAUGES

count_ingested_chunks(chunking_strategy, n)

Count chunks stored by ingestion.

Parameters:

Name	Type	Description	Default
chunking_strategy	str	Registry name of the strategy that produced them.	required
n	int	How many chunks the document stored.	required

Source code in varagity/observability/metrics.py

def count_ingested_chunks(chunking_strategy: str, n: int) -> None:
    """Count chunks stored by ingestion.

    Args:
        chunking_strategy: Registry name of the strategy that produced them.
        n: How many chunks the document stored.
    """
    INGEST_CHUNKS_TOTAL.labels(chunking_strategy=chunking_strategy).inc(n)

count_ingested_document(file_type, extraction)

Count one document stored by ingestion.

Parameters:

Name	Type	Description	Default
file_type	str	The document's file type (pdf/md/docx/…).	required
extraction	str	Extraction provenance (text or ocr_fallback) — the OCR-fallback-rate numerator.	required

Source code in varagity/observability/metrics.py

def count_ingested_document(file_type: str, extraction: str) -> None:
    """Count one document stored by ingestion.

    Args:
        file_type: The document's file type (``pdf``/``md``/``docx``/…).
        extraction: Extraction provenance (``text`` or ``ocr_fallback``) —
            the OCR-fallback-rate numerator.
    """
    INGEST_DOCS_TOTAL.labels(file_type=file_type, extraction=extraction).inc()

count_llm_tokens(prompt_tokens, completion_tokens)

Count server-reported token usage for one streamed answer.

Parameters:

Name	Type	Description	Default
prompt_tokens	int | None	Prompt-side tokens, or None when the server reported no usage (nothing is recorded for that direction).	required
completion_tokens	int | None	Completion-side tokens, same convention.	required

Source code in varagity/observability/metrics.py

def count_llm_tokens(prompt_tokens: int | None, completion_tokens: int | None) -> None:
    """Count server-reported token usage for one streamed answer.

    Args:
        prompt_tokens: Prompt-side tokens, or ``None`` when the server
            reported no usage (nothing is recorded for that direction).
        completion_tokens: Completion-side tokens, same convention.
    """
    if prompt_tokens is not None:
        LLM_TOKENS_TOTAL.labels(direction="prompt").inc(prompt_tokens)
    if completion_tokens is not None:
        LLM_TOKENS_TOTAL.labels(direction="completion").inc(completion_tokens)

count_query(method, outcome)

Count one completed query flow.

Parameters:

Name	Type	Description	Default
method	str	The retrieval method's registry name.	required
outcome	str	ok, aborted (client disconnected mid-stream), or error (the flow raised).	required

Source code in varagity/observability/metrics.py

def count_query(method: str, outcome: str) -> None:
    """Count one completed query flow.

    Args:
        method: The retrieval method's registry name.
        outcome: ``ok``, ``aborted`` (client disconnected mid-stream), or
            ``error`` (the flow raised).
    """
    QUERY_TOTAL.labels(method=method, outcome=outcome).inc()

observe_contextualize(seconds)

Record one document's situating-blurb generation latency.

Parameters:

Name	Type	Description	Default
seconds	float	Wall-clock duration of the document's contextualize stage.	required

Source code in varagity/observability/metrics.py

def observe_contextualize(seconds: float) -> None:
    """Record one document's situating-blurb generation latency.

    Args:
        seconds: Wall-clock duration of the document's contextualize stage.
    """
    CONTEXTUALIZE_LATENCY_SECONDS.observe(seconds)

observe_query_stage(stage, method, seconds)

Record one query stage's wall-clock latency.

Parameters:

Name	Type	Description	Default
stage	str	Pipeline stage name (condense/embed/retrieve/ rerank/hyde/generate). The retrieve observation of the reranked method includes its rerank sub-stage, which is also recorded separately by the reranked retriever; likewise the embed observation of the hyde method includes its hyde (passage-generation) sub-stage.	required
method	str	The retrieval method's registry name (low-cardinality label; injected non-registry implementations record as custom).	required
seconds	float	The stage's duration.	required

Source code in varagity/observability/metrics.py

def observe_query_stage(stage: str, method: str, seconds: float) -> None:
    """Record one query stage's wall-clock latency.

    Args:
        stage: Pipeline stage name (``condense``/``embed``/``retrieve``/
            ``rerank``/``hyde``/``generate``). The ``retrieve`` observation
            of the ``reranked`` method *includes* its ``rerank`` sub-stage,
            which is also recorded separately by the reranked retriever;
            likewise the ``embed`` observation of the ``hyde`` method
            includes its ``hyde`` (passage-generation) sub-stage.
        method: The retrieval method's registry name (low-cardinality
            label; injected non-registry implementations record as
            ``custom``).
        seconds: The stage's duration.
    """
    QUERY_LATENCY_SECONDS.labels(stage=stage, method=method).observe(seconds)

observe_retrieval(method, chunks)

Record the retrieved chunks' final scores and any rerank movement.

Parameters:

Name	Type	Description	Default
method	str	The retrieval method's registry name.	required
chunks	Sequence[RetrievedChunk]	The query's final retrieved chunks, best first. Each score lands in the rank-labelled histogram; chunks whose trace carries a rerank_delta also feed the rerank-movement histogram.	required

Source code in varagity/observability/metrics.py

def observe_retrieval(method: str, chunks: Sequence[RetrievedChunk]) -> None:
    """Record the retrieved chunks' final scores and any rerank movement.

    Args:
        method: The retrieval method's registry name.
        chunks: The query's final retrieved chunks, best first. Each score
            lands in the ``rank``-labelled histogram; chunks whose trace
            carries a ``rerank_delta`` also feed the rerank-movement
            histogram.
    """
    for rank, chunk in enumerate(chunks, start=1):
        RETRIEVAL_SCORE.labels(method=method, rank=str(rank)).observe(chunk.score)
        if chunk.trace is not None and chunk.trace.rerank_delta is not None:
            RERANK_DELTA.observe(chunk.trace.rerank_delta)

set_dependency_up(service, up)

Record one backing service's probe outcome.

Refreshed by every health probe — in compose, the api container's own healthcheck hits GET /api/health every 15 s, so the gauge stays current without a dedicated poller.

Parameters:

Name	Type	Description	Default
service	str	The service name (llamacpp/infinity/postgres/ elasticsearch/prefect).	required
up	bool	Whether the probe succeeded.	required

Source code in varagity/observability/metrics.py

def set_dependency_up(service: str, up: bool) -> None:
    """Record one backing service's probe outcome.

    Refreshed by every health probe — in compose, the ``api`` container's
    own healthcheck hits ``GET /api/health`` every 15 s, so the gauge stays
    current without a dedicated poller.

    Args:
        service: The service name (``llamacpp``/``infinity``/``postgres``/
            ``elasticsearch``/``prefect``).
        up: Whether the probe succeeded.
    """
    DEPENDENCY_UP.labels(service=service).set(1.0 if up else 0.0)

set_last_ingest_run(finished_at, duration_seconds, documents)

Record the last completed ingest run's headline facts.

Set by the API's ingest runner on completion (spec_v3 §6.1a). CLI ingests can't feed these — they record into their own never-scraped registry (ADR-007) — which is why the corpus gauges, not these, are what the dashboard's size panels read.

Parameters:

Name	Type	Description	Default
finished_at	float	Unix timestamp of the run's completion.	required
duration_seconds	float	The run's wall-clock duration.	required
documents	int	How many documents the run processed.	required

Source code in varagity/observability/metrics.py

def set_last_ingest_run(finished_at: float, duration_seconds: float, documents: int) -> None:
    """Record the last completed ingest run's headline facts.

    Set by the API's ingest runner on completion (spec_v3 §6.1a). CLI
    ingests can't feed these — they record into their own never-scraped
    registry (ADR-007) — which is why the corpus gauges, not these, are
    what the dashboard's size panels read.

    Args:
        finished_at: Unix timestamp of the run's completion.
        duration_seconds: The run's wall-clock duration.
        documents: How many documents the run processed.
    """
    INGEST_LAST_RUN_TIMESTAMP_SECONDS.set(finished_at)
    INGEST_LAST_RUN_DURATION_SECONDS.set(duration_seconds)
    INGEST_LAST_RUN_DOCUMENTS.set(documents)

varagity.observability.corpus

Store-derived corpus gauges: the spec_v3 §6.1a scrape-time collector.

The ingest counters in :mod:varagity.observability.metrics answer "what did this process do" — and they are the wrong instrument for "how big is my corpus". Ingestion is rare and bursty, so a counter sits flat for hours and increase() over it is 0; worse, after an API restart the series is absent and then reappears at its full value, so Prometheus never sees the 0 → 1 rise and increase() is 0 even over $__range. That is what made every Ingestion panel read zero while the metrics themselves were correct (spec_v3 §2.3, §6.1).

Gauges read from pgvector at scrape time have neither problem: they reflect store state, survive an API restart, and — unlike the counters (ADR-007) — they also show CLI ingests, which record into a never-scraped registry.

Failure posture: a store outage must degrade /metrics, never 500 it. The last good snapshot is re-served (going stale) and a fresh process with no snapshot yet emits the gauges with no samples at all. :data:varagity.observability.metrics.DEPENDENCY_UP is what reports store health; these gauges deliberately do not. A backup restore is the same posture: its TRUNCATE holds both tables until it commits (ADR-019), so the gauge connection carries a lock timeout — a scrape during the load gives up on the lock within a second and serves the last snapshot rather than hanging for the whole restore.

CorpusCollector

Bases: Collector

Expose the corpus gauges, querying pgvector at scrape time.

Registered in the default registry by the API's app factory when METRICS_ENABLED (:func:register_corpus_collector). A short TTL (:data:CACHE_TTL_SECONDS) bounds the query rate; a store outage serves the last good snapshot rather than raising.

Source code in varagity/observability/corpus.py

class CorpusCollector(Collector):
    """Expose the corpus gauges, querying pgvector at scrape time.

    Registered in the default registry by the API's app factory when
    ``METRICS_ENABLED`` (:func:`register_corpus_collector`). A short TTL
    (:data:`CACHE_TTL_SECONDS`) bounds the query rate; a store outage
    serves the last good snapshot rather than raising.
    """

    def __init__(
        self,
        store_factory: Callable[[], ContextualVectorDB] = gauge_store,
        ttl_seconds: float = CACHE_TTL_SECONDS,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """Configure the collector's store access and caching.

        Args:
            store_factory: Builds a connected store per refresh (override
                seam for tests); :func:`gauge_store`, with its lock timeout,
                by default. Called inside a ``with`` block, so the
                connection closes after every refresh.
            ttl_seconds: How long a snapshot is served before re-querying.
            clock: Monotonic time source (override seam for tests).
        """
        self._store_factory = store_factory
        self._ttl = ttl_seconds
        self._clock = clock
        self._snapshot: CorpusSnapshot | None = None
        self._fetched_at: float | None = None

    def describe(self) -> Iterable[GaugeMetricFamily]:
        """Declare the gauge families without touching the store.

        Registration calls this to detect duplicate names; without it
        ``prometheus_client`` would call :meth:`collect` at registration
        time, turning app startup into a database query.

        Returns:
            The empty gauge families this collector emits.
        """
        return self._families(CorpusSnapshot())

    def collect(self) -> Iterator[GaugeMetricFamily]:
        """Emit the corpus gauges for one scrape.

        Yields:
            The gauge families, populated from a fresh snapshot, the
            cached one, or — if the store has never been reachable in this
            process — with no samples.
        """
        snapshot = self._current()
        if snapshot is None:
            yield from self._families(None)
            return
        yield from self._families(snapshot)

    def _current(self) -> CorpusSnapshot | None:
        """Return a snapshot, refreshing it when the TTL has expired.

        Returns:
            The fresh snapshot, the still-valid cached one, the stale one
            when a refresh fails, or ``None`` when no refresh has ever
            succeeded.
        """
        now = self._clock()
        if (
            self._snapshot is not None
            and self._fetched_at is not None
            and now - self._fetched_at < self._ttl
        ):
            return self._snapshot
        try:
            self._snapshot = self._fetch()
            self._fetched_at = now
        except psycopg.Error as error:
            # Serve stale over raising: a scrape that 500s takes the whole
            # /metrics endpoint down, including the metrics that are fine.
            serving = "the last snapshot" if self._snapshot is not None else "no samples"
            if isinstance(error, psycopg.errors.LockNotAvailable):
                # A restore holds the tables until it commits (ADR-019):
                # expected, brief, and not an outage.
                logger.info(
                    "corpus gauges not refreshed — the tables are locked; serving %s", serving
                )
            else:
                logger.warning(
                    "corpus gauges not refreshed — postgres unreachable (%s); serving %s",
                    error,
                    serving,
                )
        return self._snapshot

    def _fetch(self) -> CorpusSnapshot:
        """Query the store for the current corpus counts.

        Returns:
            A fresh snapshot.

        Raises:
            psycopg.Error: If the database is unreachable or the queries
                fail; the caller degrades to the cached snapshot.
        """
        with self._store_factory() as store:
            return CorpusSnapshot(
                documents=store.document_count(),
                chunks=store.chunk_count(),
                documents_by_type=store.document_count_by_type(),
                chunks_by_strategy=store.chunk_count_by_strategy(),
            )

    def _families(self, snapshot: CorpusSnapshot | None) -> list[GaugeMetricFamily]:
        """Render a snapshot as gauge families.

        Args:
            snapshot: The counts to render, or ``None`` to emit the
                families with no samples (never-reachable store).

        Returns:
            One family per :data:`CORPUS_GAUGES` entry, in catalog order.
        """
        documents = GaugeMetricFamily(_DOCS, "Documents currently in the corpus (from pgvector).")
        chunks = GaugeMetricFamily(_CHUNKS, "Chunks currently stored (from pgvector).")
        by_type = GaugeMetricFamily(
            _DOCS_BY_TYPE,
            "Documents currently in the corpus, by file type (from pgvector).",
            labels=["file_type"],
        )
        by_strategy = GaugeMetricFamily(
            _CHUNKS_BY_STRATEGY,
            "Chunks currently stored, by the strategy that produced them (from pgvector).",
            labels=["chunking_strategy"],
        )
        if snapshot is not None:
            documents.add_metric([], snapshot.documents)
            chunks.add_metric([], snapshot.chunks)
            for file_type, count in sorted(snapshot.documents_by_type.items()):
                by_type.add_metric([file_type], count)
            for strategy, count in sorted(snapshot.chunks_by_strategy.items()):
                by_strategy.add_metric([strategy], count)
        return [documents, chunks, by_type, by_strategy]

__init__(store_factory=gauge_store, ttl_seconds=CACHE_TTL_SECONDS, clock=time.monotonic)

Configure the collector's store access and caching.

Parameters:

Name	Type	Description	Default
store_factory	Callable[[], ContextualVectorDB]	Builds a connected store per refresh (override seam for tests); :func:gauge_store, with its lock timeout, by default. Called inside a with block, so the connection closes after every refresh.	gauge_store
ttl_seconds	float	How long a snapshot is served before re-querying.	CACHE_TTL_SECONDS
clock	Callable[[], float]	Monotonic time source (override seam for tests).	monotonic

Source code in varagity/observability/corpus.py

def __init__(
    self,
    store_factory: Callable[[], ContextualVectorDB] = gauge_store,
    ttl_seconds: float = CACHE_TTL_SECONDS,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    """Configure the collector's store access and caching.

    Args:
        store_factory: Builds a connected store per refresh (override
            seam for tests); :func:`gauge_store`, with its lock timeout,
            by default. Called inside a ``with`` block, so the
            connection closes after every refresh.
        ttl_seconds: How long a snapshot is served before re-querying.
        clock: Monotonic time source (override seam for tests).
    """
    self._store_factory = store_factory
    self._ttl = ttl_seconds
    self._clock = clock
    self._snapshot: CorpusSnapshot | None = None
    self._fetched_at: float | None = None

collect()

Emit the corpus gauges for one scrape.

Yields:

Type	Description
GaugeMetricFamily	The gauge families, populated from a fresh snapshot, the
GaugeMetricFamily	cached one, or — if the store has never been reachable in this
GaugeMetricFamily	process — with no samples.

Source code in varagity/observability/corpus.py

def collect(self) -> Iterator[GaugeMetricFamily]:
    """Emit the corpus gauges for one scrape.

    Yields:
        The gauge families, populated from a fresh snapshot, the
        cached one, or — if the store has never been reachable in this
        process — with no samples.
    """
    snapshot = self._current()
    if snapshot is None:
        yield from self._families(None)
        return
    yield from self._families(snapshot)

describe()

Declare the gauge families without touching the store.

Registration calls this to detect duplicate names; without it prometheus_client would call :meth:collect at registration time, turning app startup into a database query.

Returns:

Type	Description
Iterable[GaugeMetricFamily]	The empty gauge families this collector emits.

Source code in varagity/observability/corpus.py

def describe(self) -> Iterable[GaugeMetricFamily]:
    """Declare the gauge families without touching the store.

    Registration calls this to detect duplicate names; without it
    ``prometheus_client`` would call :meth:`collect` at registration
    time, turning app startup into a database query.

    Returns:
        The empty gauge families this collector emits.
    """
    return self._families(CorpusSnapshot())

CorpusSnapshot dataclass

One scrape's worth of corpus counts.

Attributes:

Name	Type	Description
documents	int	Total ingested documents.
chunks	int	Total stored chunks.
documents_by_type	dict[str, int]	file_type → document count.
chunks_by_strategy	dict[str, int]	chunking_strategy → chunk count.

Source code in varagity/observability/corpus.py

@dataclass(frozen=True)
class CorpusSnapshot:
    """One scrape's worth of corpus counts.

    Attributes:
        documents: Total ingested documents.
        chunks: Total stored chunks.
        documents_by_type: ``file_type`` → document count.
        chunks_by_strategy: ``chunking_strategy`` → chunk count.
    """

    documents: int = 0
    chunks: int = 0
    documents_by_type: dict[str, int] = field(default_factory=dict)
    chunks_by_strategy: dict[str, int] = field(default_factory=dict)

gauge_store()

Connect the store the gauges query, with a short lock timeout.

A restore holds documents and chunks exclusively until it commits; with the timeout, a count query against them fails fast (psycopg.errors.LockNotAvailable) and the collector serves its last snapshot instead of queueing behind the restore.

Returns:

Type	Description
ContextualVectorDB	A connected store.

Raises:

Type	Description
OperationalError	If the database is unreachable.

Source code in varagity/observability/corpus.py

def gauge_store() -> ContextualVectorDB:
    """Connect the store the gauges query, with a short lock timeout.

    A restore holds ``documents`` and ``chunks`` exclusively until it
    commits; with the timeout, a count query against them fails fast
    (``psycopg.errors.LockNotAvailable``) and the collector serves its last
    snapshot instead of queueing behind the restore.

    Returns:
        A connected store.

    Raises:
        psycopg.OperationalError: If the database is unreachable.
    """
    return ContextualVectorDB(
        psycopg.conninfo.make_conninfo(
            default_conninfo(), options=f"-c lock_timeout={GAUGE_LOCK_TIMEOUT_MS}"
        )
    )

register_corpus_collector(registry=REGISTRY, collector=None)

Register the corpus collector, replacing any previous registration.

Idempotent on purpose: create_app() runs once per process in production but many times across a test session, and re-registering the same gauge names in the process-wide default registry would raise.

Parameters:

Name	Type	Description	Default
registry	CollectorRegistry	The registry to register into (defaults to the process-wide one that GET /metrics serves).	REGISTRY
collector	CorpusCollector | None	The collector to register; a default-configured one is built when omitted.	None

Returns:

Type	Description
CorpusCollector	The registered collector.

Source code in varagity/observability/corpus.py

def register_corpus_collector(
    registry: CollectorRegistry = REGISTRY,
    collector: CorpusCollector | None = None,
) -> CorpusCollector:
    """Register the corpus collector, replacing any previous registration.

    Idempotent on purpose: ``create_app()`` runs once per process in
    production but many times across a test session, and re-registering the
    same gauge names in the process-wide default registry would raise.

    Args:
        registry: The registry to register into (defaults to the
            process-wide one that ``GET /metrics`` serves).
        collector: The collector to register; a default-configured one is
            built when omitted.

    Returns:
        The registered collector.
    """
    collector = collector or CorpusCollector()
    for existing in list(getattr(registry, "_collector_to_names", {})):
        if isinstance(existing, CorpusCollector):
            registry.unregister(existing)
    registry.register(collector)
    return collector

varagity.observability.graph

Workdir-derived graph gauges: the ADR-013 shape, applied to the graph.

The graph corpus has exactly the problem the chunk-RAG corpus had before ADR-013: "how big is my graph" is a state question, and an in-process counter cannot answer it. A backfill runs for hours and then stops; the counter that recorded it is born at its full value after the next API restart, so increase() over it is 0 — over any window. Gauges read at scrape time have neither problem.

Where they read from is the difference. The corpus gauges query pgvector; these read the files in the engine's working directory, and never open the engine (stage-2 decision #18):

• varagity_graph_documents{status} counts the engine's own doc-status records — the same pending/processing/processed/failed vocabulary the Graph RAG tab shows, so a stalled or failed backfill is visible on the dashboard rather than only in the tab.
• varagity_graph_entities / _relations / _messages come from the summary sidecar every build and delete refreshes (:mod:varagity.graph.manifest), which exists precisely so a status poll or a scrape never walks a multi-megabyte graphml.

Opening a session here would be worse than slow: the engine is single-writer per working directory and this process is that writer, so a scrape must not initialize storages a build is holding — and on an unbuilt graph it would create the workdir, turning a Prometheus scrape into "the graph now exists".

Failure posture matches the corpus collector's. An unreadable file serves the last good snapshot (going stale) rather than raising, because a scrape that 500s takes /metrics down including the metrics that are fine. An unbuilt graph is not an error and not a zero: the families are emitted with no samples at all, the same honesty :class:~varagity.api.schemas.GraphStatusOut keeps when it reports None rather than 0 for a graph nothing has been extracted into yet.

GraphCollector

Bases: Collector

Expose the graph gauges, reading the workdir at scrape time.

Registered in the default registry by the API's app factory when METRICS_ENABLED (:func:register_graph_collector). A short TTL (:data:CACHE_TTL_SECONDS) bounds the read rate; an unreadable workdir serves the last good snapshot rather than raising.

Source code in varagity/observability/graph.py

class GraphCollector(Collector):
    """Expose the graph gauges, reading the workdir at scrape time.

    Registered in the default registry by the API's app factory when
    ``METRICS_ENABLED`` (:func:`register_graph_collector`). A short TTL
    (:data:`CACHE_TTL_SECONDS`) bounds the read rate; an unreadable workdir
    serves the last good snapshot rather than raising.
    """

    def __init__(
        self,
        workdir_factory: Callable[[], Path] = graph_workdir,
        ttl_seconds: float = CACHE_TTL_SECONDS,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """Configure the collector's workdir access and caching.

        Args:
            workdir_factory: Resolves the graph's working directory per
                refresh (override seam for tests). Deliberately the shared
                settings derivation rather than the graph service's own
                property: a scrape must not depend on the engine registry
                being warm, nor construct the service singleton. Called per
                refresh, not once, so a settings change lands without a
                restart.
            ttl_seconds: How long a snapshot is served before re-reading.
            clock: Monotonic time source (override seam for tests).
        """
        self._workdir_factory = workdir_factory
        self._ttl = ttl_seconds
        self._clock = clock
        self._snapshot: GraphSnapshot | None = None
        self._fetched_at: float | None = None

    def describe(self) -> Iterable[GaugeMetricFamily]:
        """Declare the gauge families without touching the workdir.

        Registration calls this to detect duplicate names; without it
        ``prometheus_client`` would call :meth:`collect` at registration
        time, turning app startup into a disk read.

        Returns:
            The empty gauge families this collector emits.
        """
        return self._families(None)

    def collect(self) -> Iterator[GaugeMetricFamily]:
        """Emit the graph gauges for one scrape.

        Yields:
            The gauge families, populated from a fresh snapshot, the cached
            one, or — for an unbuilt graph or a workdir that has never been
            readable — with no samples.
        """
        yield from self._families(self._current())

    def _current(self) -> GraphSnapshot | None:
        """Return a snapshot, refreshing it when the TTL has expired.

        Returns:
            The fresh snapshot, the still-valid cached one, the stale one
            when a refresh fails, or ``None`` when no refresh has ever
            succeeded.
        """
        now = self._clock()
        if (
            self._snapshot is not None
            and self._fetched_at is not None
            and now - self._fetched_at < self._ttl
        ):
            return self._snapshot
        try:
            self._snapshot = self._fetch()
            self._fetched_at = now
        except (OSError, ValueError) as error:
            # Serve stale over raising: /metrics must survive a workdir a
            # build is churning through, and the corpus gauges beside these
            # are none of this failure's business.
            logger.warning(
                "graph gauges not refreshed — the workdir would not read (%s); serving %s",
                error,
                "the last snapshot" if self._snapshot is not None else "no samples",
            )
        return self._snapshot

    def _fetch(self) -> GraphSnapshot:
        """Read the workdir's sidecars for the current graph size.

        Returns:
            A fresh snapshot; an empty one when the working directory does
            not exist, which renders as no samples rather than zeros.

        Raises:
            OSError: If a present file cannot be read.
            ValueError: If the doc-status file is not readable JSON.
        """
        workdir = self._workdir_factory()
        if not workdir.is_dir():
            return GraphSnapshot()
        summary = load_summary(workdir)
        return GraphSnapshot(
            documents_by_status=read_document_statuses(workdir),
            entities=summary.entities if summary is not None else None,
            relations=summary.relations if summary is not None else None,
            messages=summary.message_guids if summary is not None else None,
        )

    def _families(self, snapshot: GraphSnapshot | None) -> list[GaugeMetricFamily]:
        """Render a snapshot as gauge families.

        Args:
            snapshot: The counts to render, or ``None`` to emit the
                families with no samples (never-readable workdir).

        Returns:
            One family per :data:`GRAPH_GAUGES` entry, in catalog order.
            Fields the workdir does not know stay sample-less: an unbuilt
            graph must not report itself as a graph with zero entities.
        """
        documents = GaugeMetricFamily(
            _DOCUMENTS,
            "Transcript documents the graph engine holds, by processing status.",
            labels=["status"],
        )
        entities = GaugeMetricFamily(_ENTITIES, "Entities in the extracted graph.")
        relations = GaugeMetricFamily(_RELATIONS, "Relations in the extracted graph.")
        messages = GaugeMetricFamily(
            _MESSAGES, "Messages the graph's workdir manifest accounts for."
        )
        if snapshot is not None:
            for status, count in sorted(snapshot.documents_by_status.items()):
                documents.add_metric([status], count)
            if snapshot.entities is not None:
                entities.add_metric([], snapshot.entities)
            if snapshot.relations is not None:
                relations.add_metric([], snapshot.relations)
            if snapshot.messages is not None:
                messages.add_metric([], snapshot.messages)
        return [documents, entities, relations, messages]

__init__(workdir_factory=graph_workdir, ttl_seconds=CACHE_TTL_SECONDS, clock=time.monotonic)

Configure the collector's workdir access and caching.

Parameters:

Name	Type	Description	Default
workdir_factory	Callable[[], Path]	Resolves the graph's working directory per refresh (override seam for tests). Deliberately the shared settings derivation rather than the graph service's own property: a scrape must not depend on the engine registry being warm, nor construct the service singleton. Called per refresh, not once, so a settings change lands without a restart.	graph_workdir
ttl_seconds	float	How long a snapshot is served before re-reading.	CACHE_TTL_SECONDS
clock	Callable[[], float]	Monotonic time source (override seam for tests).	monotonic

Source code in varagity/observability/graph.py

def __init__(
    self,
    workdir_factory: Callable[[], Path] = graph_workdir,
    ttl_seconds: float = CACHE_TTL_SECONDS,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    """Configure the collector's workdir access and caching.

    Args:
        workdir_factory: Resolves the graph's working directory per
            refresh (override seam for tests). Deliberately the shared
            settings derivation rather than the graph service's own
            property: a scrape must not depend on the engine registry
            being warm, nor construct the service singleton. Called per
            refresh, not once, so a settings change lands without a
            restart.
        ttl_seconds: How long a snapshot is served before re-reading.
        clock: Monotonic time source (override seam for tests).
    """
    self._workdir_factory = workdir_factory
    self._ttl = ttl_seconds
    self._clock = clock
    self._snapshot: GraphSnapshot | None = None
    self._fetched_at: float | None = None

collect()

Emit the graph gauges for one scrape.

Yields:

Type	Description
GaugeMetricFamily	The gauge families, populated from a fresh snapshot, the cached
GaugeMetricFamily	one, or — for an unbuilt graph or a workdir that has never been
GaugeMetricFamily	readable — with no samples.

Source code in varagity/observability/graph.py

def collect(self) -> Iterator[GaugeMetricFamily]:
    """Emit the graph gauges for one scrape.

    Yields:
        The gauge families, populated from a fresh snapshot, the cached
        one, or — for an unbuilt graph or a workdir that has never been
        readable — with no samples.
    """
    yield from self._families(self._current())

describe()

Declare the gauge families without touching the workdir.

Registration calls this to detect duplicate names; without it prometheus_client would call :meth:collect at registration time, turning app startup into a disk read.

Returns:

Type	Description
Iterable[GaugeMetricFamily]	The empty gauge families this collector emits.

Source code in varagity/observability/graph.py

def describe(self) -> Iterable[GaugeMetricFamily]:
    """Declare the gauge families without touching the workdir.

    Registration calls this to detect duplicate names; without it
    ``prometheus_client`` would call :meth:`collect` at registration
    time, turning app startup into a disk read.

    Returns:
        The empty gauge families this collector emits.
    """
    return self._families(None)

GraphSnapshot dataclass

One scrape's worth of graph size, as the workdir reports it.

Every field is optional-shaped because "nothing has been built here" and "the graph would not say" are both real states, and a 0 would read like an empty graph on a dashboard.

Attributes:

Name	Type	Description
documents_by_status	dict[str, int]	Engine doc-status name → document count.
entities	int | None	Nodes in the graph, or None when unknown.
relations	int | None	Edges in the graph, or None when unknown.
messages	int | None	Distinct messages the manifest accounts for, or None.

Source code in varagity/observability/graph.py

@dataclass(frozen=True)
class GraphSnapshot:
    """One scrape's worth of graph size, as the workdir reports it.

    Every field is optional-shaped because "nothing has been built here"
    and "the graph would not say" are both real states, and a ``0`` would
    read like an empty graph on a dashboard.

    Attributes:
        documents_by_status: Engine doc-status name → document count.
        entities: Nodes in the graph, or ``None`` when unknown.
        relations: Edges in the graph, or ``None`` when unknown.
        messages: Distinct messages the manifest accounts for, or ``None``.
    """

    documents_by_status: dict[str, int] = field(default_factory=dict)
    entities: int | None = None
    relations: int | None = None
    messages: int | None = None

read_document_statuses(workdir)

Count the engine's documents by processing status, from disk.

Parameters:

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

Returns:

Type	Description
dict[str, int]	Status name → document count. An absent file is an empty mapping:
dict[str, int]	a workdir that has never had a document enqueued genuinely has no
dict[str, int]	statuses, which is not a failure.

Raises:

Type	Description
OSError	If the file exists but cannot be read.
ValueError	If it is not readable JSON — the caller degrades to the last good snapshot rather than reporting a suddenly empty graph.

Source code in varagity/observability/graph.py

def read_document_statuses(workdir: Path) -> dict[str, int]:
    """Count the engine's documents by processing status, from disk.

    Args:
        workdir: The engine's working directory.

    Returns:
        Status name → document count. An absent file is an empty mapping:
        a workdir that has never had a document enqueued genuinely has no
        statuses, which is not a failure.

    Raises:
        OSError: If the file exists but cannot be read.
        ValueError: If it is not readable JSON — the caller degrades to the
            last good snapshot rather than reporting a suddenly empty graph.
    """
    path = workdir / DOC_STATUS_FILENAME
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return {}
    if not isinstance(raw, Mapping):
        raise ValueError(f"{path} is not a document-status mapping")
    counts: Counter[str] = Counter()
    for record in raw.values():
        status = record.get("status") if isinstance(record, Mapping) else None
        counts[str(status) if status else "unknown"] += 1
    return dict(counts)

register_graph_collector(registry=REGISTRY, collector=None)

Register the graph collector, replacing any previous registration.

Idempotent for the corpus collector's reason: create_app() runs once per process in production but many times across a test session, and re-registering the same gauge names in the process-wide default registry would raise.

Parameters:

Name	Type	Description	Default
registry	CollectorRegistry	The registry to register into (defaults to the process-wide one that GET /metrics serves).	REGISTRY
collector	GraphCollector | None	The collector to register; a default-configured one is built when omitted.	None

Returns:

Type	Description
GraphCollector	The registered collector.

Source code in varagity/observability/graph.py

def register_graph_collector(
    registry: CollectorRegistry = REGISTRY,
    collector: GraphCollector | None = None,
) -> GraphCollector:
    """Register the graph collector, replacing any previous registration.

    Idempotent for the corpus collector's reason: ``create_app()`` runs once
    per process in production but many times across a test session, and
    re-registering the same gauge names in the process-wide default registry
    would raise.

    Args:
        registry: The registry to register into (defaults to the
            process-wide one that ``GET /metrics`` serves).
        collector: The collector to register; a default-configured one is
            built when omitted.

    Returns:
        The registered collector.
    """
    collector = collector or GraphCollector()
    for existing in list(getattr(registry, "_collector_to_names", {})):
        if isinstance(existing, GraphCollector):
            registry.unregister(existing)
    registry.register(collector)
    return collector