diff --git a/CLAUDE.md b/CLAUDE.md index c9862b59..6a83066d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -187,6 +187,9 @@ The Dreamer is an orchestrated multi-specialist system that runs during schedule - **LLM subsystem** (`src/llm/`): provider-agnostic `honcho_llm_call()`. Backends in `src/llm/backends/` (`anthropic.py`, `gemini.py`, `openai.py`). Includes prompt caching (`caching.py`), structured output (`structured_output.py`), tool loop (`tool_loop.py`), history adapters for cross-provider message formats, and a model registry. Per-retry provider selection is pinned via an `AttemptPlan` so stream-final retries don't bounce back to primary after the tool loop has settled on fallback. - **Per-agent model config**: each agent has its own `MODEL_CONFIG` in `src/config.py` with fallback chains (see `ConfiguredModelSettings`, `FallbackModelSettings`). - **Telemetry**: cloudevents in `src/telemetry/events/` cover API routes, dialectic, dream, deletion, reconciliation, representation, and per-call LLM accounting (`llm.py` — `LLMCallCompletedEvent` fires once per provider hit with full cost-attribution context). High-volume events are sampled deterministically per `run_id` via `TelemetrySettings.HIGH_VOLUME_SAMPLE_RATE`. +- **Prometheus metrics** (`src/telemetry/prometheus/`): every metric carries a `namespace` label and every recorder is fail-soft (a metrics error never propagates into a request or a worker loop). Counter children with a *bounded* label domain are zero-initialized per process at startup — `initialize_bounded_metrics(instance_type=...)`, called from the `src/main.py` lifespan (`api`) and `src/deriver/__main__.py` (`deriver`) — so an absent series means a broken scrape rather than "nothing happened". Two consequences worth knowing before touching telemetry: + - **Adding a `BaseEvent` subclass requires adding its `_event_type` to `ALL_EVENT_TYPES`** in `src/telemetry/events/__init__.py` (and to `HIGH_VOLUME_EVENT_TYPES` if `_volume_class == "high_volume"`). Enforced by the drift guards in `tests/telemetry/test_metric_zero_init.py`, which assert set-equality against the discovered subclasses. + - **A service-wide, non-additive gauge must be refreshed by every replica on its own timer**, and aggregated with `max()`/`avg()`, never `sum()`. `message_embeddings_pending` is the example: it reports a DB-global count, so it is driven from `ReconcilerScheduler._scheduler_loop` (runs on all replicas) rather than from the work-unit-deduped reconciliation cycle — otherwise, combined with the zero-init, every replica that never won the work unit would export a confident permanent `0`. ### Project Structure diff --git a/src/config.py b/src/config.py index 092cb6ee..993f9bfb 100644 --- a/src/config.py +++ b/src/config.py @@ -2,7 +2,7 @@ import logging import math import os from pathlib import Path -from typing import Annotated, Any, ClassVar, Literal, cast +from typing import Annotated, Any, ClassVar, Literal, cast, get_args from urllib.parse import urlparse import tomllib @@ -998,15 +998,14 @@ class PeerCardSettings(HonchoSettings): ENABLED: bool = True -# Reasoning levels for dialectic - defined here to avoid circular imports with schemas +# Reasoning levels for dialectic - defined here to avoid circular imports with schemas. +# region ai +# REASONING_LEVELS is derived from the Literal, not hand-listed: the annotation +# rejects an invalid member but not a MISSING one, so a hand-written copy could +# silently drop a level and still typecheck. +# endregion ReasoningLevel = Literal["minimal", "low", "medium", "high", "max"] -REASONING_LEVELS: list[ReasoningLevel] = [ - "minimal", - "low", - "medium", - "high", - "max", -] +REASONING_LEVELS: list[ReasoningLevel] = list(get_args(ReasoningLevel)) class DialecticLevelSettings(BaseModel): diff --git a/src/deriver/__main__.py b/src/deriver/__main__.py index c56ed6a0..ce3969b5 100644 --- a/src/deriver/__main__.py +++ b/src/deriver/__main__.py @@ -10,6 +10,7 @@ from src.db import engine, register_db_query_instrumentation 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,12 @@ def start_metrics_server() -> None: # Expose DB connection-pool stats for this deriver instance. register_db_pool_collector("deriver") register_db_query_instrumentation("deriver") + + # region ai + # Zero-init bounded-label counters so a missing series signals a broken scrape, + # not "no events" — see initialize_bounded_metrics. No-op if metrics off. + # endregion + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") logger.info("Prometheus metrics server started on port 9090") diff --git a/src/main.py b/src/main.py index 75eac455..6930d4c1 100644 --- a/src/main.py +++ b/src/main.py @@ -110,6 +110,12 @@ async def lifespan(_: FastAPI): register_db_pool_collector("api") register_db_query_instrumentation("api") + # region ai + # Zero-init bounded-label counters so a missing series signals a broken scrape, + # not "no events" — see initialize_bounded_metrics. No-op if metrics off. + # endregion + 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 diff --git a/src/reconciler/scheduler.py b/src/reconciler/scheduler.py index e171c4fa..08941970 100644 --- a/src/reconciler/scheduler.py +++ b/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__) @@ -145,15 +146,26 @@ class ReconcilerScheduler: async def _scheduler_loop(self) -> None: """ - Main scheduler loop that enqueues tasks based on their intervals. + Main scheduler loop that enqueues tasks based on their intervals, and + refreshes the service-wide pending-embeddings backlog gauge each pass. Each task has its own interval and the loop checks all tasks on each - iteration, enqueueing any that are due. + iteration, enqueueing any that are due. The loop sleeps until the next + task is due, so the gauge's refresh cadence tracks the SHORTEST task + interval. """ try: while not self._shutdown_event.is_set(): now = datetime.now(timezone.utc) + # region ai + # Refresh on EVERY replica, not just whichever 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. Full rationale in record_pending_embeddings_backlog. + # endregion + 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) diff --git a/src/reconciler/sync_vectors.py b/src/reconciler/sync_vectors.py index 68e9ac59..1ff94542 100644 --- a/src/reconciler/sync_vectors.py +++ b/src/reconciler/sync_vectors.py @@ -23,6 +23,7 @@ from src.config import settings 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,42 @@ 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').""" + # region ai + # Called from ``ReconcilerScheduler._scheduler_loop``, deliberately NOT from + # ``run_vector_reconciliation_cycle``: the cycle runs off the queue behind + # 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, since + # this metric is zero-initialized, a confident permanent 0 it never measured. The + # count is a property of the database, not the process, so every replica must + # refresh it on its own timer for ``max()``/``avg()`` to mean anything. + # + # Cost: one COUNT per replica per scheduler interval (~5 min by default). + # ``ix_message_embeddings_sync_state_last_sync_at`` keeps the scan proportional to + # the pending backlog, not the whole table — which is not the same as cheap: after + # an embedding outage the backlog is exactly what is large. Still a small duty + # cycle, and the cost shrinks as the reconciler drains. + # + # Best-effort: a metrics/DB hiccup here must never break the scheduler loop. + # endregion + 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. diff --git a/src/telemetry/emitter.py b/src/telemetry/emitter.py index 201d16aa..397ecd90 100644 --- a/src/telemetry/emitter.py +++ b/src/telemetry/emitter.py @@ -172,6 +172,21 @@ class TelemetryEmitter: ) self._running = True self._flush_task = asyncio.create_task(self._periodic_flush()) + + # region ai + # Pre-create the dropped-event counter children at 0: a labeled counter + # exports nothing until its first observation, so this makes the metric + # visible before any drop and lets us tell "no drops" from "metric missing". + # endregion + 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: diff --git a/src/telemetry/events/__init__.py b/src/telemetry/events/__init__.py index 000e70b4..f8950d0f 100644 --- a/src/telemetry/events/__init__.py +++ b/src/telemetry/events/__init__.py @@ -138,9 +138,74 @@ __all__ = [ # 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 zero-initialize the +# telemetry_events_emitted / telemetry_events_sampled_out counter children. +# region ai +# See metrics.py:initialize_bounded_metrics for why absent and zero are worth +# distinguishing. Explicit literal, not a set derived from BaseEvent subclasses: a +# derived set would follow whatever happens to be imported at init time, so a type +# could drop out of the registry with no code change. A hand-maintained list plus a +# drift-guard test fails loud at the right moment instead. +# +# When you add a 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. +# endregion +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 + # region ai + # Only emitted when TELEMETRY.TRACE_PAYLOADS_ENABLED, but they flow through the + # same emit() path and increment the same counters, so they belong in the set. + # endregion + "llm.call.traced", + "embedding.call.traced", + "trace.content", +) + +# Subset of ALL_EVENT_TYPES whose `_volume_class == "high_volume"`. +# region ai +# Only these can ever be counted by ``telemetry_events_sampled_out``; ground_truth +# events skip the sampler entirely (pre-creating their sampled_out series would be a +# permanently-misleading 0). +# endregion +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. diff --git a/src/telemetry/prometheus/metrics.py b/src/telemetry/prometheus/metrics.py index 01d7be4f..749591d9 100644 --- a/src/telemetry/prometheus/metrics.py +++ b/src/telemetry/prometheus/metrics.py @@ -20,7 +20,8 @@ from prometheus_client.core import GaugeMetricFamily from starlette.requests import Request from starlette.responses import Response -from src.config import settings +from src.config import REASONING_LEVELS, settings +from src.utils.types import walk_subclasses disable_created_metrics() @@ -66,6 +67,32 @@ class DialecticComponents(Enum): TOTAL = "total" +# Valid (token_type, component) pairs for deriver_tokens_processed, per task_type, +# used to zero-initialize counter children (see initialize_bounded_metrics). +# region ai +# 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 (src/utils/tokens.py) + the OUTPUT_TOTAL sites +# in src/deriver/deriver.py and src/utils/summarizer.py. +# endregion +_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 +182,23 @@ telemetry_buffer_size_gauge = NamespacedGauge( ["namespace"], ) +# Embedding backlog: MessageEmbedding rows still awaiting a vector +# (sync_state='pending'). +# region ai +# Distinct from embed_now_tasks_in_flight (in-flight fast-path work in the API +# process) — this is the durable, DB-wide backlog the reconciler drains. Every +# deriver replica refreshes it on its own timer from +# ReconcilerScheduler._scheduler_loop, so replicas disagree by at most one interval. +# Service-wide, not per-process — hence the help string's "never sum()". +# endregion +message_embeddings_pending_gauge = NamespacedGauge( + "message_embeddings_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"], +) + # 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 +369,159 @@ class PrometheusMetrics: 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.""" + # region ai + # 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. + # endregion + try: + counter.labels(**labels) + except Exception as e: + self._handle_metric_error("_touch", e) + + def initialize_telemetry_dropped_metrics(self, *, reasons: list[str]) -> None: + """Pre-create telemetry_events_dropped ``(namespace, reason)`` children at 0. + + Args: + reasons: The reason label values the calling emitter can produce. + """ + # region ai + # The metric stays invisible in Prometheus/Grafana until an event is actually + # dropped, so materializing the children at startup keeps it present at 0 — a + # missing series then means a broken scrape, not "no drops" (see _touch). + # + # Called per-emitter from ``TelemetryEmitter.start()`` rather than hoisted into + # the process-level ``initialize_bounded_metrics``: the trace emitter (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. + # endregion + 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, so an + absent series means a broken scrape rather than "nothing happened". + + Args: + instance_type: "api" or "deriver" — selects the process-specific + counters. Event-type and buffer metrics are initialized in both. + """ + # region ai + # A Prometheus counter does not exist until its first increment, so a + # never-yet-incremented metric is indistinguishable from a broken scrape: + # you cannot graph or alert on a series that is absent. Materializing the + # 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 reports the whole service's + # value, so the instances are N witnesses to one fact rather than N parts of + # one whole. ``sum()`` is therefore never correct here: it scales with the + # replica count. Scale-preserving aggregations (``max()``, ``avg()``, + # quantiles) ARE correct, but only while the witnesses disagree by a bounded + # amount — which requires every instance to refresh on its own timer (see + # ``message_embeddings_pending``, refreshed per replica from + # ``ReconcilerScheduler._scheduler_loop``). A bucket-3 metric that cannot + # meet that bar does not belong in the app at all — it belongs in an exporter + # that yields exactly 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). + # endregion + if not settings.METRICS.ENABLED: + return + + # ai: lazy import avoids an import-time cycle (metrics is imported widely) + from src.telemetry.events import ALL_EVENT_TYPES, HIGH_VOLUME_EVENT_TYPES + + # region ai + # 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. + # endregion + 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_telemetry_buffer_size(size=0) + + 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, + ) + # ai: embed_now fast path runs as an API-process background task + self._touch(embed_now_tasks_shed_counter) + self.set_embed_now_tasks_in_flight(0) + + elif instance_type == "deriver": + # deriver tokens: only the valid (token_type, component) tuples per + # task_type (see _DERIVER_TOKEN_COMBOS_BY_TASK). + 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. + # region ai + # Names come from the concrete BaseSpecialist subclasses (walked recursively + # via walk_subclasses) so a new specialist can't silently miss init. + # endregion + from src.dreamer.specialists import BaseSpecialist + + for specialist in walk_subclasses(BaseSpecialist): + for token_type in TokenTypes: + self._touch( + dreamer_tokens_processed_counter, + specialist_name=specialist.name, + token_type=token_type.value, + ) + # ai: init at 0 so the gauge is visible before its first per-replica refresh + self.set_message_embeddings_pending(count=0) + 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() diff --git a/src/utils/types.py b/src/utils/types.py index 24a13ae6..0bfada75 100644 --- a/src/utils/types.py +++ b/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.""" + # region ai + # ``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. + # endregion + 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) diff --git a/tests/bench/runner_common.py b/tests/bench/runner_common.py index be262cb0..b150a3f8 100644 --- a/tests/bench/runner_common.py +++ b/tests/bench/runner_common.py @@ -17,21 +17,18 @@ from dataclasses import dataclass, field from datetime import datetime from logging import Logger from pathlib import Path -from typing import Any, Generic, Literal, TypeVar, cast +from typing import Any, Generic, TypeVar, cast from anthropic import AsyncAnthropic from honcho import Honcho from honcho.api_types import SessionConfiguration, SummaryConfiguration from openai import AsyncOpenAI +from src.config import REASONING_LEVELS, ReasoningLevel from src.telemetry.metrics_collector import MetricsCollector _logger = logging.getLogger(__name__) -# Valid reasoning levels for dialectic chat -ReasoningLevel = Literal["minimal", "low", "medium", "high", "max"] -REASONING_LEVELS: list[str] = ["minimal", "low", "medium", "high", "max"] - # Type variable for result types ResultT = TypeVar("ResultT") diff --git a/tests/reconciler/test_pending_backlog_gauge.py b/tests/reconciler/test_pending_backlog_gauge.py new file mode 100644 index 00000000..226a71e8 --- /dev/null +++ b/tests/reconciler/test_pending_backlog_gauge.py @@ -0,0 +1,86 @@ +"""The pending-embeddings backlog gauge must be refreshed per-replica. + +These tests pin both halves: the scheduler loop drives the refresh, and the +queue-driven reconciliation cycle does not. +""" +# region ai +# ``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. +# So the count is driven from ``ReconcilerScheduler._scheduler_loop`` (runs on every +# replica, every interval), NOT from ``run_vector_reconciliation_cycle`` (runs off +# the queue behind work-unit dedup, so exactly one replica per cycle executes it). +# endregion + +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(): # pyright: ignore[reportUnusedFunction] + 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() + + # region ai + # Patched onto the class, so it is invoked as a bound method — it needs the + # ``self`` parameter or the call raises TypeError, which ``_scheduler_loop`` would + # then swallow, leaving this guard silently inert. + # endregion + async def _never_enqueue(_self: object, _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.""" + # region ai + # 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. + # endregion + 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 diff --git a/tests/telemetry/test_metric_zero_init.py b/tests/telemetry/test_metric_zero_init.py new file mode 100644 index 00000000..e69f50df --- /dev/null +++ b/tests/telemetry/test_metric_zero_init.py @@ -0,0 +1,357 @@ +"""Tests for startup zero-initialization of bounded-label metrics. + +Asserts that: +- bounded-label counter children are materialized at 0 before any event, +- high-cardinality / impossible label combinations are deliberately NOT, +- per-process init doesn't materialize the other process's counters, +- the explicit registries stay in sync with the source of truth (drift guards). +""" +# region ai +# Reads use ``REGISTRY.get_sample_value`` (returns the value if a series exists, +# ``None`` if it does not) rather than ``counter.labels(...)``, because ``.labels`` +# would itself materialize the child and destroy the presence/absence signal. +# endregion + +from collections.abc import Iterator +from typing import cast +from uuid import uuid4 + +import pytest +from prometheus_client import REGISTRY + +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] + DeriverComponents, + DeriverTaskTypes, + DialecticComponents, + TokenTypes, + prometheus_metrics, +) +from src.utils.types import walk_subclasses + + +def unique_ns(tag: str) -> str: + """A ``namespace`` label value no other test can have materialized under.""" + # region ai + # Every assertion here reads the process-global ``REGISTRY``, which keeps a child + # series for the rest of the session once anything materializes it. A shared + # namespace (several other suites pin ``"test"``) would let another test's children + # satisfy a presence assertion, or break an absence assertion, independently of + # what the initializer under test actually did. + # endregion + return f"test_metric_zero_init_{tag}_{uuid4().hex[:8]}" + + +@pytest.fixture +def metrics_enabled(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]: + """Enable metrics under a namespace unique to the requesting test.""" + ns = unique_ns("enabled") + monkeypatch.setattr("src.config.settings.METRICS.ENABLED", True) + monkeypatch.setattr("src.config.settings.METRICS.NAMESPACE", ns) + yield ns + + +def sample(name: str, **labels: str) -> float | None: + """Value of a series if it exists, else None. Never materializes it. + + Resolves the namespace from settings, so it always reads the unique one the + active test pinned. + """ + ns = cast(str, settings.METRICS.NAMESPACE) + return REGISTRY.get_sample_value(name, {"namespace": ns, **labels}) + + +# --------------------------------------------------------------------------- +# Drift guards (pure logic — no registry). Adding an event type / token component +# without updating the registry fails here, with a pointer to what to fix. +# --------------------------------------------------------------------------- + + +def test_all_event_types_registry_matches_subclasses(): + """ALL_EVENT_TYPES must equal every BaseEvent subclass's _event_type. + + If this fails you added/removed a BaseEvent subclass without updating + ALL_EVENT_TYPES in src/telemetry/events/__init__.py — its Prometheus counter + would not be zero-initialized. Update the registry. + """ + discovered = { + event_type + for cls in walk_subclasses(BaseEvent) + if (event_type := getattr(cls, "_event_type", None)) is not None + } + assert set(ALL_EVENT_TYPES) == discovered + assert len(ALL_EVENT_TYPES) == len(set(ALL_EVENT_TYPES)), "duplicate event types" + + +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_subclasses(BaseEvent) + if (event_type := getattr(cls, "_event_type", None)) is not None + and getattr(cls, "_volume_class", None) == "high_volume" + } + assert set(HIGH_VOLUME_EVENT_TYPES) == discovered + assert set(HIGH_VOLUME_EVENT_TYPES) <= set(ALL_EVENT_TYPES) + + +def test_deriver_token_combos_are_valid_and_complete(): + """Every combo uses real enum values; the union across tasks covers every + DeriverComponent; and no task enumerates an impossible pair. + + Fails if a DeriverComponent/DeriverTaskType is added without deciding which + task_type + token_type it pairs with in _DERIVER_TOKEN_COMBOS_BY_TASK. + """ + valid_token_types = {t.value for t in TokenTypes} + valid_components = {c.value for c in DeriverComponents} + valid_task_types = {t.value for t in DeriverTaskTypes} + + assert set(_DERIVER_TOKEN_COMBOS_BY_TASK) == valid_task_types + all_components: set[str] = set() + for task_type, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items(): + assert task_type in valid_task_types + for token_type, component in combos: + assert token_type in valid_token_types + assert component in valid_components + # each task enumerates fewer than its cartesian product (no impossible pairs) + assert len(combos) < len(valid_token_types) * len(valid_components) + all_components.update(comp for _, comp in combos) + + # every component is reachable via some task + assert all_components == valid_components + # previous_summary is summary-only: ingestion must NOT enumerate it + ingestion = _DERIVER_TOKEN_COMBOS_BY_TASK[DeriverTaskTypes.INGESTION.value] + assert ( + TokenTypes.INPUT.value, + DeriverComponents.PREVIOUS_SUMMARY.value, + ) not in ingestion + + +# --------------------------------------------------------------------------- +# API-process zero-init +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_api_init_materializes_event_type_children(): + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + for event_type in ALL_EVENT_TYPES: + assert sample("telemetry_events_emitted_total", type=event_type) is not None + for event_type in HIGH_VOLUME_EVENT_TYPES: + assert sample("telemetry_events_sampled_out_total", type=event_type) is not None + + +@pytest.mark.usefixtures("metrics_enabled") +def test_api_init_materializes_dialectic_and_embed(): + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + for token_type in TokenTypes: + for level in REASONING_LEVELS: + assert ( + sample( + "dialectic_tokens_processed_total", + token_type=token_type.value, + component=DialecticComponents.TOTAL.value, + reasoning_level=level, + ) + is not None + ) + assert sample("embed_now_tasks_shed_total") is not None + assert sample("embed_now_tasks_in_flight") == 0.0 # gauge, explicit .set(0) + + +@pytest.mark.usefixtures("metrics_enabled") +def test_sampled_out_excludes_ground_truth_event_types(): + """Ground-truth events can never be sampled out, so their sampled_out series + must NOT be pre-created (they'd be permanently misleading zeros).""" + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + ground_truth = set(ALL_EVENT_TYPES) - set(HIGH_VOLUME_EVENT_TYPES) + for event_type in ground_truth: + assert sample("telemetry_events_sampled_out_total", type=event_type) is None + + +# --------------------------------------------------------------------------- +# Deriver-process zero-init +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_deriver_init_materializes_token_and_backlog(): + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + for task_type, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items(): + for token_type, component in combos: + assert ( + sample( + "deriver_tokens_processed_total", + task_type=task_type, + token_type=token_type, + component=component, + ) + is not None + ) + # region ai + # Specialist names are derived from the concrete BaseSpecialist subclasses here + # too, rather than hardcoded: a hardcoded list would keep passing when a new + # specialist is added (it only asserts presence), silently leaving it uncovered. + # endregion + 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", + specialist_name=specialist_name, + 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 + + +@pytest.mark.usefixtures("metrics_enabled") +def test_deriver_init_omits_impossible_token_combos(): + """The cartesian product includes combos that never occur (e.g. output tokens + with an input component). Those must not be materialized.""" + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + # output tokens never pair with an input component + assert ( + sample( + "deriver_tokens_processed_total", + task_type=DeriverTaskTypes.INGESTION.value, + token_type=TokenTypes.OUTPUT.value, + component=DeriverComponents.PROMPT.value, + ) + is None + ) + # previous_summary is summary-only — ingestion must not materialize it + assert ( + sample( + "deriver_tokens_processed_total", + task_type=DeriverTaskTypes.INGESTION.value, + token_type=TokenTypes.INPUT.value, + component=DeriverComponents.PREVIOUS_SUMMARY.value, + ) + is None + ) + # base specialist is abstract and never emits — must not be materialized + assert ( + sample( + "dreamer_tokens_processed_total", + specialist_name="base", + token_type=TokenTypes.INPUT.value, + ) + is None + ) + + +# --------------------------------------------------------------------------- +# High-cardinality counters are left open, and per-process isolation holds +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_high_cardinality_counters_not_materialized(): + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + # no endpoint/workspace_name series fabricated + assert ( + sample( + "api_requests_total", + method="GET", + endpoint="/v3/does-not-exist", + status_code="200", + ) + is None + ) + assert sample("messages_created_total", workspace_name="nope_ws") is None + + +@pytest.mark.usefixtures("metrics_enabled") +def test_api_init_does_not_touch_deriver_counters(): + """api-only init must not materialize or change a deriver-only counter. + + Delta-based (before == after) so it's robust to prior tests having + materialized the series. + """ + labels = dict( + task_type=DeriverTaskTypes.INGESTION.value, + token_type=TokenTypes.INPUT.value, + component=DeriverComponents.PROMPT.value, + ) + before = sample("deriver_tokens_processed_total", **labels) + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + after = sample("deriver_tokens_processed_total", **labels) + assert before == after + + +@pytest.mark.usefixtures("metrics_enabled") +def test_deriver_init_does_not_touch_api_counters(): + """The inverse: deriver-only init must not materialize an API-only counter. + + Without this, a deriver-startup regression could silently fabricate API + series (permanently-0 dialectic tokens on a process that never serves chat). + """ + labels = dict( + token_type=TokenTypes.INPUT.value, + component=DialecticComponents.TOTAL.value, + reasoning_level=REASONING_LEVELS[0], + ) + before = sample("dialectic_tokens_processed_total", **labels) + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + after = sample("dialectic_tokens_processed_total", **labels) + assert before == after + # the API-process embed_now counters are equally off-limits + assert sample("embed_now_tasks_shed_total") is None + assert sample("embed_now_tasks_in_flight") is None + + +# --------------------------------------------------------------------------- +# telemetry_events_dropped: per-emitter child materialization +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_dropped_counter_children_materialized(): + prometheus_metrics.initialize_telemetry_dropped_metrics( + reasons=["buffer_full", "send_failed"] + ) + assert sample("telemetry_events_dropped_total", reason="buffer_full") is not None + assert sample("telemetry_events_dropped_total", reason="send_failed") is not None + + +def test_dropped_counter_init_noop_when_metrics_disabled( + monkeypatch: pytest.MonkeyPatch, +): + """The per-emitter initializer must no-op when metrics are disabled.""" + # region ai + # The enabled/disabled pair above and below this line exists for + # ``initialize_bounded_metrics`` (see ``test_init_noop_when_metrics_disabled``); + # without this test the sibling initializer had only the enabled half, so its + # ``METRICS.ENABLED`` guard could be deleted with the suite staying green. The + # unique namespace is what makes the absence assertion mean anything — the enabled + # test above materializes these same two reason values under a different one. + # endregion + monkeypatch.setattr("src.config.settings.METRICS.ENABLED", False) + monkeypatch.setattr( + "src.config.settings.METRICS.NAMESPACE", unique_ns("dropped_disabled") + ) + prometheus_metrics.initialize_telemetry_dropped_metrics( + reasons=["buffer_full", "send_failed"] + ) + assert sample("telemetry_events_dropped_total", reason="buffer_full") is None + assert sample("telemetry_events_dropped_total", reason="send_failed") is None + + +def test_init_noop_when_metrics_disabled(monkeypatch: pytest.MonkeyPatch): + """With metrics disabled, init must not fabricate series for a fresh label.""" + monkeypatch.setattr("src.config.settings.METRICS.ENABLED", False) + monkeypatch.setattr("src.config.settings.METRICS.NAMESPACE", unique_ns("disabled")) + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + assert sample("telemetry_events_emitted_total", type="message.created") is None