diff --git a/conformance.toml b/conformance.toml index f38dd6c..153d5cd 100644 --- a/conformance.toml +++ b/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 + diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 8066c57..a5ead2d 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/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: diff --git a/tests/conformance/test_fixture_parsing.py b/tests/conformance/test_fixture_parsing.py index 67bc2af..754126a 100644 --- a/tests/conformance/test_fixture_parsing.py +++ b/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.) diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index fe0224a..f5df5b7 100644 --- a/tests/conformance/test_observability.py +++ b/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: diff --git a/tests/unit/test_observability_langfuse.py b/tests/unit/test_observability_langfuse.py index bc7d054..6a0542c 100644 --- a/tests/unit/test_observability_langfuse.py +++ b/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 diff --git a/tests/unit/test_prompts.py b/tests/unit/test_prompts.py index 1311a3f..5276496 100644 --- a/tests/unit/test_prompts.py +++ b/tests/unit/test_prompts.py @@ -264,6 +264,27 @@ async def fetch( assert len(result.messages) == 1 +def test_render_propagates_token_budget() -> None: + # Proposal 0083: PromptManager.render propagates Prompt.token_budget onto + # PromptResult verbatim, as a distinct copy -- the production path the + # conformance harness's _render_prompt_result shortcut bypasses. + from openarmature.prompts import TokenBudget + + budget = TokenBudget(input_max_tokens=10, total_max_tokens=40) + prompt = _make_prompt().model_copy(update={"token_budget": budget}) + + class _Backend: + async def fetch( + self, name: str, label: str = "production", *, cache_ttl_seconds: int | None = None + ) -> Prompt: + return prompt + + manager = PromptManager(_Backend()) + result = manager.render(prompt, {"user": "Alice"}) + assert result.token_budget == budget + assert result.token_budget is not budget + + async def test_fetch_rejects_negative_cache_ttl_seconds() -> None: prompt = _make_prompt() @@ -720,6 +741,22 @@ async def test_filesystem_backend_token_budget_unified(tmp_path: Path) -> None: assert prompt.sampling.temperature == 0.0 +async def test_filesystem_backend_unified_malformed_token_budget_is_fallback_eligible(tmp_path: Path) -> None: + # The malformed-budget -> fallback-eligible PromptStoreUnavailable conversion + # applies on the unified path too, not just per-prompt-sidecar (invariant: a + # malformed advisory budget must never break the LLM call). + (tmp_path / "production").mkdir() + (tmp_path / "production" / "classify.j2").write_text("Classify: {{ topic }}", encoding="utf-8") + (tmp_path / "prompt_configs.json").write_text( + '{"classify": {"token_budget": "1000"}}', + encoding="utf-8", + ) + + backend = FilesystemPromptBackend(tmp_path, token_budget_source="unified") + with pytest.raises(PromptStoreUnavailable): + await backend.fetch("classify", "production") + + async def test_filesystem_backend_malformed_token_budget_is_fallback_eligible(tmp_path: Path) -> None: # A malformed advisory token_budget (a string, not an object) surfaces as the # fallback-eligible PromptStoreUnavailable -- NOT a bare exception that would @@ -784,6 +821,13 @@ def test_langfuse_token_budget_from_config() -> None: assert _token_budget_from_config({"token_budget": {"unknown": 5}}) is None assert _token_budget_from_config({"token_budget": {"input_max_tokens": None}}) is None assert _token_budget_from_config({"token_budget": {"input_max_tokens": -5}}) is None + # Extra keys alongside a valid bound are dropped, the valid bound kept (the + # langfuse tolerate model -- a remote config may carry unrecognized keys). + # The filesystem sidecar rejects the whole budget on an extra key (extra= + # forbid); divergence flagged to spec. + assert _token_budget_from_config({"token_budget": {"input_max_tokens": 10, "unknown": 5}}) == TokenBudget( + input_max_tokens=10 + ) # LabelResolver fallback chain — covered by fixture 015 end-to-end,