Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions src/openarmature/observability/otel/observer.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,18 +220,20 @@ def _token_budget_evaluations(token_budget: Any, usage: Any) -> list[dict[str, A
# DECLARED bound that also has a usable actual to compare against, so a bound
# whose count the provider did not report (None) is omitted -- mirroring the
# token.usage gate rather than coercing a missing count to 0. A declared
# bound of 0 is degenerate (ratio undefined) and is skipped for both
# instruments; its declared value still surfaces as a prompt span attribute.
# Empty when there is no budget, no usage, or no evaluable bound; the caller
# then leaves the exceeded signal absent (not false).
# bound of 0 IS evaluated: the exceeded test is a strict ``actual > max``, so
# a 0 bound is exceeded by any positive usage per §5.5.15; only the
# utilization ratio is undefined for a 0 denominator, and the metric recorder
# skips that one sample. Empty when there is no budget, no usage, or no bound
# with a reported count; the caller then leaves the exceeded signal absent
# (not false).
if token_budget is None or usage is None:
return []
evaluations: list[dict[str, Any]] = []
input_max = getattr(token_budget, "input_max_tokens", None)
if input_max and usage.prompt_tokens is not None:
if input_max is not None and usage.prompt_tokens is not None:
evaluations.append({"kind": "input", "actual": usage.prompt_tokens, "max": input_max})
total_max = getattr(token_budget, "total_max_tokens", None)
if total_max:
if total_max is not None:
if usage.total_tokens is not None:
total_actual = usage.total_tokens
elif usage.prompt_tokens is not None and usage.completion_tokens is not None:
Expand Down Expand Up @@ -1518,7 +1520,11 @@ def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None:
):
for ev in _token_budget_evaluations(event.token_budget, usage):
dims = {**base_dims, "openarmature.gen_ai.token_budget.kind": ev["kind"]}
self._token_budget_utilization_histogram.record(ev["actual"] / ev["max"], dims)
# The utilization ratio is undefined for a 0 denominator; skip
# that one sample. A 0 bound is still counted as exceeded below
# (strict actual > max), per §5.5.15.
if ev["max"] > 0:
self._token_budget_utilization_histogram.record(ev["actual"] / ev["max"], dims)
if ev["actual"] > ev["max"]:
self._token_budget_exceeded_counter.add(1, dims)

Expand Down
58 changes: 58 additions & 0 deletions tests/unit/test_observability_otel.py
Original file line number Diff line number Diff line change
Expand Up @@ -1419,6 +1419,64 @@ async def test_token_budget_missing_input_count_omits_evaluation() -> None:
assert [p for p in points if "token_budget" in p[0]] == []


async def test_token_budget_zero_bound_exceeds_but_skips_utilization() -> None:
# §5.5.15 (proposal 0083): a declared bound of 0 is exceeded by any positive
# usage (the exceeded test is a strict actual > max), so the exceeded span
# attr + counter fire. The utilization ratio is undefined for a 0 denominator,
# so that one histogram sample is skipped -- not a fabricated sentinel.
from openarmature.llm.response import Usage
from openarmature.prompts import TokenBudget
from tests._helpers.typed_event import make_retry_attempt_event

event = make_retry_attempt_event(
model="test-model",
provider="openai",
usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6),
token_budget=TokenBudget(input_max_tokens=0),
)
points, llm_spans = await _drive_metrics_events([event])

attrs: dict[str, Any] = dict(llm_spans[0].attributes or {})
assert attrs.get("openarmature.prompt.token_budget.input_max_tokens") == 0
assert attrs.get("openarmature.llm.token_budget.exceeded") is True

exceeded = [p for p in points if p[0] == _TB_EXCEEDED]
util = [p for p in points if p[0] == _TB_UTILIZATION]
assert len(exceeded) == 1
assert exceeded[0][1] == 1
assert exceeded[0][3]["openarmature.gen_ai.token_budget.kind"] == "input"
# No utilization sample -- the 0-denominator ratio is undefined and skipped.
assert util == []


async def test_token_budget_zero_total_bound_exceeds_but_skips_utilization() -> None:
# §5.5.15 (proposal 0083): the total branch mirrors the input branch -- a
# total_max of 0 is exceeded by any positive total usage (exceeded attr +
# counter fire, kind total), and the undefined 0-denominator utilization
# sample is skipped.
from openarmature.llm.response import Usage
from openarmature.prompts import TokenBudget
from tests._helpers.typed_event import make_retry_attempt_event

event = make_retry_attempt_event(
model="test-model",
provider="openai",
usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6),
token_budget=TokenBudget(total_max_tokens=0),
)
points, llm_spans = await _drive_metrics_events([event])

attrs: dict[str, Any] = dict(llm_spans[0].attributes or {})
assert attrs.get("openarmature.prompt.token_budget.total_max_tokens") == 0
assert attrs.get("openarmature.llm.token_budget.exceeded") is True

exceeded = [p for p in points if p[0] == _TB_EXCEEDED]
util = [p for p in points if p[0] == _TB_UTILIZATION]
assert len(exceeded) == 1
assert exceeded[0][3]["openarmature.gen_ai.token_budget.kind"] == "total"
assert util == []


async def test_llm_span_zero_duration_when_latency_missing() -> None:
# When the typed event omits latency_ms (None), the handler falls
# back to a zero-duration span at end_time rather than guessing
Expand Down