telemetry: materialize dropped-event counter children at 0

src/deriver/__main__.py

@@ -10,6 +10,7 @@
 from src.startup import validate_embedding_schema
 from src.telemetry import (
     initialize_telemetry_async,
+    prometheus_metrics,
     register_db_pool_collector,
     shutdown_telemetry,
 )
@@ -25,6 +26,10 @@ def start_metrics_server() -> None:
     # Expose DB connection-pool stats for this deriver instance.
     register_db_pool_collector("deriver")
     register_db_query_instrumentation("deriver")
+
+    # Pre-materialize bounded-label counter children at 0 so metrics are visible
+    # in Prometheus before the first event (no-op if metrics off).
+    prometheus_metrics.initialize_bounded_metrics(instance_type="deriver")
     logger.info("Prometheus metrics server started on port 9090")
 
 

src/main.py

@@ -136,6 +136,10 @@ async def lifespan(_: FastAPI):
     register_db_pool_collector("api")
     register_db_query_instrumentation("api")
 
+    # Pre-materialize bounded-label counter children at 0 so metrics are visible
+    # in Prometheus before the first event (no-op if metrics off).
+    prometheus_metrics.initialize_bounded_metrics(instance_type="api")
+
     # Validate embedding schema before serving any traffic. Fails closed: if
     # the configured EMBEDDING_VECTOR_DIMENSIONS does not match the physical
     # pgvector columns, the process refuses to start rather than silently

src/reconciler/sync_vectors.py

@@ -23,6 +23,7 @@
 from src.dependencies import tracked_db
 from src.embedding_client import embedding_client
 from src.exceptions import VectorStoreError
+from src.telemetry import prometheus_metrics
 from src.telemetry.events import EmbeddingCallPurpose
 from src.utils.types import embedding_call_purpose
 from src.vector_store import VectorRecord, VectorStore, get_external_vector_store
@@ -700,6 +701,30 @@ async def _cleanup_pgvector_batch(
         return True
 
 
+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
+    residual backlog after the sweep. Best-effort: a metrics/DB hiccup here must
+    never fail the reconciliation cycle.
+    """
+    if not settings.METRICS.ENABLED:
+        return
+    try:
+        async with tracked_db("reconciler_pending_count", read_only=True) as db:
+            count = await db.scalar(
+                select(func.count())
+                .select_from(models.MessageEmbedding)
+                .where(models.MessageEmbedding.sync_state == "pending")
+            )
+        prometheus_metrics.set_message_embeddings_pending(count=count or 0)
+    except Exception:
+        logger.warning(
+            "Failed to record pending-embeddings backlog gauge", exc_info=True
+        )
+
+
 async def run_vector_reconciliation_cycle() -> ReconciliationMetrics:
     """
     Run a complete reconciliation cycle.
@@ -729,6 +754,7 @@ 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
@@ -756,4 +782,5 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics:
             break
 
     logger.debug("Vector reconciliation cycle completed")
+    await _record_pending_embeddings_backlog()
     return metrics

src/telemetry/emitter.py

@@ -172,6 +172,20 @@ async def start(self) -> None:
         )
         self._running = True
         self._flush_task = asyncio.create_task(self._periodic_flush())
+
+        # Pre-create the dropped-event counter children at 0 so the metric is
+        # visible in Prometheus/Grafana before any drop occurs — a labeled
+        # counter exports nothing until its first observation. Lets us alert on
+        # telemetry drops and tell "no drops" apart from "metric missing".
+        from src.telemetry.prometheus.metrics import prometheus_metrics
+
+        prometheus_metrics.initialize_telemetry_dropped_metrics(
+            reasons=[
+                f"{self.drop_reason_prefix}buffer_full",
+                f"{self.drop_reason_prefix}send_failed",
+            ]
+        )
+
         logger.info("Telemetry emitter started, endpoint: %s", self.endpoint)
 
     async def shutdown(self) -> None:

src/telemetry/events/__init__.py

@@ -138,9 +138,64 @@
     # Lifecycle
     "initialize_telemetry_events",
     "shutdown_telemetry_events",
+    # Zero-init registry
+    "ALL_EVENT_TYPES",
+    "HIGH_VOLUME_EVENT_TYPES",
 ]
 
 
+# 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
+# .meta design telemetry-counter-zero-init).
+#
+# ⚠️ 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
+# test tests/telemetry/test_metric_zero_init.py fails until you do — it asserts
+# this registry equals the set discovered by walking BaseEvent subclasses.
+ALL_EVENT_TYPES: tuple[str, ...] = (
+    # api
+    "message.created",
+    "file.uploaded",
+    "context.retrieved",
+    # agent
+    "agent.iteration",
+    "agent.tool.conclusions.created",
+    "agent.tool.conclusions.deleted",
+    "agent.tool.peer_card.updated",
+    "agent.tool.summary.created",
+    "agent.tool.call.completed",
+    # deletion / dialectic / dream / representation
+    "deletion.completed",
+    "dialectic.completed",
+    "dream.run",
+    "dream.specialist",
+    "representation.completed",
+    # llm / embedding
+    "llm.call.completed",
+    "embedding.call.completed",
+    # reconciliation
+    "reconciliation.sync_vectors.completed",
+    "reconciliation.cleanup_stale_items.completed",
+    # trace stream (only emitted when TELEMETRY.TRACE_PAYLOADS_ENABLED, but they
+    # flow through the same emit() path and increment the same counters)
+    "llm.call.traced",
+    "embedding.call.traced",
+    "trace.content",
+)
+
+# Subset of ALL_EVENT_TYPES whose `_volume_class == "high_volume"` — only these
+# can ever be counted by `telemetry_events_sampled_out` (ground_truth events skip
+# the sampler entirely).
+HIGH_VOLUME_EVENT_TYPES: tuple[str, ...] = (
+    "agent.iteration",
+    "agent.tool.call.completed",
+    "llm.call.completed",
+    "embedding.call.completed",
+)
+
+
 def emit(event: BaseEvent) -> None:
     """Queue an event for emission to the telemetry backend.
 

src/telemetry/prometheus/metrics.py

@@ -5,7 +5,7 @@
 import logging
 from collections.abc import Iterator
 from enum import Enum
-from typing import cast, final
+from typing import cast, final, get_args
 
 from prometheus_client import (
     CONTENT_TYPE_LATEST,
@@ -20,7 +20,7 @@
 from starlette.requests import Request
 from starlette.responses import Response
 
-from src.config import settings
+from src.config import ReasoningLevel, settings
 
 disable_created_metrics()
 
@@ -66,6 +66,34 @@ class DialecticComponents(Enum):
     TOTAL = "total"
 
 
+# Bounded label domains used to zero-initialize counter children at startup (see
+# initialize_bounded_metrics). REASONING_LEVELS is derived from the config
+# Literal so it never drifts.
+REASONING_LEVELS: tuple[str, ...] = get_args(ReasoningLevel)
+
+# 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
+# (ingestion has no previous summary). Enumerating anything broader would
+# fabricate impossible always-0 series (e.g. output/prompt, or
+# ingestion/previous_summary). Explicit literal, drift-guarded by
+# tests/telemetry/test_metric_zero_init.py. Sources: track_deriver_input_tokens
+# (utils/tokens.py) + the OUTPUT_TOTAL sites in deriver.py / summarizer.py.
+_DERIVER_TOKEN_COMBOS_BY_TASK: dict[str, tuple[tuple[str, str], ...]] = {
+    DeriverTaskTypes.INGESTION.value: (
+        (TokenTypes.INPUT.value, DeriverComponents.PROMPT.value),
+        (TokenTypes.INPUT.value, DeriverComponents.MESSAGES.value),
+        (TokenTypes.OUTPUT.value, DeriverComponents.OUTPUT_TOTAL.value),
+    ),
+    DeriverTaskTypes.SUMMARY.value: (
+        (TokenTypes.INPUT.value, DeriverComponents.PROMPT.value),
+        (TokenTypes.INPUT.value, DeriverComponents.MESSAGES.value),
+        (TokenTypes.INPUT.value, DeriverComponents.PREVIOUS_SUMMARY.value),
+        (TokenTypes.OUTPUT.value, DeriverComponents.OUTPUT_TOTAL.value),
+    ),
+}
+
+
 api_requests_counter = NamespacedCounter(
     "api_requests",
     "Total API requests",
@@ -155,6 +183,16 @@ class DialecticComponents(Enum):
     ["namespace"],
 )
 
+# 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.
+message_embeddings_pending_gauge = NamespacedGauge(
+    "message_embeddings_pending",
+    "MessageEmbedding rows awaiting embedding (sync_state='pending')",
+    ["namespace"],
+)
+
 # DB connection-pool health. The in-flight gauge counts statements actually
 # executing on the wire, so checked_out minus in_flight reveals connections held
 # but parked (the "idle in transaction during an external call" antipattern).
@@ -325,12 +363,131 @@ def record_telemetry_event_dropped(self, *, reason: str) -> None:
         except Exception as e:
             self._handle_metric_error("record_telemetry_event_dropped", e)
 
+    def _touch(self, counter: NamespacedCounter, **labels: str) -> None:
+        """Pre-create a counter child series at 0 without incrementing it.
+
+        A labeled Prometheus counter exports no time series until its first
+        ``labels(...)`` call, so pre-touching a child keeps it present at 0 —
+        a missing series then signals a broken scrape rather than "no events".
+        Fail-soft (like the recorders): a bad init must never crash startup.
+        """
+        try:
+            counter.labels(**labels)
+        except Exception as e:
+            self._handle_metric_error("_touch", e)
+
+    def _set_gauge_zero(self, gauge: NamespacedGauge) -> None:
+        """Materialize a (namespace-only) gauge at 0. Fail-soft, like ``_touch``:
+        a startup init must never propagate an exception into process boot."""
+        try:
+            gauge.labels().set(0)
+        except Exception as e:
+            self._handle_metric_error("_set_gauge_zero", e)
+
+    def initialize_telemetry_dropped_metrics(self, *, reasons: list[str]) -> None:
+        """Pre-create telemetry_events_dropped child series at 0.
+
+        ``telemetry_events_dropped`` stays invisible in Prometheus/Grafana until
+        an event is actually dropped — you cannot alert on or graph a metric that
+        does not exist yet. Materializing the ``(namespace, reason)`` children at
+        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
+        values are prefix-dependent (the trace emitter uses a ``trace_`` prefix),
+        which the process-level ``initialize_bounded_metrics`` does not know.
+
+        Args:
+            reasons: The reason label values the calling emitter can produce.
+        """
+        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
+        label domain is bounded, enumerable at startup, and actually emitted by
+        THIS process are materialized; high-cardinality labels (endpoint,
+        workspace_name) and impossible label tuples are deliberately left absent.
+
+        ``telemetry_events_dropped`` is handled separately, per-emitter, in
+        ``TelemetryEmitter.start()`` (prefix-dependent — see above).
+
+        Args:
+            instance_type: "api" or "deriver" — selects the process-specific
+                counters. Event-type and buffer metrics are initialized in both.
+        """
+        if not settings.METRICS.ENABLED:
+            return
+
+        # Lazy import to avoid any import-time cycle (metrics is imported widely).
+        from src.telemetry.events import ALL_EVENT_TYPES, HIGH_VOLUME_EVENT_TYPES
+
+        # --- Common: both processes run a TelemetryEmitter, so both emit their
+        # own subset of event types. The domain is bounded/low-cardinality (~21
+        # types), so init the full set in each process rather than maintain a
+        # fragile per-event-type -> process map.
+        for event_type in ALL_EVENT_TYPES:
+            self._touch(telemetry_events_emitted_counter, type=event_type)
+        for event_type in HIGH_VOLUME_EVENT_TYPES:
+            self._touch(telemetry_events_sampled_out_counter, type=event_type)
+        self._set_gauge_zero(telemetry_buffer_size_gauge)
+
+        if instance_type == "api":
+            # dialectic tokens: token_type x component(total) x reasoning_level
+            for token_type in TokenTypes:
+                for level in REASONING_LEVELS:
+                    self._touch(
+                        dialectic_tokens_processed_counter,
+                        token_type=token_type.value,
+                        component=DialecticComponents.TOTAL.value,
+                        reasoning_level=level,
+                    )
+            # embed_now fast path runs as an API-process background task
+            self._touch(embed_now_tasks_shed_counter)
+            self._set_gauge_zero(embed_now_tasks_in_flight_gauge)
+
+        elif instance_type == "deriver":
+            # deriver tokens: only the VALID (token_type, component) tuples per
+            # task_type (see _DERIVER_TOKEN_COMBOS_BY_TASK) — never the cartesian
+            # product, which would fabricate impossible always-0 series.
+            for task_type_value, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items():
+                for token_type_value, component_value in combos:
+                    self._touch(
+                        deriver_tokens_processed_counter,
+                        task_type=task_type_value,
+                        token_type=token_type_value,
+                        component=component_value,
+                    )
+            # 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.
+            from src.dreamer.specialists import BaseSpecialist
+
+            for specialist in BaseSpecialist.__subclasses__():
+                for token_type in TokenTypes:
+                    self._touch(
+                        dreamer_tokens_processed_counter,
+                        specialist_name=specialist.name,
+                        token_type=token_type.value,
+                    )
+            # embedding backlog gauge — set live each reconciliation cycle; init
+            # at 0 so it's visible before the first cycle.
+            self._set_gauge_zero(message_embeddings_pending_gauge)
+
     def set_telemetry_buffer_size(self, *, size: int) -> None:
         try:
             telemetry_buffer_size_gauge.labels().set(size)
         except Exception as e:
             self._handle_metric_error("set_telemetry_buffer_size", e)
 
+    def set_message_embeddings_pending(self, *, count: int) -> None:
+        try:
+            message_embeddings_pending_gauge.labels().set(count)
+        except Exception as e:
+            self._handle_metric_error("set_message_embeddings_pending", e)
+
 
 prometheus_metrics = PrometheusMetrics()