Complete 0083 with the token-budget failure path (#238)

* Add 0083 failure path and mark implemented

Complete proposal 0083 (per-prompt token-budget observability) with
the structured-output-failure path. No source change was needed: the
budget already evaluates on any usage-bearing per-attempt event, so a
structured_output_invalid failure (which carries usage, proposal 0082)
drives the span, metrics, and section 7 log via the generic path, and
a no-usage provider_unavailable failure records nothing token-budget.

- conformance harness: route fixture 131 to the token-budget runner;
  add expected_error handling (raises + raised_from), make the exact
  metric assertion conditional on a declared metrics block (a failure
  still records the baseline operation.duration), and implement the
  two without-usage invariants.
- conformance.toml: mark 0083 implemented (all fixtures 126-131 run).

* Fold whole-proposal review findings into 0083

Address the six findings from the whole-proposal adversarial review
(0 blockers): cross-surface robustness plus the end-to-end coverage
the per-PR reviews could not see.

- langfuse: read token_budget via getattr in the metadata block,
  matching the OTel span + shared helper, so a non-TokenBudget value
  degrades instead of crashing the Generation.
- harness: fail loudly on an unmapped token-budget invariant name
  (no more vacuous pass) and handle exception_propagates.
- tests: a structured_output_invalid over-budget failure through
  Langfuse (ERROR wins, metadata.token_budget present, no WARNING);
  the PromptManager.render -> PromptResult.token_budget propagation;
  the langfuse extra-key tolerate case; and the filesystem unified
  malformed-budget fallback-eligible path.

Author: chris-colinsky

conformance.toml

@@ -825,13 +825,12 @@ since = "0.17.0"
 note = "The graph-engine §6 event surface + llm-provider §7 error surface (PR A). LlmFailedEvent gains five response-side fields (output_content / finish_reason / usage / response_id / response_model), populated ONLY for structured_output_invalid (None for every other category); output_content is payload-bearing, the other four are not gated. StructuredOutputInvalid gains finish_reason / usage (the §7-mandated pair) plus response_id / response_model (internal, for the event builder), attached at the OpenAI provider's parse/validate call site where the intact wire response is in hand; the failed-event builder reads them off the enriched error. Un-defers observability fixtures 120 (truncation, finish_reason=length) / 121 (schema-mismatch, finish_reason=stop) / 122 (null-on-non-body companion), driven via the typed-event collector with a new calls_llm.response_schema harness lever; and llm-provider fixtures 022 / 023, whose carries block now asserts finish_reason + usage (the carries reader gained subset matching for a mapping-valued field like usage). PR B adds the OTel rendering + the metric: the per-attempt LlmRetryAttemptEvent now carries the same response-side surface on a structured_output_invalid failed attempt, so the OTel error span renders it (a structured_output_invalid failure falls through the failed-attempt early-return to render finish_reason / usage / gated output.content / response id+model while keeping ERROR status + category) and the §11.2 token-usage histogram records the failure's usage (the metric already recorded whenever usage was present; a structured failure now carries it). Un-defers observability fixtures 124 (OTel error span, payload on/off) and 125 (token metric on the failure). PR C completes it with the Langfuse failed-Generation rendering: _handle_typed_llm_failed renders output (payload-gated) + usage on the ERROR Generation and _typed_event_metadata renders finish_reason / response id+model for a structured_output_invalid failure (the isinstance(LlmCompletionEvent) guard widened to include it), alongside the existing level=ERROR + statusMessage=category. Un-defers observability fixture 123, driven in the dedicated test_observability_langfuse harness (calls_llm.response_schema threaded + expected_error handled). 0082 now fully implemented."
 
 # 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 (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.
+# (prompt-management §3 / graph-engine §6 / observability §5.5.15 / §7 / §8.4.3
+# / §11.2).  Implemented across three PRs; all six fixtures 126-131 run.
 [proposals."0083"]
-status = "not-yet"
+status = "implemented"
+since = "0.17.0"
+note = "Advisory, observability-only per-prompt token budget. PR A (completion path): a TokenBudget sub-record ({input_max_tokens?, total_max_tokens?}, non-negative) on Prompt / PromptResult, sourced from the same config sidecar as sampling (filesystem + langfuse backends); carried on LlmCompletionEvent / LlmFailedEvent / LlmRetryAttemptEvent (the OTel span + §11 metrics render from the per-attempt event). The OTel LLM span gains openarmature.prompt.token_budget.input_max_tokens / .total_max_tokens (each only when declared) + the reactive openarmature.llm.token_budget.exceeded boolean (emitted only when a bound is declared AND usage is present -- absent, not false, otherwise); two opt-in §11.2 instruments: openarmature.gen_ai.client.token_budget.exceeded counter (per breached bound) + .utilization histogram (actual / max per declared bound on every budgeted call), both dimensioned by openarmature.gen_ai.token_budget.kind = input | total. Evaluation gates each bound on its reported count being present (mirrors token.usage, does NOT coerce a missing count to 0); a declared bound of 0 is exceeded by any positive usage (strict >) but its utilization sample is skipped (ratio undefined). Un-defers fixtures 126-129. PR B (WARNING surfaces): the shared _token_budget_evaluations helper lifted to observability/llm_event.py; the Langfuse success Generation ends at level WARNING + statusMessage naming the breached bound(s) + metadata.token_budget on a breach (a hard ERROR still wins on the failure path); a §7 WARNING log per over-budget attempt on the openarmature.observability logger, dispatched independently of the disable_llm_spans / enable_metrics gates (correlates via correlation_id -- OA never activates spans, so no trace_id/span_id). Un-defers fixture 130. PR C (failure path): the budget evaluates on a usage-bearing structured_output_invalid failure exactly as on a completion (span + metrics + §7 log fall out of the generic per-attempt handling), and is populated-but-not-evaluated on a no-usage failure (provider_unavailable); the token-budget runner gained expected_error handling + the without-usage invariants. Un-defers fixture 131 (both cases). 0083 now fully implemented."
 
 # Spec v0.81.0 (proposal 0084).  Nested-fan-out span lineage chain
 # (graph-engine §6 fan_out_index_chain / branch_name_chain +

src/openarmature/observability/langfuse/observer.py

@@ -2236,10 +2236,15 @@ def _typed_event_metadata(
         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
+            # token_budget is typed Any; read defensively (getattr) so a
+            # non-TokenBudget value degrades to no budget metadata rather than
+            # crashing the Generation, matching the OTel span + shared helper.
+            input_max = getattr(token_budget, "input_max_tokens", None)
+            total_max = getattr(token_budget, "total_max_tokens", None)
+            if input_max is not None:
+                budget["input_max_tokens"] = input_max
+            if total_max is not None:
+                budget["total_max_tokens"] = total_max
             if budget:
                 metadata["token_budget"] = budget
         if event.caller_invocation_metadata is not None:

tests/conformance/test_fixture_parsing.py

@@ -611,12 +611,9 @@ def _id(case: tuple[str, Path]) -> str:
         "Proposal 0082 structured-output failure event; harness typed_event_collector schema pending"
     ),
     # 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) 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"
-    ),
+    # completion-path fixtures (126-129) + the structured-output-failure path
+    # (131) parse + run in test_observability; the Langfuse WARNING (130) parses
+    # + runs via test_observability_langfuse. All of 0083's fixtures now run.
     # Proposal 0084 (nested-fan-out span lineage, v0.81.0) -- the
     # lineage-chain directive shapes. (132 parses cleanly; accounted in
     # test_observability.)

tests/conformance/test_observability.py

@@ -228,11 +228,15 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
         # metrics (utilization histogram + exceeded counter) for a budgeted
         # LLM completion. 126 input-exceeded, 127 total-exceeded, 128
         # under-budget (utilization records, no exceeded), 129 no-budget
-        # baseline. The Langfuse WARNING (130) + failure path (131) are later.
+        # baseline. 131 is the failure path: the budget evaluates on a
+        # structured_output_invalid failure (carries usage, proposal 0082) and
+        # NOT on a no-usage provider_unavailable failure. The Langfuse WARNING
+        # (130) runs in the dedicated test_observability_langfuse harness.
         "126-token-budget-input-exceeded",
         "127-token-budget-total-exceeded",
         "128-token-budget-under-budget-no-warning",
         "129-token-budget-absent-unchanged",
+        "131-token-budget-on-structured-output-failure",
         # v0.69.0 — proposal 0063 (tool-execution observability). A
         # calls_tool node enters the with_tool_call scope; the typed
         # ToolCallEvent / ToolCallFailedEvent drive the OTel tool span +
@@ -372,16 +376,13 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
         "Langfuse failed-Generation rendering; driven in test_observability_langfuse"
     ),
     # Proposal 0083 (per-prompt token-budget observability, spec v0.78.0).
-    # The completion-path span attributes + §11.2 metrics (126-129) run here
-    # (see _SUPPORTED_FIXTURES). The Langfuse WARNING-level rendering (130) and
-    # the structured-output-failure path (131) land in later v0.16.0 PRs.
-    **{
-        fixture_id: "per-prompt token-budget observability (proposal 0083) not-yet implemented"
-        for fixture_id in (
-            "130-langfuse-token-budget-warning-level",
-            "131-token-budget-on-structured-output-failure",
-        )
-    },
+    # The completion path (126-129) and the failure path (131) run here (see
+    # _SUPPORTED_FIXTURES); the Langfuse WARNING-level rendering (130) is driven
+    # in the dedicated test_observability_langfuse harness (its langfuse_trace
+    # shape lives there, like 123).
+    "130-langfuse-token-budget-warning-level": (
+        "Langfuse WARNING-level rendering; driven in test_observability_langfuse"
+    ),
     # Proposal 0084 (nested-fan-out span lineage, spec v0.81.0).
     # Fixture 132's nested-lineage dispatch KEYING shipped in #194 (the
     # OTel + Langfuse observers key dispatches by the full enclosing
@@ -715,6 +716,7 @@ async def test_observability_fixture(fixture_path: Path) -> None:
         "127-token-budget-total-exceeded",
         "128-token-budget-under-budget-no-warning",
         "129-token-budget-absent-unchanged",
+        "131-token-budget-on-structured-output-failure",
     }:
         await _run_token_budget_fixture(spec)
     elif fixture_id in {
@@ -3950,8 +3952,25 @@ async def _run_token_budget_case(case: Mapping[str, Any]) -> None:
     handles += [graph.attach_observer(c) for c in collectors.values()]
 
     state = _make_state_instance(case, state_cls)
+    # 131's failure path raises out of the node; the budget still evaluates on
+    # the (usage-bearing) structured_output_invalid attempt. The §7 error
+    # category rides the span status_description (checked below), NOT the outer
+    # NodeException (whose category is always "node_exception"), so only
+    # raised_from is asserted against the exception here.
+    expected_error = cast("dict[str, Any] | None", case.get("expected_error"))
     try:
-        await graph.invoke(state)
+        if expected_error is not None:
+            from openarmature.graph import NodeException  # noqa: PLC0415
+
+            with pytest.raises(NodeException) as exc_info:
+                await graph.invoke(state)
+            raised_from = expected_error.get("raised_from")
+            if raised_from is not None:
+                assert exc_info.value.node_name == raised_from, (
+                    f"expected error raised_from: expected {raised_from!r}, got {exc_info.value.node_name!r}"
+                )
+        else:
+            await graph.invoke(state)
         await graph.drain()
     finally:
         for handle in handles:
@@ -3985,8 +4004,13 @@ async def _run_token_budget_case(case: Mapping[str, Any]) -> None:
 
     # ---- metrics + absence invariants ----
     points = _collect_metric_points(reader)
-    expected_metrics = cast("list[dict[str, Any]]", expected.get("metrics") or [])
-    _assert_metric_points(points, expected_metrics)
+    # Only exact-assert when the fixture declares a metrics block. A no-usage
+    # failure (131 case 2) declares none: it still records the §11 baseline
+    # operation.duration (with error.type), and asserts the token-budget-metric
+    # absence via its invariants instead of an exact empty-set match.
+    expected_metrics = cast("list[dict[str, Any]] | None", expected.get("metrics"))
+    if expected_metrics is not None:
+        _assert_metric_points(points, expected_metrics)
     _assert_token_budget_invariants(case, points, collectors)
 
 
@@ -3995,11 +4019,24 @@ def _assert_token_budget_invariants(
     points: Sequence[tuple[str, float, int, dict[str, Any]]],
     collectors: Mapping[str, Any],
 ) -> None:
-    """Evaluate the token-budget named invariants (128/129) as absence
-    predicates over the captured measurement set + the completion event.
+    """Evaluate the token-budget named invariants (128/129/131) as absence
+    predicates over the captured measurement set + the typed events.
     The ``metrics:`` list shape asserts presence only, so these cover the
     "no observation recorded" claims the list cannot express."""
     invariants = cast("dict[str, Any]", case["expected"].get("invariants") or {})
+    # Fail loudly on a declared invariant this runner does not map, so a future
+    # fixture's new invariant name cannot pass vacuously (harness-fidelity).
+    _known_invariants = {
+        "no_token_budget_exceeded_observation_when_under_budget",
+        "utilization_records_under_budget",
+        "no_token_budget_instrument_observations_when_no_budget",
+        "token_budget_null_on_completion_event",
+        "no_token_budget_instrument_observations_without_usage",
+        "token_budget_populated_but_not_evaluated_without_usage",
+        "exception_propagates_alongside_typed_event",
+    }
+    unknown = set(invariants) - _known_invariants
+    assert not unknown, f"unhandled token-budget invariant(s): {sorted(unknown)}"
     exceeded_points = [p for p in points if p[0] == "openarmature.gen_ai.client.token_budget.exceeded"]
     utilization_points = [p for p in points if p[0] == "openarmature.gen_ai.client.token_budget.utilization"]
 
@@ -4027,6 +4064,35 @@ def _assert_token_budget_invariants(
         assert all(getattr(e, "token_budget", "missing") is None for e in completion_events), (
             "no-budget call MUST carry token_budget=None on the completion event"
         )
+    if invariants.get("no_token_budget_instrument_observations_without_usage"):
+        assert not exceeded_points and not utilization_points, (
+            "no-usage failure MUST record no token-budget instrument observations; got "
+            f"exceeded={exceeded_points} utilization={utilization_points}"
+        )
+    if invariants.get("token_budget_populated_but_not_evaluated_without_usage"):
+        failed_events = [
+            e
+            for collector in collectors.values()
+            for e in collector.events
+            if type(e).__name__ == "LlmFailedEvent"
+        ]
+        assert failed_events, "expected at least one LlmFailedEvent to assert token_budget populated"
+        assert all(getattr(e, "token_budget", None) is not None for e in failed_events), (
+            "a no-usage failure from a budgeted prompt MUST still carry token_budget on the event"
+        )
+        assert not exceeded_points and not utilization_points, (
+            "a no-usage failure MUST NOT evaluate the budget (no token-budget metric observation)"
+        )
+    if invariants.get("exception_propagates_alongside_typed_event"):
+        # The runner's pytest.raises already asserts the exception propagated;
+        # assert the typed failure event fired alongside it (both, not either).
+        failed_events = [
+            e
+            for collector in collectors.values()
+            for e in collector.events
+            if type(e).__name__ == "LlmFailedEvent"
+        ]
+        assert failed_events, "expected an LlmFailedEvent alongside the propagated exception"
 
 
 def _assert_error_span_extras(spans: Sequence[Any], expected_tree: list[dict[str, Any]]) -> None:

tests/unit/test_observability_langfuse.py

@@ -1513,6 +1513,43 @@ async def test_typed_llm_failed_generation_carries_token_budget_metadata() -> No
     assert gen.metadata.get("token_budget") == {"input_max_tokens": 10}
 
 
+async def test_structured_over_budget_failure_generation_is_error_not_warning() -> None:
+    # §8.4.3 (proposal 0083): a structured_output_invalid failure that ALSO
+    # exceeds budget renders the ERROR Generation (a hard ERROR wins over the
+    # advisory WARNING) with metadata.token_budget still present -- the failed
+    # handler never applies the success-path WARNING.
+    from openarmature.llm.response import Usage
+    from openarmature.observability.correlation import (
+        _reset_invocation_id,
+        _set_invocation_id,
+    )
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_failed_event
+
+    client = InMemoryLangfuseClient()
+    observer = LangfuseObserver(client=client)
+    token = _set_invocation_id("inv-tb-soi")
+    try:
+        await observer(
+            make_failed_event(
+                invocation_id="inv-tb-soi",
+                error_category="structured_output_invalid",
+                error_type="StructuredOutputInvalid",
+                error_message="schema mismatch",
+                usage=Usage(prompt_tokens=20, completion_tokens=1, total_tokens=21),
+                token_budget=TokenBudget(input_max_tokens=10),
+            )
+        )
+    finally:
+        _reset_invocation_id(token)
+
+    gen = next(o for o in client.traces["inv-tb-soi"].observations if o.type == "generation")
+    assert gen.level == "ERROR"
+    assert gen.status_message == "structured_output_invalid"
+    assert "token budget exceeded" not in (gen.status_message or "")
+    assert gen.metadata.get("token_budget") == {"input_max_tokens": 10}
+
+
 async def test_structured_output_failure_generation_renders_response_surface() -> None:
     # Proposal 0082: a structured_output_invalid failure renders the response-side
     # surface (output payload-gated, usage, metadata.finish_reason) on the ERROR