review: per-replica backlog gauge, drop duplicated constants and .meta refs

Addresses Vineeth's review on #927.

Blocking:
- message_embeddings_pending is a DB-global count, so drive it from
  ReconcilerScheduler._scheduler_loop (runs on every replica, every
  interval) instead of run_vector_reconciliation_cycle (runs off the
  queue behind work-unit dedup, so one replica per cycle). Combined with
  the zero-init, the old placement made every replica that never won the
  work unit export a confident permanent 0. Help string now names the
  owner so dashboards don't reach for sum().
- guard initialize_telemetry_dropped_metrics on METRICS.ENABLED,
  matching its sibling initializer.
- drop the duplicate REASONING_LEVELS; import the one in src/config.

Non-blocking:
- walk BaseSpecialist recursively via a shared utils.types.walk_subclasses
  (replaces the direct-children-only __subclasses__() and the test's
  private copy of the same helper).
- derive the specialist assertion from the subclasses instead of
  hardcoding two names — the hardcoded pair kept passing after
  CardRefreshSpecialist landed, leaving it uncovered.
- inline the zero-init rationale and the multi-instance bucket taxonomy;
  removes both pointers to a .meta design doc that is not in the repo.

Tests: new tests/reconciler/test_pending_backlog_gauge.py pins both
halves of the relocation (verified it fails when reverted).

src/reconciler/scheduler.py

@@ -21,6 +21,7 @@ from src import models
 from src.config import settings
 from src.dependencies import tracked_db
 from src.models import QueueItem
+from src.reconciler.sync_vectors import record_pending_embeddings_backlog
 
 logger = logging.getLogger(__name__)
 
@@ -154,6 +155,13 @@ class ReconcilerScheduler:
             while not self._shutdown_event.is_set():
                 now = datetime.now(timezone.utc)
 
+                # Refresh the pending-embeddings backlog gauge on EVERY replica,
+                # not just whichever one wins the sync_vectors work unit — the
+                # count is DB-global, so a replica that never ran a cycle would
+                # otherwise export a stale (or zero-initialized) value forever.
+                # See record_pending_embeddings_backlog for the full rationale.
+                await record_pending_embeddings_backlog()
+
                 # Check each task and enqueue if due
                 for task_name, task in RECONCILER_TASKS.items():
                     next_run = self._next_run.get(task_name, now)

src/reconciler/sync_vectors.py

@@ -701,13 +701,21 @@ async def _cleanup_pgvector_batch(
         return True
 
 
-async def _record_pending_embeddings_backlog() -> None:
+async def record_pending_embeddings_backlog() -> None:
     """Set the pending-embeddings backlog gauge to the current count of
     MessageEmbedding rows awaiting a vector (sync_state='pending').
 
-    Called at the end of each reconciliation cycle so the gauge reflects the
+    Called from ``ReconcilerScheduler._scheduler_loop`` — deliberately NOT from
-    residual backlog after the sweep. Best-effort: a metrics/DB hiccup here must
+    ``run_vector_reconciliation_cycle``. The cycle runs off the queue behind
-    never fail the reconciliation cycle.
+    work-unit dedup, so exactly one deriver replica executes it; driving the
+    gauge from there would leave every other replica exporting a stale value, or
+    (worse, given this metric is zero-initialized) a confident permanent 0 it had
+    never measured. The count is a property of the database, not of the process,
+    so every replica must refresh it on its own timer for ``max()``/``avg()`` to
+    mean anything. One indexed COUNT per replica per interval is negligible —
+    ``ix_message_embeddings_sync_state_last_sync_at`` covers it.
+
+    Best-effort: a metrics/DB hiccup here must never break the scheduler loop.
     """
     if not settings.METRICS.ENABLED:
         return
@@ -754,7 +762,6 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics:
             if not (embs_work or cleanup_work):
                 break
         logger.debug("Vector reconciliation cycle completed (pgvector mode)")
-        await _record_pending_embeddings_backlog()
         return metrics
 
     # External vector store mode - reconcile documents, embeddings, and cleanup
@@ -782,5 +789,4 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics:
             break
 
     logger.debug("Vector reconciliation cycle completed")
-    await _record_pending_embeddings_backlog()
     return metrics

src/telemetry/events/__init__.py

@@ -147,8 +147,14 @@ __all__ = [
 # Explicit registry of CloudEvents `type` values, used to pre-materialize the
 # `telemetry_events_emitted` / `telemetry_events_sampled_out` counter children at
 # 0 so those metrics are visible in Prometheus before any event fires (see
-# src/telemetry/prometheus/metrics.py:initialize_bounded_metrics and
+# src/telemetry/prometheus/metrics.py:initialize_bounded_metrics for why absent
-# .meta design telemetry-counter-zero-init).
+# and zero are worth distinguishing).
+#
+# This is an explicit literal rather than a set derived from BaseEvent
+# subclasses: the derived version would silently follow whatever happens to be
+# imported at init time, so a type could drop out of the registry without any
+# code change. Pairing a hand-maintained list with a drift-guard test keeps the
+# failure loud and at the right moment.
 #
 # ⚠️ When you add a new BaseEvent subclass, add its `_event_type` here (and to
 # HIGH_VOLUME_EVENT_TYPES if `_volume_class == "high_volume"`). The drift-guard

src/telemetry/prometheus/metrics.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import logging
 from collections.abc import Iterator
 from enum import Enum
-from typing import cast, final, get_args
+from typing import cast, final
 
 from prometheus_client import (
     CONTENT_TYPE_LATEST,
@@ -20,7 +20,8 @@ from prometheus_client.core import GaugeMetricFamily
 from starlette.requests import Request
 from starlette.responses import Response
 
-from src.config import ReasoningLevel, settings
+from src.config import REASONING_LEVELS, settings
+from src.utils.types import walk_subclasses
 
 disable_created_metrics()
 
@@ -66,11 +67,11 @@ class DialecticComponents(Enum):
     TOTAL = "total"
 
 
-# Bounded label domains used to zero-initialize counter children at startup (see
+# Bounded label domains used to zero-initialize counter children at startup are
-# initialize_bounded_metrics). REASONING_LEVELS is derived from the config
+# defined below (see initialize_bounded_metrics). Reasoning levels come from
-# Literal so it never drifts.
+# src.config.REASONING_LEVELS, which is derived from the config Literal so it
-REASONING_LEVELS: tuple[str, ...] = get_args(ReasoningLevel)
+# never drifts.
-
+#
 # Valid (token_type, component) pairs for deriver_tokens_processed, per task_type.
 # NOT the cartesian product: input tokens only pair with input components, output
 # only with OUTPUT_TOTAL, and PREVIOUS_SUMMARY occurs only for summary tasks
@@ -186,10 +187,16 @@ telemetry_buffer_size_gauge = NamespacedGauge(
 # Embedding backlog: MessageEmbedding rows still awaiting a vector
 # (sync_state='pending'). Distinct from embed_now_tasks_in_flight (which counts
 # in-flight fast-path work in the API process) — this is the durable, DB-wide
-# backlog the reconciler drains. Set once per reconciliation cycle in the deriver.
+# backlog the reconciler drains. Every deriver replica refreshes it on its own
+# timer from ReconcilerScheduler._scheduler_loop, so replicas can disagree by at
+# most one interval. The help string names the owner explicitly because this is
+# the only gauge here whose value is service-wide rather than per-process, and a
+# dashboard author must not reach for sum().
 message_embeddings_pending_gauge = NamespacedGauge(
     "message_embeddings_pending",
-    "MessageEmbedding rows awaiting embedding (sync_state='pending')",
+    "MessageEmbedding rows awaiting embedding (sync_state='pending'). "
+    + "Service-wide DB count, reported independently by every replica — "
+    + "aggregate with max() or avg(), never sum()",
     ["namespace"],
 )
 
@@ -393,23 +400,54 @@ class PrometheusMetrics:
         startup keeps the metric present at 0, so a missing series signals a broken
         scrape rather than "no drops".
 
-        Called per-emitter from ``TelemetryEmitter.start()`` because the reason
+        Called per-emitter from ``TelemetryEmitter.start()`` rather than hoisted
-        values are prefix-dependent (the trace emitter uses a ``trace_`` prefix),
+        into the process-level ``initialize_bounded_metrics``: the trace emitter
-        which the process-level ``initialize_bounded_metrics`` does not know.
+        (whose reasons carry a ``trace_`` prefix) only exists when
+        ``TRACE_PAYLOADS_ENABLED`` is set, so hoisting would fabricate ``trace_*``
+        series on deployments that run with tracing off.
 
         Args:
             reasons: The reason label values the calling emitter can produce.
         """
+        if not settings.METRICS.ENABLED:
+            return
+
         for reason in reasons:
             self._touch(telemetry_events_dropped_counter, reason=reason)
 
     def initialize_bounded_metrics(self, *, instance_type: str) -> None:
         """Pre-create bounded-label counter children at 0 for this process.
 
-        See .meta design telemetry-counter-zero-init. Only counters whose full
+        A Prometheus counter does not exist until its first increment, so a
-        label domain is bounded, enumerable at startup, and actually emitted by
+        never-yet-incremented metric is indistinguishable from a broken scrape:
-        THIS process are materialized; high-cardinality labels (endpoint,
+        you cannot graph or alert on a series that is absent. Materializing the
-        workspace_name) and impossible label tuples are deliberately left absent.
+        children at 0 inverts that — a missing series now means something is
+        wrong, and "no events" reads as a flat 0 instead of a gap.
+
+        That only holds for label sets we can enumerate honestly, so a metric is
+        initialized here only when its full label domain is bounded, enumerable
+        at startup, and actually emitted by THIS process. High-cardinality labels
+        (endpoint, workspace_name) and impossible label tuples are deliberately
+        left absent — fabricating a permanently-0 series that no code path can
+        ever increment is the same lie in the other direction.
+
+        Multi-instance safety splits the metrics here into three buckets:
+
+        1. instance-scoped (``telemetry_buffer_size``,
+           ``embed_now_tasks_in_flight``) — per-process by nature, so any
+           aggregation is meaningful and zero-init is unambiguously right.
+        2. service-scoped additive (the token counters,
+           ``telemetry_events_emitted``) — each instance holds a partial count
+           and ``sum()`` reconstructs the whole, so multi-instance safe.
+        3. service-scoped non-additive — every instance claims to speak for the
+           whole service, so no aggregation is correct once they disagree. A
+           metric in this bucket MUST be refreshed by every instance on its own
+           timer (see ``message_embeddings_pending``, refreshed per replica from
+           ``ReconcilerScheduler._scheduler_loop``), or it does not belong in the
+           app at all — it belongs in an exporter that yields one series.
+
+        Prometheus stamps ``instance``/``job`` at scrape time, which is why
+        buckets 1 and 2 need no special handling.
 
         ``telemetry_events_dropped`` is handled separately, per-emitter, in
         ``TelemetryEmitter.start()`` (prefix-dependent — see above).
@@ -462,10 +500,11 @@ class PrometheusMetrics:
                     )
             # dreamer tokens: specialist_name x token_type. Specialist names are
             # derived from the concrete BaseSpecialist subclasses so a new
-            # specialist can't silently miss init.
+            # specialist can't silently miss init. Walked recursively — a
+            # specialist that subclasses another specialist is still a specialist.
             from src.dreamer.specialists import BaseSpecialist
 
-            for specialist in BaseSpecialist.__subclasses__():
+            for specialist in walk_subclasses(BaseSpecialist):
                 for token_type in TokenTypes:
                     self._touch(
                         dreamer_tokens_processed_counter,

src/utils/types.py

@@ -1,4 +1,4 @@
-from collections.abc import Awaitable, Callable, Generator
+from collections.abc import Awaitable, Callable, Generator, Iterator
 from contextlib import contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass, field
@@ -6,6 +6,20 @@ from typing import Any, Generic, Literal, TypeVar
 
 T = TypeVar("T")
 
+
+def walk_subclasses(cls: type[T]) -> Iterator[type[T]]:
+    """Yield every subclass of ``cls``, recursively.
+
+    ``type.__subclasses__()`` is direct-children-only, so a grandchild class is
+    silently invisible to it. Any registry that enumerates subclasses to decide
+    what to initialize or validate wants the transitive closure — otherwise
+    subclassing a concrete class is enough to slip past the check.
+    """
+    for subclass in cls.__subclasses__():
+        yield subclass
+        yield from walk_subclasses(subclass)
+
+
 # Context variable for tracking current iteration in tool execution loop
 # This is used for telemetry to associate tool calls with their iteration
 _current_iteration: ContextVar[int] = ContextVar("current_iteration", default=0)

tests/reconciler/test_pending_backlog_gauge.py

@@ -0,0 +1,79 @@
+"""The pending-embeddings backlog gauge must be refreshed per-replica.
+
+``message_embeddings_pending`` reports a DB-global count, so it is the one gauge
+here whose value is service-wide rather than per-process. It is also zero-
+initialized at startup, which makes a missing refresh actively harmful: a replica
+that never measured the backlog would export a confident, permanently-healthy 0.
+
+The count therefore has to be driven from ``ReconcilerScheduler._scheduler_loop``
+(runs on every replica, every interval) and NOT from
+``run_vector_reconciliation_cycle`` (runs off the queue behind work-unit dedup, so
+exactly one replica per cycle executes it). These tests pin both halves of that.
+"""
+
+import asyncio
+
+import pytest
+
+from src.reconciler import scheduler as scheduler_module
+from src.reconciler import sync_vectors
+from src.reconciler.scheduler import ReconcilerScheduler
+
+
+@pytest.fixture(autouse=True)
+def _reset_scheduler_singleton():
+    ReconcilerScheduler.reset_singleton()
+    yield
+    ReconcilerScheduler.reset_singleton()
+
+
+async def test_scheduler_loop_refreshes_backlog_gauge(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Every scheduler iteration refreshes the gauge, on every replica.
+
+    Patched at the scheduler's own reference so this asserts the call site, not
+    just that the function exists.
+    """
+    calls = 0
+    refreshed = asyncio.Event()
+
+    async def _fake_refresh() -> None:
+        nonlocal calls
+        calls += 1
+        refreshed.set()
+
+    async def _never_enqueue(_task: object) -> bool:
+        return False
+
+    monkeypatch.setattr(
+        scheduler_module, "record_pending_embeddings_backlog", _fake_refresh
+    )
+    monkeypatch.setattr(
+        ReconcilerScheduler, "_try_enqueue_task", _never_enqueue, raising=True
+    )
+
+    scheduler = ReconcilerScheduler()
+    await scheduler.start()
+    try:
+        await asyncio.wait_for(refreshed.wait(), timeout=5.0)
+    finally:
+        await scheduler.shutdown()
+
+    assert calls >= 1, "scheduler loop never refreshed the backlog gauge"
+
+
+def test_reconciliation_cycle_does_not_drive_the_gauge() -> None:
+    """The queue-driven cycle must not be the thing that sets the gauge.
+
+    If the refresh moves back into ``run_vector_reconciliation_cycle``, only the
+    replica that wins the ``sync_vectors`` work unit would ever measure the
+    backlog, and the zero-init would go back to lying on all the others.
+
+    Structural guard: the cycle is a long DB-driven coroutine, so this inspects
+    the global names it references rather than executing it.
+    """
+    assert hasattr(sync_vectors, "record_pending_embeddings_backlog")
+
+    referenced = sync_vectors.run_vector_reconciliation_cycle.__code__.co_names
+    assert "record_pending_embeddings_backlog" not in referenced

tests/telemetry/test_metric_zero_init.py

@@ -18,18 +18,19 @@ from uuid import uuid4
 import pytest
 from prometheus_client import REGISTRY
 
-from src.config import settings
+from src.config import REASONING_LEVELS, settings
+from src.dreamer.specialists import BaseSpecialist
 from src.telemetry.events import ALL_EVENT_TYPES, HIGH_VOLUME_EVENT_TYPES
 from src.telemetry.events.base import BaseEvent
 from src.telemetry.prometheus.metrics import (
     _DERIVER_TOKEN_COMBOS_BY_TASK,  # pyright: ignore[reportPrivateUsage]
-    REASONING_LEVELS,
     DeriverComponents,
     DeriverTaskTypes,
     DialecticComponents,
     TokenTypes,
     prometheus_metrics,
 )
+from src.utils.types import walk_subclasses
 
 
 def unique_ns(tag: str) -> str:
@@ -70,12 +71,6 @@ def sample(name: str, **labels: str) -> float | None:
 # ---------------------------------------------------------------------------
 
 
-def _walk_event_subclasses(cls: type[BaseEvent]) -> Iterator[type[BaseEvent]]:
-    for sub in cls.__subclasses__():
-        yield sub
-        yield from _walk_event_subclasses(sub)
-
-
 def test_all_event_types_registry_matches_subclasses():
     """ALL_EVENT_TYPES must equal every BaseEvent subclass's _event_type.
 
@@ -85,7 +80,7 @@ def test_all_event_types_registry_matches_subclasses():
     """
     discovered = {
         event_type
-        for cls in _walk_event_subclasses(BaseEvent)
+        for cls in walk_subclasses(BaseEvent)
         if (event_type := getattr(cls, "_event_type", None)) is not None
     }
     assert set(ALL_EVENT_TYPES) == discovered
@@ -96,7 +91,7 @@ def test_high_volume_registry_matches_subclasses():
     """HIGH_VOLUME_EVENT_TYPES must equal the high_volume-classed subclasses."""
     discovered = {
         event_type
-        for cls in _walk_event_subclasses(BaseEvent)
+        for cls in walk_subclasses(BaseEvent)
         if (event_type := getattr(cls, "_event_type", None)) is not None
         and getattr(cls, "_volume_class", None) == "high_volume"
     }
@@ -197,8 +192,18 @@ def test_deriver_init_materializes_token_and_backlog():
                 )
                 is not None
             )
-    # dreamer specialists are derived from the concrete BaseSpecialist subclasses
+    # dreamer specialists are derived from the concrete BaseSpecialist subclasses.
-    for specialist_name in ("deduction", "induction"):
+    # Derived here too, rather than hardcoded: a hardcoded pair would keep passing
+    # when a third specialist is added (it only asserts presence), silently leaving
+    # the new one uncovered — which is exactly what happened when CardRefreshSpecialist
+    # landed.
+    specialist_names = {
+        name
+        for cls in walk_subclasses(BaseSpecialist)
+        if (name := getattr(cls, "name", None)) is not None
+    }
+    assert {"deduction", "induction", "card_refresh"} <= specialist_names
+    for specialist_name in specialist_names:
         assert (
             sample(
                 "dreamer_tokens_processed_total",
@@ -206,7 +211,7 @@ def test_deriver_init_materializes_token_and_backlog():
                 token_type=TokenTypes.INPUT.value,
             )
             is not None
-        )
+        ), f"specialist {specialist_name!r} was not zero-initialized"
     assert sample("message_embeddings_pending") == 0.0  # gauge zero-init