Add 0083 token-budget WARNING surfaces

conformance.toml

@@ -826,9 +826,10 @@ 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 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.
+# Not-yet overall: the completion-path OTel rendering + fixtures 126-129 (PR A)
+# and the Langfuse WARNING + §7 log + fixture 130 (PR B) run; only the
+# structured-output-failure path (131) stays deferred until PR C, which is when
+# this flips to implemented.
 [proposals."0083"]
 status = "not-yet"
 

src/openarmature/observability/langfuse/observer.py

@@ -47,6 +47,7 @@
 )
 from openarmature.graph.observer import ObserverEvent
 from openarmature.observability.lineage import is_strict_prefix
+from openarmature.observability.llm_event import _token_budget_evaluations
 
 from .client import (
     LangfuseClient,
@@ -1753,6 +1754,20 @@ def _handle_typed_llm_completion(self, event: LlmCompletionEvent) -> None:
         end_kwargs: dict[str, Any] = {}
         if usage is not None:
             end_kwargs["usage"] = usage
+        # §8.4.3 (proposal 0083): a token-budget exceedance sets the Generation's
+        # advisory WARNING level + a statusMessage naming each breached bound. The
+        # call succeeded (this is the success handler), so WARNING never displaces
+        # a hard ERROR — the failure path is untouched. No breach -> end as before.
+        breached = [
+            ev
+            for ev in _token_budget_evaluations(event.token_budget, event.usage)
+            if ev["actual"] > ev["max"]
+        ]
+        if breached:
+            end_kwargs["level"] = "WARNING"
+            end_kwargs["status_message"] = "token budget exceeded: " + ", ".join(
+                f"{ev['kind']} {ev['actual']} > {ev['max']}" for ev in breached
+            )
         handle.end(end_time=end_time, **end_kwargs)
 
     def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None:
@@ -2214,6 +2229,19 @@ def _typed_event_metadata(
         active_group = event.active_prompt_group
         if active_group is not None:
             metadata["prompt_group_name"] = active_group.group_name
+        # §8.4.3 (proposal 0083): the declared token-budget bounds map to
+        # metadata.token_budget.*, each key present only when that bound is
+        # non-None. Omitted entirely when no budget was declared. Shared with the
+        # failure handler (harmless there; a failed Generation carries it too).
+        token_budget = event.token_budget
+        if token_budget is not None:
+            budget: dict[str, Any] = {}
+            if token_budget.input_max_tokens is not None:
+                budget["input_max_tokens"] = token_budget.input_max_tokens
+            if token_budget.total_max_tokens is not None:
+                budget["total_max_tokens"] = token_budget.total_max_tokens
+            if budget:
+                metadata["token_budget"] = budget
         if event.caller_invocation_metadata is not None:
             _apply_caller_metadata(metadata, event.caller_invocation_metadata)
         # Response-side metadata. A completion always carries it; a

src/openarmature/observability/llm_event.py

@@ -162,4 +162,40 @@ def serialize_tool_calls(tool_calls: Sequence[ToolCall]) -> list[dict[str, Any]]
     return [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls]
 
 
-__all__ = ["LLM_NAMESPACE", "LlmEventPayload", "serialize_tool_calls"]
+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 OTel span attrs, the metric recorder, and the Langfuse
+    # WARNING / §7 log surfaces. 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 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 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 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:
+            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
+
+
+__all__ = [
+    "LLM_NAMESPACE",
+    "LlmEventPayload",
+    "_token_budget_evaluations",
+    "serialize_tool_calls",
+]

src/openarmature/observability/otel/observer.py

@@ -75,6 +75,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import time
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field
@@ -117,7 +118,17 @@
 )
 from openarmature.graph.observer import ObserverEvent
 from openarmature.observability.lineage import is_strict_prefix
-from openarmature.observability.llm_event import serialize_tool_calls
+from openarmature.observability.llm_event import _token_budget_evaluations, serialize_tool_calls
+
+# §7 (proposal 0083): the vendor-neutral token-budget WARNING log surface.
+# The logger name is our choice (the spec doesn't pin one). The OTLP log
+# bridge attaches correlation_id (via its LogRecord factory) but NOT OTel
+# trace_id / span_id -- the observer never activates its spans and this record
+# is emitted outside any span context -- so §7 correlation is via
+# correlation_id. The record names the prompt + the breached bound(s), one per
+# over-budget attempt, independent of the §11 metrics opt-in.
+logger = logging.getLogger("openarmature.observability")
+
 
 # Span-stack key shape:
 # ``(namespace, attempt_index, fan_out_index, branch_name)`` — these
@@ -214,37 +225,6 @@
 ]
 
 
-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 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 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 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:
-            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:
     """Read the spec version pinned at package level. Lazy import
     avoids a circular at module-load time (the package's ``__init__``
@@ -836,6 +816,10 @@ async def __call__(
         # LlmRetryAttemptEvent — one span per attempt under call-level
         # retry (attempt_index 0..N-1), one for a no-retry call.
         if isinstance(event, LlmRetryAttemptEvent):
+            # §7 token-budget WARNING log is its own surface: it fires
+            # independently of both the span (disable_llm_spans) and metric
+            # (enable_metrics) gates.
+            self._maybe_log_token_budget_exceedance(event)
             # §11 metrics record per attempt, independent of span emission
             # (§11.1) — so this runs regardless of disable_llm_spans.
             if self.enable_metrics:
@@ -1528,6 +1512,31 @@ def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None:
                 if ev["actual"] > ev["max"]:
                     self._token_budget_exceeded_counter.add(1, dims)
 
+    def _maybe_log_token_budget_exceedance(self, event: LlmRetryAttemptEvent) -> None:
+        # §7 (proposal 0083): one vendor-neutral WARNING log record per
+        # over-budget attempt (not per bound). Its own surface -- independent of
+        # disable_llm_spans and enable_metrics -- so it is dispatched here rather
+        # than from the span/metric handlers. Names the prompt + each breached
+        # bound (kind, actual, budget). The OTLP log bridge attaches
+        # correlation_id via its LogRecord factory; it does NOT populate OTel
+        # trace_id / span_id, because the observer never makes its spans current
+        # (start_span with explicit parents) and this record is emitted from the
+        # detached delivery worker, so no span is active at log time. The record
+        # correlates via correlation_id.
+        breached = [
+            ev
+            for ev in _token_budget_evaluations(event.token_budget, event.usage)
+            if ev["actual"] > ev["max"]
+        ]
+        if not breached:
+            return
+        active_prompt = event.active_prompt
+        prompt_ident: Any = active_prompt.name if active_prompt is not None else None
+        if active_prompt is not None and active_prompt.version is not None:
+            prompt_ident = f"{active_prompt.name} {active_prompt.version}"
+        breaches = ", ".join(f"{ev['kind']} {ev['actual']} > {ev['max']}" for ev in breached)
+        logger.warning("token budget exceeded for prompt %r: %s", prompt_ident, breaches)
+
     def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None:
         """Open + close one ``openarmature.llm.complete`` span from a
         per-attempt LlmRetryAttemptEvent.

tests/conformance/test_fixture_parsing.py

@@ -612,9 +612,8 @@ def _id(case: tuple[str, Path]) -> str:
     ),
     # Proposal 0083 (per-prompt token-budget observability, v0.78.0) -- the
     # completion-path fixtures (126-129) parse + run in test_observability; the
-    # Langfuse WARNING (130) + structured-output-failure (131) paths stay
-    # deferred until their PRs.
-    "observability/130-langfuse-token-budget-warning-level": "Proposal 0083 token-budget; not implemented",
+    # Langfuse WARNING (130) parses + runs via test_observability_langfuse; the
+    # structured-output-failure (131) path stays deferred until its PR.
     "observability/131-token-budget-on-structured-output-failure": (
         "Proposal 0083 token-budget; not implemented"
     ),

tests/conformance/test_observability_langfuse.py

@@ -41,6 +41,7 @@
     PromptManager,
     SamplingConfig,
     TextPrompt,
+    TokenBudget,
 )
 from openarmature.prompts.context import with_active_prompt
 
@@ -122,6 +123,12 @@
         # on the ERROR failed Generation. Drives the real failure path via
         # calls_llm.response_schema + expected_error.
         "123-langfuse-failed-generation-renders-output-usage-finish-reason",
+        # 130 (proposal 0083): a SUCCESSFUL completion whose prompt_tokens exceed
+        # the declared input_max_tokens renders the advisory WARNING level +
+        # statusMessage naming the breached bound, and maps the declared budget
+        # to metadata.token_budget.*. Drives the input-exceeded path via
+        # renders_prompt + the prompt's token_budget block.
+        "130-langfuse-token-budget-warning-level",
     }
 )
 
@@ -425,6 +432,18 @@ def __init__(self, prompts: dict[str, dict[str, Any]], *, with_langfuse_referenc
             observability_entities: dict[str, Any] | None = None
             if with_langfuse_reference and "langfuse_prompt_reference" in spec:
                 observability_entities = {"langfuse_prompt": spec["langfuse_prompt_reference"]}
+            # Proposal 0083: the prompt's token_budget block propagates onto the
+            # TextPrompt so PromptManager.render carries it to the PromptResult
+            # (and thence the typed event). An empty / all-null budget collapses
+            # to None, mirroring the real backends.
+            token_budget_spec = cast("dict[str, Any] | None", spec.get("token_budget"))
+            token_budget = TokenBudget(**token_budget_spec) if token_budget_spec is not None else None
+            if (
+                token_budget is not None
+                and token_budget.input_max_tokens is None
+                and token_budget.total_max_tokens is None
+            ):
+                token_budget = None
             self._prompts[prompt_name] = TextPrompt(
                 name=spec["name"],
                 version=spec["version"],
@@ -433,6 +452,7 @@ def __init__(self, prompts: dict[str, dict[str, Any]], *, with_langfuse_referenc
                 template_hash=spec["template_hash"],
                 fetched_at=now,
                 observability_entities=observability_entities,
+                token_budget=token_budget,
             )
 
     async def fetch(