RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

telemetry: zero-initialize bounded-label metrics so an absent series means a broken scrape (#927) 

* telemetry: materialize dropped-event counter children at 0

A labeled Prometheus counter exports no series until its first labels()
call, so telemetry_events_dropped stayed invisible until an event was
actually dropped — impossible to alert on or graph, and "no drops" was
indistinguishable from "metric missing / scrape broken".

Pre-create the (namespace, reason) children at 0 on emitter start, for
each reason the emitter can emit, so the metric is always present.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* telemetry: generalize counter zero-init to all bounded-label counters

Extends #927 (which zero-inited telemetry_events_dropped) to every counter
whose label domain is bounded and known at startup, so metrics are present in
Prometheus before their first event — a missing series then signals a broken
scrape rather than "nothing happened yet".

- add initialize_bounded_metrics(instance_type) on PrometheusMetrics; call it
  per-process from main.py (api) and deriver/__main__.py (deriver).
- extract a shared _touch() helper; refactor initialize_telemetry_dropped_metrics
  onto it (that one stays per-emitter in start() — it's prefix-dependent).
- explicit ALL_EVENT_TYPES / HIGH_VOLUME_EVENT_TYPES registry in telemetry.events,
  drift-guarded by tests that walk BaseEvent subclasses.
- only VALID (task_type, token_type, component) tuples for deriver_tokens (the
  cartesian product would fabricate impossible always-0 series); only high-volume
  event types for sampled_out; high-cardinality labels (endpoint, workspace_name)
  left open.
- gauges: zero-init embed_now_tasks_in_flight + telemetry_buffer_size; add a new
  message_embeddings_pending backlog gauge, set each reconciliation cycle and
  zero-inited at deriver startup (Rajat's pending/in-flight ask).
- backfills the tests #927 shipped without.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* review: task-aware deriver combos + fail-soft gauge zero-init

I1: _DERIVER_TOKEN_COMBOS was factored task-independently, materializing the
impossible (ingestion, input, previous_summary) series — previous_summary is
summary-only. Make combos task-aware (_DERIVER_TOKEN_COMBOS_BY_TASK) so no
always-0 impossible series is fabricated, matching the PR's own goal. Tests
tightened to assert the ingestion/previous_summary series is absent.

I2: the three gauge .set(0) zero-inits were bare while the counter inits go
through the fail-soft _touch. Add _set_gauge_zero() so a gauge init can't
propagate an exception into process startup either.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(telemetry): isolate zero-init namespaces, add deriver-to-api guard

Global-REGISTRY assertions used a fixed "test" namespace, which several
other suites also pin, so another test's materialized children could
satisfy a presence assertion or break an absence one. Each test now runs
under a unique namespace resolved from settings at read time.

Adds the inverse per-process isolation test: deriver-only init must not
materialize API-only series (dialectic tokens, embed_now).

Co-Authored-By: Claude <noreply@anthropic.com>

* 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).

* review: fix inert test guard, stale comments, and the REASONING_LEVELS drift claim

Second review pass on the branch. Findings, most severe first:

- tests/reconciler/test_pending_backlog_gauge.py: the _try_enqueue_task stub
  was patched onto the class but declared without `self`, so calling it
  raised TypeError — which _scheduler_loop swallows. The guard was inert and
  the test passed for the wrong reason. Fixed the arity.

- metrics.py still commented that the backlog gauge is "set live each
  reconciliation cycle". That is the exact claim the previous commit
  overturned; it now contradicted the help string, the bucket-3 docstring
  and sync_vectors.py.

- metrics.py claimed REASONING_LEVELS is "derived from the config Literal so
  it never drifts", but config.py hand-listed it, so the earlier dedup had
  quietly traded away the guarantee the original get_args() call provided.
  Made it true instead: config.REASONING_LEVELS = list(get_args(...)), which
  keeps the dedup and restores the invariant.

- dropped _set_gauge_zero: all three gauges it zeroed already have identical
  fail-soft setters, so it was a second way to do one thing. Using the
  setters also makes _handle_metric_error name the actual gauge.

- record_pending_embeddings_backlog's docstring oversold the covering index
  as making the COUNT "negligible". The index makes cost proportional to the
  pending backlog, not to the table — which is worst precisely when the
  backlog matters. Stated honestly.

- _scheduler_loop's docstring said it only enqueues; it also refreshes the
  gauge, at a cadence set by the shortest task interval.

- comment reconciliation: stripped #927 / "the generalization" temporal
  anchoring, a CardRefreshSpecialist change-narration clause, and
  reviewer-directed phrasing from the test file; disambiguated the
  src/utils/summarizer.py path.

- CLAUDE.md had no Prometheus section at all, so the new "add a BaseEvent
  subclass -> update ALL_EVENT_TYPES" obligation and the never-sum() rule
  for non-additive gauges were undiscoverable from the architecture doc.

Verified: ruff + basedpyright clean (0 errors), tests/telemetry + reconciler
+ dialectic + llm 497 passed, full suite 1768 passed with only the 4
pre-existing test_document failures (OpenAI key required, reproduced on
clean origin/main). Re-confirmed the relocation guard fails when reverted.

* fix: silence the two basedpyright warnings inherited from main

CI runs `uv run basedpyright` bare, and basedpyright exits non-zero on any
warning — so these two have been failing the staticanalysis job on every
branch cut from current main, not just this one:

- src/vector_store/__init__.py:209 implicit string concatenation (#496)
- tests/test_cache_redaction.py:5 private import (#869)

Both predate this branch and are unrelated to the telemetry work; fixed
here only because they block this PR from going green. Verified: clean
origin/main also reports "0 errors, 2 warnings" and exits 1.

basedpyright now 0 errors, 0 warnings, exit 0.

* docs(telemetry): make the bucket-3 aggregation rule precise

The multi-instance taxonomy said a service-scoped non-additive metric has
"no aggregation correct once they disagree", then immediately mandated that
every instance refresh on its own timer. Those undercut each other: staggered
timers ALWAYS disagree slightly, so as written the rule reads as "ensure they
don't", which is unachievable, and it leaves the reader unsure whether max()
and avg() survived the fix.

The actual rule is bounded disagreement plus a scale-preserving aggregator.
Instances are N witnesses to one fact, not N parts of one whole, so sum() can
never be correct (it scales with replica count) while max()/avg()/quantiles
are correct precisely because the per-instance timer bounds the spread.

Wording only; no behavior change. The gauge help string already said
"max() or avg(), never sum()" — this makes the normative docstring agree
with it. Surfaced walking Vineeth's comment 3668208059 for comprehension.

* refactor(bench): import REASONING_LEVELS from config instead of re-listing

Third copy of the constant, missed when ee781c0/694e07f deduped the other
two. This one re-declared the ReasoningLevel Literal as well as the list,
so the type alias could diverge from config's with nothing to catch it —
and the list was hand-written, the variant that typechecks clean while
missing a member.

No import barrier justified it: this module already imports from src, as do
seven of its siblings in tests/bench. Concrete effect of the drift was that
a newly added sixth reasoning level would be rejected by the bench CLI's
argparse choices=.

src.config.REASONING_LEVELS is now the single definition repo-wide.

* test(telemetry): pin the METRICS.ENABLED guard on the per-emitter initializer

initialize_telemetry_dropped_metrics gained a METRICS.ENABLED guard in
ee781c0, addressing Vineeth's asymmetry comment, but nothing asserted it —
it had only the enabled half of the pair its sibling has. Deleting the guard
left the suite green, so the fix closed the asymmetry in the guards and
reproduced it one level up in the tests.

Mirrors test_init_noop_when_metrics_disabled. Verified live rather than
assumed: deleting the two guard lines turns this test red.

Uses a unique namespace, without which the absence assertion would be
satisfied by the enabled test's children rather than by the guard.

* docs(telemetry): fold zero-init why-prose behind # region ai markers

Comment/docstring-only pass over the changed files, per the groudon
comment-marker standard: the terse human-facing "what" stays visible, and
load-bearing "why" (the zero-init / absent-series-means-broken-scrape
rationale, gotchas, receipts) folds into # region ai / # ai: blocks.

Behavior-preserving: AST-identical modulo docstrings/comments vs the
pre-pass merge; ruff, ruff format --check, and basedpyright all clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Phil 2026-08-20 10:24:49 -04:00 committed by GitHub
parent 7bafee5de1
commit 4797489281
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 807 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CLAUDE.md
View File

@ -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

17
src/config.py
View File

@ -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):

7
src/deriver/__main__.py
View File

@ -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")

6
src/main.py
View File

@ -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

16
src/reconciler/scheduler.py
View File

@ -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)

37
src/reconciler/sync_vectors.py
View File

@ -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.

15
src/telemetry/emitter.py
View File

@ -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:

65
src/telemetry/events/__init__.py
View File

@ -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.

193
src/telemetry/prometheus/metrics.py
View File

@ -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()

16
src/utils/types.py
View File

@ -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)

7
tests/bench/runner_common.py
View File

@ -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")

86
tests/reconciler/test_pending_backlog_gauge.py Normal file
View File

@ -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

357
tests/telemetry/test_metric_zero_init.py Normal file
View File

@ -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