Add 0083 token-budget signal (completion path) (#235)

* Add 0083 token-budget signal (completion path)

Implement the completion-path half of proposal 0083 (per-prompt
token-budget observability): the advisory Prompt.token_budget data
model, its carriage on the LLM events, and the OTel rendering.

- prompts: TokenBudget sub-record on Prompt/PromptResult, sourced
  from the same config sidecar as sampling; propagated on render.
- events: token_budget on the completion / failed / retry-attempt
  events, populated from the active prompt at dispatch.
- otel: openarmature.prompt.token_budget.* attrs + the reactive
  openarmature.llm.token_budget.exceeded signal, and two opt-in
  metric instruments (exceeded counter + utilization histogram,
  kind dimension), evaluated against reported usage.
- conformance: un-defer fixtures 126-129 (completion path); the
  Langfuse WARNING (130) and failure path (131) follow in PR B/C.

* Refine 0083 harness collapse and error message

Address PR #235 review threads:

- test harness: _render_prompt_result collapses an empty / all-null
  token_budget to None, mirroring the filesystem and Langfuse
  backends so conformance rendering matches real backend behavior.
- filesystem: the malformed-token_budget PromptStoreUnavailable
  message now includes the (name, label) source, matching the
  other filesystem backend errors.

Author: chris-colinsky

conformance.toml

@@ -826,7 +826,9 @@ note = "The graph-engine §6 event surface + llm-provider §7 error surface (PR
 
 # Spec v0.78.0 (proposal 0083).  Per-prompt token-budget observability
 # (prompt-management §3 / graph-engine §6 / observability §5.5.15).
-# Not-yet; observability fixtures 126-131 defer with it.
+# Not-yet overall: the completion-path OTel rendering + fixtures 126-129 run
+# (PR A), but the Langfuse WARNING (130) and structured-output-failure (131)
+# paths stay deferred until their PRs; flips to implemented when they land.
 [proposals."0083"]
 status = "not-yet"
 

src/openarmature/graph/events.py

@@ -583,6 +583,14 @@ class LlmCompletionEvent:
     active_prompt_group: Any
     call_id: str
     caller_invocation_metadata: Mapping[str, AttributeValue] | None = None
+    # Proposal 0083 (graph-engine §6 / observability §5.5.15): the active
+    # prompt's advisory per-prompt token budget at LLM-call time (the
+    # Prompt.token_budget snapshot). None when the call ran outside a
+    # budgeted prompt-context binding. Typed as Any -- same rationale as
+    # active_prompt (the prompts package isn't imported here); the observer
+    # reads input_max_tokens / total_max_tokens off it directly and compares
+    # them reactively against the reported usage to drive the budget signals.
+    token_budget: Any = None
     # Proposal 0076 (spec v0.67.0): the assistant message's output tool
     # calls in typed-event-native form (the ToolCall records, not a
     # pre-serialized shape — they carry no inline-image bytes, so the
@@ -694,6 +702,13 @@ class LlmFailedEvent:
     usage: "Usage | None" = None
     response_id: str | None = None
     response_model: str | None = None
+    # Proposal 0083 (graph-engine §6 / observability §5.5.15): the active
+    # prompt's advisory per-prompt token budget at LLM-call time. None when
+    # the call ran outside a budgeted prompt-context binding. Typed as Any --
+    # same rationale as active_prompt. On a structured_output_invalid failure
+    # (which carries the response-side usage) the budget eval fires as it does
+    # on a completion; None-carrying every other category leaves it inert.
+    token_budget: Any = None
 
 
 # Python-internal per-attempt LLM event. NOT a spec-normative event type
@@ -770,6 +785,13 @@ class LlmRetryAttemptEvent:
     error_message: str | None = None
     error_type: str | None = None
     caller_invocation_metadata: Mapping[str, AttributeValue] | None = None
+    # Proposal 0083 (graph-engine §6 / observability §5.5.15): the active
+    # prompt's advisory per-prompt token budget at LLM-call time. None outside
+    # a budgeted prompt-context binding. Typed as Any -- same rationale as
+    # active_prompt. This per-attempt event is the OTel budget-signal source:
+    # the observer compares its bounds against event.usage on every attempt
+    # carrying usage (success OR a structured_output_invalid failure).
+    token_budget: Any = None
     # Proposal 0076: the attempt's output tool calls (ToolCall records),
     # mirroring LlmCompletionEvent.output_tool_calls. Populated on a
     # successful attempt; empty list on a failed one (no response). The

src/openarmature/llm/providers/openai.py

@@ -513,6 +513,7 @@ def emit_attempt(
                         request_extras=request_extras,
                         active_prompt=active_prompt,
                         active_prompt_group=active_prompt_group,
+                        token_budget=(active_prompt.token_budget if active_prompt is not None else None),
                         response=attempt_response,
                         exc=attempt_exc,
                     ),
@@ -541,6 +542,7 @@ def emit_attempt(
                         request_extras=request_extras,
                         active_prompt=active_prompt,
                         active_prompt_group=active_prompt_group,
+                        token_budget=(active_prompt.token_budget if active_prompt is not None else None),
                     ),
                 )
             raise
@@ -562,6 +564,7 @@ def emit_attempt(
                     request_extras=request_extras,
                     active_prompt=active_prompt,
                     active_prompt_group=active_prompt_group,
+                    token_budget=(active_prompt.token_budget if active_prompt is not None else None),
                 ),
             )
         return response
@@ -645,6 +648,7 @@ def _build_llm_completion_event(
         request_extras: dict[str, Any],
         active_prompt: Any,
         active_prompt_group: Any,
+        token_budget: Any,
     ) -> LlmCompletionEvent:
         """Construct the typed LlmCompletionEvent for the success path.
 
@@ -714,6 +718,9 @@ def _build_llm_completion_event(
             request_extras=request_extras,
             active_prompt=active_prompt,
             active_prompt_group=active_prompt_group,
+            # Proposal 0083: the active prompt's advisory token budget, sourced
+            # by the caller from active_prompt.token_budget at dispatch time.
+            token_budget=token_budget,
             call_id=call_id,
             caller_invocation_metadata=caller_metadata,
         )
@@ -729,6 +736,7 @@ def _build_llm_failed_event(
         request_extras: dict[str, Any],
         active_prompt: Any,
         active_prompt_group: Any,
+        token_budget: Any,
     ) -> LlmFailedEvent:
         """Construct the typed LlmFailedEvent for the failure path.
 
@@ -792,6 +800,11 @@ def _build_llm_failed_event(
             request_extras=request_extras,
             active_prompt=active_prompt,
             active_prompt_group=active_prompt_group,
+            # Proposal 0083: advisory budget carried on the failure event too;
+            # None-carrying keeps it inert for non-response failures, while a
+            # structured_output_invalid failure (which does carry usage) drives
+            # the same reactive budget eval as a completion.
+            token_budget=token_budget,
             call_id=call_id,
             error_category=exc.category,
             error_type=type(exc).__name__,
@@ -815,6 +828,7 @@ def _build_llm_retry_attempt_event(
         request_extras: dict[str, Any],
         active_prompt: Any,
         active_prompt_group: Any,
+        token_budget: Any,
         response: Response | None = None,
         exc: LlmProviderError | None = None,
     ) -> LlmRetryAttemptEvent:
@@ -853,6 +867,8 @@ def _build_llm_retry_attempt_event(
             "request_extras": request_extras,
             "active_prompt": active_prompt,
             "active_prompt_group": active_prompt_group,
+            # Proposal 0083: advisory per-prompt token budget snapshot.
+            "token_budget": token_budget,
             "caller_invocation_metadata": caller_metadata,
         }
         if response is not None:

src/openarmature/observability/otel/observer.py

@@ -83,7 +83,7 @@
 from opentelemetry import context as otel_context
 from opentelemetry import metrics as otel_metrics
 from opentelemetry import trace as otel_trace
-from opentelemetry.metrics import Histogram, Meter, MeterProvider
+from opentelemetry.metrics import Counter, Histogram, Meter, MeterProvider
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
 from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
@@ -195,6 +195,52 @@
     40.96,
     81.92,
 ]
+# §11.2 (proposal 0083) token-budget utilization histogram bucket advisory.
+# Unitless ratio (actual / declared bound) so the boundaries cluster around
+# 1.0 -- the crossover between under- and over-budget -- to resolve how close
+# prompts run to (and how far past) their declared budget.
+_TOKEN_BUDGET_UTILIZATION_BUCKETS: list[float] = [
+    0.1,
+    0.25,
+    0.5,
+    0.75,
+    0.9,
+    1.0,
+    1.1,
+    1.25,
+    1.5,
+    2.0,
+    4.0,
+]
+
+
+def _token_budget_evaluations(token_budget: Any, usage: Any) -> list[dict[str, Any]]:
+    # §5.5.15 / §11.2 (proposal 0083): the per-bound token-budget evaluation
+    # shared by the span attrs and the metric recorder. Returns one entry per
+    # 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).
+    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:
+        evaluations.append({"kind": "input", "actual": usage.prompt_tokens, "max": input_max})
+    total_max = getattr(token_budget, "total_max_tokens", None)
+    if total_max:
+        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:
+            total_actual = usage.prompt_tokens + usage.completion_tokens
+        else:
+            total_actual = None
+        if total_actual is not None:
+            evaluations.append({"kind": "total", "actual": total_actual, "max": total_max})
+    return evaluations
 
 
 def _read_spec_version() -> str:
@@ -634,6 +680,10 @@ class OTelObserver:
     _meter: Meter | None = field(init=False, repr=False, default=None)
     _token_histogram: Histogram | None = field(init=False, repr=False, default=None)
     _duration_histogram: Histogram | None = field(init=False, repr=False, default=None)
+    # §11.2 (proposal 0083) per-prompt token-budget instruments — created
+    # alongside the token/duration histograms when enable_metrics is True.
+    _token_budget_exceeded_counter: Counter | None = field(init=False, repr=False, default=None)
+    _token_budget_utilization_histogram: Histogram | None = field(init=False, repr=False, default=None)
     # Per-invocation_id span state — concurrent invocations on a
     # shared observer each get their own ``_InvState`` so internal
     # maps never collide.
@@ -695,6 +745,19 @@ def __post_init__(self) -> None:
                 unit="s",
                 explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
             )
+            # §11.2 (proposal 0083) per-prompt token-budget instruments: an
+            # exceeded counter ({call}) that increments per breached bound and
+            # a utilization histogram (unitless ratio) that records on every
+            # budgeted call, exceeded or not.
+            self._token_budget_exceeded_counter = self._meter.create_counter(
+                "openarmature.gen_ai.client.token_budget.exceeded",
+                unit="{call}",
+            )
+            self._token_budget_utilization_histogram = self._meter.create_histogram(
+                "openarmature.gen_ai.client.token_budget.utilization",
+                unit="1",
+                explicit_bucket_boundaries_advisory=_TOKEN_BUDGET_UTILIZATION_BUCKETS,
+            )
 
     # ------------------------------------------------------------------
     # Enricher invocation (friction-roundup #7p2)
@@ -1444,6 +1507,20 @@ def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None:
                     usage.completion_tokens,
                     {**base_dims, "openarmature.gen_ai.token.type": "output"},
                 )
+        # §11.2 (proposal 0083) per-prompt token-budget instruments, from the
+        # shared per-bound evaluation (same per-attempt path as token.usage, so a
+        # structured_output_invalid failure -- which carries usage -- records
+        # too). Utilization on every declared+evaluable bound; the exceeded
+        # counter only when that bound is breached.
+        if (
+            self._token_budget_utilization_histogram is not None
+            and self._token_budget_exceeded_counter is not 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)
+                if ev["actual"] > ev["max"]:
+                    self._token_budget_exceeded_counter.add(1, dims)
 
     def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None:
         """Open + close one ``openarmature.llm.complete`` span from a
@@ -1508,6 +1585,31 @@ def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None:
             attrs["openarmature.prompt.label"] = active_prompt.label
             attrs["openarmature.prompt.template_hash"] = active_prompt.template_hash
             attrs["openarmature.prompt.rendered_hash"] = active_prompt.rendered_hash
+        # §5.5.15 (proposal 0083) token-budget span surface. Each declared
+        # bound renders its own attribute (omitted when that bound is null);
+        # the exceeded signal is emitted ONLY when a budget is declared
+        # (>= 1 bound) AND the attempt carries usage, value = OR of per-bound
+        # exceedances. No budget or no usage -> the attribute is absent (not
+        # false), matching the §5.5 omit-when-undeclared convention.
+        token_budget = event.token_budget
+        if token_budget is not None:
+            # Declared bounds surface regardless of usage (each only when the
+            # prompt declared it); a 0 bound is still a declared value.
+            input_max = getattr(token_budget, "input_max_tokens", None)
+            total_max = getattr(token_budget, "total_max_tokens", None)
+            if input_max is not None:
+                attrs["openarmature.prompt.token_budget.input_max_tokens"] = input_max
+            if total_max is not None:
+                attrs["openarmature.prompt.token_budget.total_max_tokens"] = total_max
+            # The exceeded signal is emitted ONLY when at least one bound was
+            # actually evaluated (declared, non-degenerate, with a reported
+            # count); value = OR of per-bound breaches. No evaluable bound ->
+            # the attribute is absent (not false), per §5.5.15.
+            evaluations = _token_budget_evaluations(token_budget, event.usage)
+            if evaluations:
+                attrs["openarmature.llm.token_budget.exceeded"] = any(
+                    ev["actual"] > ev["max"] for ev in evaluations
+                )
         active_group = event.active_prompt_group
         if active_group is not None:
             attrs["openarmature.prompt.group_name"] = active_group.group_name

src/openarmature/prompts/__init__.py

@@ -37,6 +37,7 @@
     SamplingConfig,
     TextBlockTemplate,
     TextPrompt,
+    TokenBudget,
 )
 
 __all__ = [
@@ -69,6 +70,7 @@
     "SamplingConfig",
     "TextBlockTemplate",
     "TextPrompt",
+    "TokenBudget",
     "compute_rendered_hash",
     "compute_template_hash",
     "current_prompt_group",