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

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>
This commit is contained in:
Phil 2026-07-23 09:52:43 -04:00
parent fb8543a0c2
commit 01cd20aa91
2 changed files with 78 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -71,17 +71,27 @@ class DialecticComponents(Enum):
# Literal so it never drifts.
REASONING_LEVELS: tuple[str, ...] = get_args(ReasoningLevel)
# Valid (token_type, component) pairs for deriver_tokens_processed. NOT the full
# cartesian product: input tokens only have input components, output tokens only
# OUTPUT_TOTAL. Enumerating the cartesian product would fabricate impossible
# always-0 series (e.g. output/prompt). Kept as an explicit literal, drift-guarded
# by tests/telemetry/test_metric_zero_init.py.
_DERIVER_TOKEN_COMBOS: tuple[tuple[str, str], ...] = (
(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),
)
# Valid (token_type, component) pairs for deriver_tokens_processed, per task_type.
# NOT the cartesian product: input tokens only pair with input components, output
# only with OUTPUT_TOTAL, and PREVIOUS_SUMMARY occurs only for summary tasks
# (ingestion has no previous summary). Enumerating anything broader would
# fabricate impossible always-0 series (e.g. output/prompt, or
# ingestion/previous_summary). Explicit literal, drift-guarded by
# tests/telemetry/test_metric_zero_init.py. Sources: track_deriver_input_tokens
# (utils/tokens.py) + the OUTPUT_TOTAL sites in deriver.py / summarizer.py.
_DERIVER_TOKEN_COMBOS_BY_TASK: dict[str, tuple[tuple[str, str], ...]] = {
DeriverTaskTypes.INGESTION.value: (
(TokenTypes.INPUT.value, DeriverComponents.PROMPT.value),
(TokenTypes.INPUT.value, DeriverComponents.MESSAGES.value),
(TokenTypes.OUTPUT.value, DeriverComponents.OUTPUT_TOTAL.value),
),
DeriverTaskTypes.SUMMARY.value: (
(TokenTypes.INPUT.value, DeriverComponents.PROMPT.value),
(TokenTypes.INPUT.value, DeriverComponents.MESSAGES.value),
(TokenTypes.INPUT.value, DeriverComponents.PREVIOUS_SUMMARY.value),
(TokenTypes.OUTPUT.value, DeriverComponents.OUTPUT_TOTAL.value),
),
}
api_requests_counter = NamespacedCounter(
@ -366,6 +376,14 @@ class PrometheusMetrics:
except Exception as e:
self._handle_metric_error("_touch", e)
def _set_gauge_zero(self, gauge: NamespacedGauge) -> None:
"""Materialize a (namespace-only) gauge at 0. Fail-soft, like ``_touch``:
a startup init must never propagate an exception into process boot."""
try:
gauge.labels().set(0)
except Exception as e:
self._handle_metric_error("_set_gauge_zero", e)
def initialize_telemetry_dropped_metrics(self, *, reasons: list[str]) -> None:
"""Pre-create telemetry_events_dropped child series at 0.
@ -414,7 +432,7 @@ class PrometheusMetrics:
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)
telemetry_buffer_size_gauge.labels().set(0)
self._set_gauge_zero(telemetry_buffer_size_gauge)
if instance_type == "api":
# dialectic tokens: token_type x component(total) x reasoning_level
@ -428,16 +446,17 @@ class PrometheusMetrics:
)
# embed_now fast path runs as an API-process background task
self._touch(embed_now_tasks_shed_counter)
embed_now_tasks_in_flight_gauge.labels().set(0)
self._set_gauge_zero(embed_now_tasks_in_flight_gauge)
elif instance_type == "deriver":
# deriver tokens: only the VALID (task_type, token_type, component)
# tuples (see _DERIVER_TOKEN_COMBOS).
for task_type in DeriverTaskTypes:
for token_type_value, component_value in _DERIVER_TOKEN_COMBOS:
# deriver tokens: only the VALID (token_type, component) tuples per
# task_type (see _DERIVER_TOKEN_COMBOS_BY_TASK) — never the cartesian
# product, which would fabricate impossible always-0 series.
for task_type_value, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items():
for token_type_value, component_value in combos:
self._touch(
deriver_tokens_processed_counter,
task_type=task_type.value,
task_type=task_type_value,
token_type=token_type_value,
component=component_value,
)
@ -455,7 +474,7 @@ class PrometheusMetrics:
)
# embedding backlog gauge — set live each reconciliation cycle; init
# at 0 so it's visible before the first cycle.
message_embeddings_pending_gauge.labels().set(0)
self._set_gauge_zero(message_embeddings_pending_gauge)
def set_telemetry_buffer_size(self, *, size: int) -> None:
try:

55
tests/telemetry/test_metric_zero_init.py
View File

@ -19,7 +19,7 @@ from prometheus_client import REGISTRY
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, # pyright: ignore[reportPrivateUsage]
_DERIVER_TOKEN_COMBOS_BY_TASK, # pyright: ignore[reportPrivateUsage]
REASONING_LEVELS,
DeriverComponents,
DeriverTaskTypes,
@ -85,21 +85,35 @@ def test_high_volume_registry_matches_subclasses():
def test_deriver_token_combos_are_valid_and_complete():
"""Every combo uses real enum values, and every DeriverComponent is covered.
"""Every combo uses real enum values; the union across tasks covers every
DeriverComponent; and no task enumerates an impossible pair.
Fails if a DeriverComponent is added to the enum without deciding which
token_type it pairs with in _DERIVER_TOKEN_COMBOS.
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}
for token_type, component in _DERIVER_TOKEN_COMBOS:
assert token_type in valid_token_types
assert component in valid_components
# every component appears in exactly one combo
combo_components = {comp for _, comp in _DERIVER_TOKEN_COMBOS}
assert combo_components == valid_components
# the cartesian product would be larger — we intentionally enumerate fewer
assert len(_DERIVER_TOKEN_COMBOS) < len(valid_token_types) * len(valid_components)
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
# ---------------------------------------------------------------------------
@ -152,12 +166,12 @@ def test_sampled_out_excludes_ground_truth_event_types():
@pytest.mark.usefixtures("metrics_enabled")
def test_deriver_init_materializes_token_and_backlog():
prometheus_metrics.initialize_bounded_metrics(instance_type="deriver")
for task_type in DeriverTaskTypes:
for token_type, component in _DERIVER_TOKEN_COMBOS:
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.value,
task_type=task_type,
token_type=token_type,
component=component,
)
@ -181,6 +195,7 @@ 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",
@ -190,6 +205,16 @@ def test_deriver_init_omits_impossible_token_combos():
)
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(