From 0a7c855bc40c04345608dc5c622332bea71dc03e Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sun, 9 Aug 2026 13:38:31 -0700 Subject: [PATCH 1/4] Add credentials-in Langfuse client construction The Langfuse observer could only take a caller-built client, and a Langfuse v4 client constructed without an explicit tracer_provider binds the global one, so OA's observations were exported to every processor on the application's provider. LangfuseObserver and LangfuseSDKAdapter gain from_credentials, which builds an OA-owned client on a dedicated TracerProvider reused per credential. The SDK caches one client per public_key, so a dedicated provider only takes effect when OA constructs first. from_credentials reads the binding back and fails closed: when a payload channel is live and the client landed on a provider OA did not isolate, construction raises LangfuseProviderIsolationUnavailable before anything is emitted; when the binding cannot be established it suppresses every channel and warns; accept_shared_provider opts out with a warning instead. The guarded channels are the provider payload, the Trace state payload and its hooks, and a failed provider observation's error message, which is omitted per emission with the error category retained. The failure-isolation marker span no longer carries the caught exception's message. No mapping table covers that span, so writing harvested exception content onto it was over-emission that no privacy setting gated; it now matches the node span and carries only the category, with the full exception still on the OTel side. Credentials are taken as SecretStr so they are masked in OA's reprs and logs. --- CHANGELOG.md | 2 + docs/agent/non-obvious-shapes.md | 8 + src/openarmature/AGENTS.md | 8 + .../observability/langfuse/__init__.py | 2 + .../observability/langfuse/adapter.py | 130 ++++++++- .../observability/langfuse/client.py | 10 + .../observability/langfuse/errors.py | 24 ++ .../observability/langfuse/observer.py | 149 +++++++++- .../unit/test_langfuse_provider_isolation.py | 265 ++++++++++++++++++ 9 files changed, 577 insertions(+), 21 deletions(-) create mode 100644 src/openarmature/observability/langfuse/errors.py create mode 100644 tests/unit/test_langfuse_provider_isolation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d891085f..c29587c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The - **Langfuse parallel-branches mapping parity** (proposal 0088, observability §8.4.8 / §8.3 / §8.4.2 / §3.4, spec v0.83.0). Brings the Langfuse observer's parallel-branches rendering to parity with the OTel side. The observer already synthesized the three-level Observation tree (the parallel-branches node Span, a per-branch dispatch Span named by the `branch_name`, and the branch's inner observations) and already emitted the dispatch-span `parallel_branches_parent_node_name` and `branch_name`; the two node-span attributes `parallel_branches_branch_count` and `parallel_branches_error_policy` are now flattened onto the node Span's `observation.metadata` (mirroring the `fan_out_*` attributes), the one §8.4.2 row the observer had never mapped. The three `parallel_branches_*` keys join the reserved caller-metadata set (26 to 29), so a caller passing one as invocation metadata is rejected at the `invoke()` boundary rather than shadowing the OA-emitted field. The OTel side was already complete. Conformance fixture 136 (the dedicated three-level-tree pin) is un-deferred; fixture 030's incidental coverage stands. - **Adaptive call-level retry: per-attempt request override** (proposal 0095, llm-provider §7.1, spec v0.91.0). The LLM-completion call-level retry loop gains an opt-in per-attempt request override. A new `LlmRetryConfig` (the llm-provider-scoped superset of the generic `RetryConfig`, exported from `openarmature.llm`) carries a `per_attempt_override`: a schedule of `RuntimeConfig` partials applied to retries. Attempt 0 uses the caller's base `config` unchanged; retry `i` merges `per_attempt_override[i]` onto the base (the override's non-None fields replace; a None or unspecified field inherits the base, per the §6 null-skip semantics), and the last entry carries forward when the schedule is shorter than the retry count. The canonical use is an escalating temperature schedule that breaks the "temperature 0 replays the same output" determinism trap on a retried structured-output call. `complete()` never mutates the caller's `config` (each attempt config is a fresh copy), and a plain `RetryConfig` preserves the existing byte-identical replay. The per-attempt OTel span carries a new `openarmature.llm.retry_reason` attribute (`transient`) on retries, absent on the base attempt. This is the first half of proposal 0095; the structured-output reask half follows. Spec v0.91.0 is beyond the current v0.88.0 pin, so the behavior ships ahead of the pin (unit-tested); the conformance fixtures 061-066 ride the v0.17.0 pin bump. - **Adaptive call-level retry: structured-output reask** (proposal 0095, llm-provider §7.1, spec v0.91.0). The second half of 0095. `LlmRetryConfig` gains an opt-in `reask` builder (`Callable[[StructuredOutputInvalid], str]`). When present, a `structured_output_invalid` failure becomes retryable for that call (a call-level convenience, not a classifier change; without a builder it stays non-transient and raises on the first occurrence). On each such failure the loop appends two messages to a working transcript, the model's raw output as an `assistant` message and the builder's returned correction as a `user` message, so the retry is informed rather than a byte-identical replay. OA authors no prompt of its own (the caller owns every word beyond the model's output); the builder receives the raised `StructuredOutputInvalid` (its `raw_content` and `failure_description`). The transcript accumulates reask pairs across reask retries and consumes the `max_attempts` budget; a transient retry interleaved in a reask loop re-sends the accumulated transcript unchanged. `complete()` never mutates the caller's `messages` (each reask replaces the transcript with a fresh list rather than appending in place). The retry span's `openarmature.llm.retry_reason` is `reask` on a reask retry, `transient` otherwise. A reask always appends the model output as a fresh `assistant` message (never continues a trailing one): §3 requires the last message before a call to be `user`/`tool`, so the transcript never ends in `assistant`. Ships ahead of the pin (unit-tested); fixtures 062-066 ride the pin bump. +- **Langfuse observer: credentials-in construction with tracer-provider isolation** (proposals 0114 + 0116 + 0117, observability §6 / §8.9, spec v0.108.0 / v0.110.0 / v0.111.0). The Langfuse observer gains a second construction mode alongside today's caller-supplied client: `LangfuseObserver.from_credentials(public_key=..., secret_key=..., host=...)` (over the lower-level `LangfuseSDKAdapter.from_credentials(...)`) builds an OA-owned `Langfuse` client on a dedicated `TracerProvider` by default, so its observations no longer bind the global provider and leak onto the application's OTel backend. A Langfuse v4 client constructed with no `tracer_provider=` attaches its span processor to the globally-registered provider, so in any service that registers a global provider (the standard app-tracing setup) attaching the Langfuse observer silently exported every observation, prompts and completions included, to the app backend. Because the Langfuse SDK caches one client per `public_key`, a dedicated provider takes effect only when OA is the first constructor for that credential; OA reuses one isolated provider per credential and reads the actual binding back after construction. The invariant covers every payload OA harvests from the runtime -- the provider payload (`disable_provider_payload`), the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks), and a failed Tool / Embedding / Retriever / LLM observation's `error_message` / `error_type` -- but not the dimensions the caller deliberately attaches (`correlation_id` / `session_id` / `userId` / trace name / caller metadata), which stay verbatim as cross-backend join keys. When any construction-determinable channel is live and OA establishes the client is bound to a provider it did not isolate, construction fails loud with a categorized `LangfuseProviderIsolationUnavailable` before any observation is emitted, rather than leaking payloads to a shared backend; where OA cannot establish the binding at all (a future SDK), it suppresses every channel and logs a warning. The failed-observation error message is gated per-emission (not knowable at construction): on an un-isolatable provider it is omitted, retaining only the error category where one exists (a Tool failure has no category, so it carries no message-derived status either). A single `accept_shared_provider=True` opt-out turns the whole thing into a warn-and-proceed onto the shared provider. With no channel live (the default privacy posture), an un-isolatable client neither raises nor warns. The existing caller-supplied path (mode a) is unchanged and never mutated: a caller who builds their own client stays responsible for isolating its `tracer_provider`, and OA documents the remedy rather than reaching into the supplied client. The `secret_key` is accepted as a `pydantic.SecretStr`, masked in OA's own reprs and logs with the plaintext read only at the SDK call (`public_key` and `host` stay plain strings). Spec v0.108.0 / v0.110.0 / v0.111.0 are beyond the current v0.107.0 pin, so this ships ahead of the pin (unit-tested); the conformance fixtures (157 / 158, proposals 0115 / 0116 / 0117) ride the pin bump. The LLM error-message arm ships ahead of its spec formalization (proposal 0118, in progress at time of writing). ### Changed @@ -35,6 +36,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The ### Fixed +- **The Langfuse failure-isolation marker no longer carries the caught exception's message** (proposal 0118, observability §8.4). **Behavioral for the Langfuse mapping.** The `openarmature.failure_isolated` marker span wrote the caught exception's message into `observation.metadata.error_message`, but that span is a graph-mechanism marker no §8.4.x mapping table covers, so writing harvested exception content onto it was non-conforming over-emission: an exception message that can echo application data (PII, tool arguments, an upstream API error body) reached the Langfuse backend under every privacy setting, since no knob gated it. The marker now carries only `error_category` (plus the caller-supplied `failure_isolation_event_name` and the node name), matching the node span's treatment, so an isolated node failure and an ordinary node failure render the same. The full exception is unaffected on the OTel side, where the span still records it via `record_exception` on openarmature's private provider. A sweep of every bundled Langfuse handler against the same rule found no other unmapped harvested-content emission. - **The OTel `openarmature.llm.complete` span records the exception event on a failed attempt** (observability §4.2). A failed provider-call span carried `ERROR` status but no exception event; the node and invocation spans already recorded it, the LLM span did not. It now records the OTel semconv exception event (`exception.type` / `exception.message`) on every failed attempt, matching the sibling spans. Surfaced while wiring the proposal 0082 error-span fixtures. ## [0.16.0] — 2026-07-18 diff --git a/docs/agent/non-obvious-shapes.md b/docs/agent/non-obvious-shapes.md index 21a4aaa8..d94aa707 100644 --- a/docs/agent/non-obvious-shapes.md +++ b/docs/agent/non-obvious-shapes.md @@ -115,6 +115,14 @@ Different classes, same OTel-Logs export path. If both are attached against the `install_log_bridge` detects either handler class against the same provider and skips its own `addHandler` accordingly; the `openarmature.correlation_id` LogRecord factory still installs. The check is provider-scoped, so an application that intentionally attaches a handler against a DIFFERENT `LoggerProvider` (a separate logs pipeline) still gets the OA bridge against the OA provider; the helper only dedups when the SAME provider would receive duplicate emissions. +### `LangfuseObserver.from_credentials` can raise on an un-isolatable provider; construct OA's client first, or opt in + +The trace-side sibling of the `LoggerProvider` log-bridge gotcha above. When OA constructs the Langfuse client for you (`LangfuseObserver.from_credentials(...)` / `LangfuseSDKAdapter.from_credentials(...)`), it binds the client to a dedicated `TracerProvider` so OA's observations do not leak onto a provider shared with the application's OTel backend. But the Langfuse v4 SDK caches ONE client per `public_key` process-wide: if any client for that key was constructed first (the app called `Langfuse()`, used `langfuse.openai` / `@observe`, or an earlier OA call), the SDK returns the cached client and OA's dedicated provider is silently discarded. So isolation only holds when OA is the FIRST constructor for that credential. + +Rather than leak silently, OA detects the binding and fails closed. When a harvested-payload channel is live and OA establishes the client landed on a provider it did not isolate, `from_credentials` raises `LangfuseProviderIsolationUnavailable` at construction, surfacing an init-ordering bug you would otherwise never see. The guarded channels are the provider payload (`disable_provider_payload`) and the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks). Harvested error content is emitted only on the four mapped provider observations (LLM, Embedding, Tool, Retriever), where a failed observation's `error_message` / `error_type` is omitted per-emission on a shared provider with the error category retained; a graph-mechanism span (a node span, a failure-isolation marker) carries only the error category, and its exception detail goes to OA's isolated OTel span via `record_exception`, never to Langfuse. Caller-attached dimensions (`correlation_id`, `session_id`, `userId`, trace name, your own metadata) are join keys by design and stay verbatim. + +Remedies: construct OA's Langfuse client before any other client for that `public_key`; or pass `accept_shared_provider=True` to acknowledge a shared provider (OA warns and proceeds); or build your own client with an isolated `tracer_provider=` and pass it in via the caller-supplied path (`LangfuseObserver(client=LangfuseSDKAdapter(your_client))`), which OA never mutates. Under OA's default privacy posture (no payload channel live), an un-isolatable client is harmless and neither raises nor warns. + ### Three exception hierarchies; know which one your code catches `openarmature` exceptions split across three sibling hierarchies: diff --git a/src/openarmature/AGENTS.md b/src/openarmature/AGENTS.md index ad0f9688..cbda2dd5 100644 --- a/src/openarmature/AGENTS.md +++ b/src/openarmature/AGENTS.md @@ -1597,6 +1597,14 @@ Different classes, same OTel-Logs export path. If both are attached against the `install_log_bridge` detects either handler class against the same provider and skips its own `addHandler` accordingly; the `openarmature.correlation_id` LogRecord factory still installs. The check is provider-scoped, so an application that intentionally attaches a handler against a DIFFERENT `LoggerProvider` (a separate logs pipeline) still gets the OA bridge against the OA provider; the helper only dedups when the SAME provider would receive duplicate emissions. +### `LangfuseObserver.from_credentials` can raise on an un-isolatable provider; construct OA's client first, or opt in + +The trace-side sibling of the `LoggerProvider` log-bridge gotcha above. When OA constructs the Langfuse client for you (`LangfuseObserver.from_credentials(...)` / `LangfuseSDKAdapter.from_credentials(...)`), it binds the client to a dedicated `TracerProvider` so OA's observations do not leak onto a provider shared with the application's OTel backend. But the Langfuse v4 SDK caches ONE client per `public_key` process-wide: if any client for that key was constructed first (the app called `Langfuse()`, used `langfuse.openai` / `@observe`, or an earlier OA call), the SDK returns the cached client and OA's dedicated provider is silently discarded. So isolation only holds when OA is the FIRST constructor for that credential. + +Rather than leak silently, OA detects the binding and fails closed. When a harvested-payload channel is live and OA establishes the client landed on a provider it did not isolate, `from_credentials` raises `LangfuseProviderIsolationUnavailable` at construction, surfacing an init-ordering bug you would otherwise never see. The guarded channels are the provider payload (`disable_provider_payload`) and the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks). Harvested error content is emitted only on the four mapped provider observations (LLM, Embedding, Tool, Retriever), where a failed observation's `error_message` / `error_type` is omitted per-emission on a shared provider with the error category retained; a graph-mechanism span (a node span, a failure-isolation marker) carries only the error category, and its exception detail goes to OA's isolated OTel span via `record_exception`, never to Langfuse. Caller-attached dimensions (`correlation_id`, `session_id`, `userId`, trace name, your own metadata) are join keys by design and stay verbatim. + +Remedies: construct OA's Langfuse client before any other client for that `public_key`; or pass `accept_shared_provider=True` to acknowledge a shared provider (OA warns and proceeds); or build your own client with an isolated `tracer_provider=` and pass it in via the caller-supplied path (`LangfuseObserver(client=LangfuseSDKAdapter(your_client))`), which OA never mutates. Under OA's default privacy posture (no payload channel live), an un-isolatable client is harmless and neither raises nor warns. + ### Three exception hierarchies; know which one your code catches `openarmature` exceptions split across three sibling hierarchies: diff --git a/src/openarmature/observability/langfuse/__init__.py b/src/openarmature/observability/langfuse/__init__.py index 92e35c3c..1d02fc4f 100644 --- a/src/openarmature/observability/langfuse/__init__.py +++ b/src/openarmature/observability/langfuse/__init__.py @@ -40,6 +40,7 @@ ObservationLevel, ObservationType, ) +from .errors import LangfuseProviderIsolationUnavailable from .observer import LangfuseObserver from .trace_id import langfuse_trace_id @@ -62,6 +63,7 @@ "LangfuseGenerationHandle", "LangfuseObservation", "LangfuseObserver", + "LangfuseProviderIsolationUnavailable", "LangfuseSpanHandle", "LangfuseTrace", "LangfuseUsage", diff --git a/src/openarmature/observability/langfuse/adapter.py b/src/openarmature/observability/langfuse/adapter.py index b1007405..d0a034bd 100644 --- a/src/openarmature/observability/langfuse/adapter.py +++ b/src/openarmature/observability/langfuse/adapter.py @@ -34,18 +34,27 @@ from __future__ import annotations import json +import threading from contextlib import ExitStack from datetime import datetime -from typing import TYPE_CHECKING, Any, cast - -from .client import LangfuseGenerationHandle, LangfuseSpanHandle, LangfuseUsage, ObservationLevel +from typing import Any, cast + +from pydantic import SecretStr + +from .client import ( + ISOLATION_ISOLATED, + ISOLATION_LEAKED, + ISOLATION_SHARED_ACCEPTED, + ISOLATION_UNDETECTABLE, + LangfuseGenerationHandle, + LangfuseSpanHandle, + LangfuseUsage, + ObservationLevel, +) from .trace_id import _is_uuid, _to_otel_trace_id -if TYPE_CHECKING: - from langfuse import Langfuse - try: - from langfuse import propagate_attributes + from langfuse import Langfuse, propagate_attributes from langfuse.types import TraceContext except ImportError as exc: # pragma: no cover - exercised by extras-not-installed path raise ImportError( @@ -72,6 +81,57 @@ def _stringify_metadata(metadata: dict[str, Any] | None) -> dict[str, str]: return out +# Proposal 0116 (observability §6 payload-leak invariant). OA reuses ONE isolated +# TracerProvider per credential (public_key), so a second surface that the SDK's +# per-public_key resource-manager singleton resolves to OA's own provider +# satisfies the invariant instead of reading as a leak. Process-wide and +# lock-guarded, mirroring the scope of the SDK's own singleton. +_ISOLATED_PROVIDERS: dict[str, Any] = {} +_ISOLATED_PROVIDERS_LOCK = threading.Lock() + + +def _reuse_isolated_provider(public_key: str) -> Any: + """Return the one isolated ``TracerProvider`` OA uses for ``public_key``, + building it on first use. A bare provider is the whole isolation: the SDK + adds its own span processor to whatever provider it is handed and does not + register a handed-in provider globally.""" + from opentelemetry.sdk.trace import TracerProvider + + with _ISOLATED_PROVIDERS_LOCK: + provider = _ISOLATED_PROVIDERS.get(public_key) + if provider is None: + provider = TracerProvider() + _ISOLATED_PROVIDERS[public_key] = provider + return provider + + +def _classify_isolation(client: Any, provider: Any, accept_shared_provider: bool) -> str: + """Classify what the SDK actually bound the client's observations to. + + The SDK caches one resource manager per public_key, so a handed-in provider + is honored only when OA is the first constructor for that key; otherwise the + cached client keeps its original provider and OA's would-be-isolated provider + is discarded. + """ + # Proposal 0116 payload-leak invariant: OA establishes the binding to pick the + # raise / suppress arm. + if accept_shared_provider: + return ISOLATION_SHARED_ACCEPTED + # Guarded read of the SDK-internal binding: a future SDK that stops exposing + # it leaves us unable to establish isolation, so the observer takes the + # portable suppress floor rather than a false all-clear. + resources = getattr(client, "_resources", None) + bound = getattr(resources, "tracer_provider", None) if resources is not None else None + if bound is None: + # A client with tracing disabled legitimately has no provider and exports + # nothing, so there is no leak to guard -- distinct from a future SDK that + # hides the binding (the genuine undetectable case). + if not getattr(client, "_tracing_enabled", True): + return ISOLATION_ISOLATED + return ISOLATION_UNDETECTABLE + return ISOLATION_ISOLATED if bound is provider else ISOLATION_LEAKED + + class _SpanHandle: """Wraps a langfuse LangfuseSpan / LangfuseGeneration to satisfy :class:`LangfuseSpanHandle` / :class:`LangfuseGenerationHandle`. @@ -193,6 +253,9 @@ class LangfuseSDKAdapter: def __init__(self, client: Langfuse) -> None: self._client = client + # 0116 isolation status; None for a caller-supplied client (mode a), + # set by from_credentials for the OA-constructed path (mode b). + self._isolation_status: str | None = None # Trace info cache, applied via propagate_attributes around # EVERY observation (not just the first). Langfuse v4's trace # name/metadata processing uses last-attribute-wins semantics, @@ -203,6 +266,59 @@ def __init__(self, client: Langfuse) -> None: # consistent. Cache cleanup is deferred to a future PR. self._trace_info: dict[str, dict[str, Any]] = {} + @classmethod + def from_credentials( + cls, + *, + public_key: str, + secret_key: SecretStr, + host: str | None = None, + accept_shared_provider: bool = False, + **langfuse_kwargs: Any, + ) -> LangfuseSDKAdapter: + """Build the adapter over an OA-constructed ``Langfuse`` client, from + credentials rather than a caller-supplied instance. + + ``secret_key`` is a ``pydantic.SecretStr``: it is masked in OA's own + reprs and logs, and its plaintext is read only at the ``Langfuse`` call. + The SDK holds the plaintext thereafter, so an SDK-side construction + failure can still surface it in that frame. + + By default OA constructs the client on a dedicated ``TracerProvider`` + (reused per credential) so its observations do not share a provider with + the application. The Langfuse SDK caches one resource manager per + ``public_key``, so a client for that key constructed elsewhere first + keeps its original provider and OA's dedicated provider is discarded; + ``from_credentials`` reads the actual binding back and records an + isolation status (``isolated`` / ``leaked`` / ``undetectable``) that + :meth:`LangfuseObserver.from_credentials` turns into the raise / suppress + arms. ``accept_shared_provider=True`` skips isolation, binds the ambient + provider, and records ``shared_accepted``. + + ``langfuse_kwargs`` pass through to the ``Langfuse`` constructor + (``release`` / ``environment`` / ...); ``tracer_provider`` is managed + here and may not be supplied through it. Client-config kwargs the SDK + applies only while building its own provider (``sample_rate`` and the + ``environment`` / ``release`` Resource attributes) are not reflected on + OA's dedicated provider under isolation. + """ + if "tracer_provider" in langfuse_kwargs: + raise ValueError( + "tracer_provider is managed by from_credentials " + "(via `accept_shared_provider`); do not pass it through langfuse_kwargs" + ) + provider = None if accept_shared_provider else _reuse_isolated_provider(public_key) + client = Langfuse( + public_key=public_key, + secret_key=secret_key.get_secret_value(), + host=host, + tracer_provider=provider, + **langfuse_kwargs, + ) + adapter = cls(client) + adapter._isolation_status = _classify_isolation(client, provider, accept_shared_provider) + return adapter + def trace( self, *, diff --git a/src/openarmature/observability/langfuse/client.py b/src/openarmature/observability/langfuse/client.py index c0412625..84b6f604 100644 --- a/src/openarmature/observability/langfuse/client.py +++ b/src/openarmature/observability/langfuse/client.py @@ -38,6 +38,16 @@ # Langfuse-supported `level` values per spec §8.4.2 (statusMessage pair). ObservationLevel = Literal["DEFAULT", "DEBUG", "INFO", "WARNING", "ERROR"] +# Isolation status of an OA-constructed Langfuse client (proposals 0114 / 0116 / +# 0117 payload-leak invariant). Recorded on LangfuseSDKAdapter by from_credentials +# and read by the observer's per-emission payload-leak gate. Defined here (SDK-free) +# so the observer can consult it without importing the SDK-gated adapter. ``None`` +# means a caller-supplied client (mode a), where the caller owns the provider. +ISOLATION_ISOLATED = "isolated" +ISOLATION_LEAKED = "leaked" +ISOLATION_UNDETECTABLE = "undetectable" +ISOLATION_SHARED_ACCEPTED = "shared_accepted" + @dataclass class LangfuseUsage: diff --git a/src/openarmature/observability/langfuse/errors.py b/src/openarmature/observability/langfuse/errors.py new file mode 100644 index 00000000..57b46cdb --- /dev/null +++ b/src/openarmature/observability/langfuse/errors.py @@ -0,0 +1,24 @@ +# Categorized errors for the Langfuse observability mapping. Mirrors the +# llm/errors.py pattern: a ``category`` class attribute carrying the canonical +# string so callers dispatch on the category rather than matching the message. +# +# Spec basis: observability §6 payload-leak invariant (proposal 0116). Raised on +# the raise arm -- OA establishes that its payload-bearing Langfuse observations +# would reach a TracerProvider shared with the application, and the caller has +# not accepted a shared provider. + +"""Categorized errors for the Langfuse observability mapping.""" + +from __future__ import annotations + + +class LangfuseProviderIsolationUnavailable(Exception): + """OA cannot keep its payload-bearing Langfuse observations off a + TracerProvider shared with the application, and the caller has not accepted + a shared provider, so construction fails loud rather than leaking. + + Carries the canonical ``category`` string so callers can dispatch on it + without matching the message. + """ + + category = "langfuse_provider_isolation_unavailable" diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 2cd25227..a5231a55 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -23,12 +23,15 @@ from __future__ import annotations import json +import logging import uuid from collections.abc import Callable, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any, cast +from pydantic import SecretStr + from openarmature.graph.events import ( EmbeddingEvent, EmbeddingFailedEvent, @@ -50,6 +53,9 @@ from openarmature.observability.llm_event import _token_budget_evaluations from .client import ( + ISOLATION_LEAKED, + ISOLATION_SHARED_ACCEPTED, + ISOLATION_UNDETECTABLE, LangfuseClient, LangfuseGenerationHandle, LangfuseSpanHandle, @@ -57,6 +63,9 @@ ObservationLevel, ) +# §7 log records for the 0116 payload-leak suppress / opt-out arms. +_logger = logging.getLogger("openarmature.observability") + # §5.5.5 / §8.7 truncation: when the serialized payload exceeds the # configured cap, the marker below is appended and the unparseable # JSON serves as the "this was truncated" signal in Langfuse's input @@ -442,6 +451,110 @@ def __post_init__(self) -> None: f"minimum of {_PAYLOAD_MIN_BYTES} bytes" ) + @classmethod + def from_credentials( + cls, + *, + public_key: str, + secret_key: SecretStr, + host: str | None = None, + accept_shared_provider: bool = False, + langfuse_kwargs: dict[str, Any] | None = None, + **observer_kwargs: Any, + ) -> LangfuseObserver: + """Build the observer over an OA-constructed Langfuse client from + credentials, isolating the client's ``TracerProvider`` by default so its + payload-bearing observations do not share a provider with the + application. + + When this observer emits provider payloads (``disable_provider_payload`` + is False) and OA cannot isolate the client -- the Langfuse SDK's + per-``public_key`` singleton returned a client bound to a provider OA did + not establish as isolated -- construction fails loud with + :class:`LangfuseProviderIsolationUnavailable` before any observation is + emitted, rather than leaking payloads to a shared backend. If OA cannot + establish the binding at all (a future SDK), it instead suppresses its + own payloads and logs a warning. ``accept_shared_provider=True`` opts out + of both: OA logs a warning and proceeds onto the shared provider. + + The default ``disable_provider_payload=True`` emits no payloads, so an + un-isolatable client is harmless and neither raises nor warns. + + ``langfuse_kwargs`` are forwarded to the Langfuse client constructor + (``release`` / ``environment`` / ...); ``observer_kwargs`` are the + ordinary observer fields (``disable_provider_payload`` / + ``disable_llm_spans`` / ...). + """ + # Any construction-determinable payload channel makes a shared provider a + # leak (0117): the provider payload, the Trace-level state payload, or a + # supplied trace_input/output-from-state hook (a supplied hook emits + # regardless of the knob, so treat *supplied* as potentially-live). The + # failed-observation error-message channel is not construction-knowable, + # so it does not gate here; it is suppressed per-emission via the status. + emits_payloads = ( + not bool(observer_kwargs.get("disable_provider_payload", True)) + or not bool(observer_kwargs.get("disable_state_payload", True)) + or observer_kwargs.get("trace_input_from_state") is not None + or observer_kwargs.get("trace_output_from_state") is not None + ) + + from openarmature.observability.langfuse.adapter import LangfuseSDKAdapter + + client = LangfuseSDKAdapter.from_credentials( + public_key=public_key, + secret_key=secret_key, + host=host, + accept_shared_provider=accept_shared_provider, + **(langfuse_kwargs or {}), + ) + + if emits_payloads: + status = client._isolation_status + if status == ISOLATION_LEAKED: + from openarmature.observability.langfuse.errors import ( + LangfuseProviderIsolationUnavailable, + ) + + raise LangfuseProviderIsolationUnavailable( + "OA constructed a Langfuse client whose observations would reach a " + "TracerProvider shared with the application (the SDK caches one client " + "per public_key, and one was already bound to a provider OA did not " + "isolate), and a payload channel is live so payloads would leak. Pass " + "accept_shared_provider=True to proceed onto the shared provider, or " + "construct OA's Langfuse client before any other client for this key." + ) + if status == ISOLATION_UNDETECTABLE: + # Portable floor: cannot establish the binding, so suppress EVERY + # construction-time channel (fail-safe) and log a warning. The + # error-message channel is suppressed per-emission via the status. + _logger.warning( + "cannot establish the Langfuse client's TracerProvider binding; " + "suppressing all provider and state payloads to avoid a possible " + "leak to a shared provider" + ) + observer_kwargs["disable_provider_payload"] = True + observer_kwargs["disable_state_payload"] = True + observer_kwargs["trace_input_from_state"] = None + observer_kwargs["trace_output_from_state"] = None + elif status == ISOLATION_SHARED_ACCEPTED: + _logger.warning( + "accept_shared_provider=True with payload channels enabled: OA's " + "Langfuse observations may reach a TracerProvider shared with the " + "application (acknowledged)" + ) + + return cls(client=client, **observer_kwargs) + + def _omit_harvested_error(self) -> bool: + # 0117: on a provider OA did not establish is isolated -- LEAKED, or the + # non-detectable suppress floor -- a failed observation's harvested + # error_message / error_type must not reach the shared provider. Isolated, + # opted-in, and caller-supplied (mode a) clients emit normally. + return getattr(self.client, "_isolation_status", None) in ( + ISOLATION_LEAKED, + ISOLATION_UNDETECTABLE, + ) + async def __call__( self, event: ObserverEvent, @@ -807,9 +920,13 @@ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: fan_out_index_chain=event.fan_out_index_chain, branch_name_chain=event.branch_name_chain, ) + # The caught exception's MESSAGE is deliberately absent: this marker is a + # graph-mechanism span, which no §8.4.x table maps, so writing harvested + # exception content onto it is non-conforming over-emission (0118). Like + # the node Span, it carries only the error category; the full exception + # reaches the OTel span via record_exception on OA's private provider. metadata: dict[str, Any] = { "failure_isolation_event_name": event.event_name, - "error_message": event.caught_exception.message, } if event.namespace: metadata["failure_isolation_node"] = event.namespace[-1] @@ -1880,9 +1997,10 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: # observation. error_type is null when no impl-side type was # available; the metadata key is omitted in that case so the # absence-is-meaningful semantic is preserved. - if event.error_type is not None: - metadata["error_type"] = event.error_type - metadata["error_message"] = event.error_message + if not self._omit_harvested_error(): + if event.error_type is not None: + metadata["error_type"] = event.error_type + metadata["error_message"] = event.error_message model_parameters: dict[str, Any] = dict(event.request_params or {}) input_value: Any = None output_value: Any = None @@ -1978,10 +2096,11 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: status_message: str | None = None if isinstance(event, ToolCallFailedEvent): level = "ERROR" - if event.error_type is not None: - metadata["error_type"] = event.error_type - metadata["error_message"] = event.error_message - status_message = event.error_message + if not self._omit_harvested_error(): + if event.error_type is not None: + metadata["error_type"] = event.error_type + metadata["error_message"] = event.error_message + status_message = event.error_message target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) handle = self.client.tool( trace_id=target_trace_id, @@ -2085,9 +2204,10 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non # Failure path: request-side input_count survives; the response-derived # rows do not. No output. ERROR level + category-as-statusMessage. metadata["openarmature_input_count"] = len(event.input_strings) - if event.error_type is not None: - metadata["error_type"] = event.error_type - metadata["error_message"] = event.error_message + if not self._omit_harvested_error(): + if event.error_type is not None: + metadata["error_type"] = event.error_type + metadata["error_message"] = event.error_message target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) handle = self.client.embedding( trace_id=target_trace_id, @@ -2208,9 +2328,10 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: return # Failure path: the request-side metadata survives; the response-derived # rows do not. No output. ERROR level + category-as-statusMessage. - if event.error_type is not None: - metadata["error_type"] = event.error_type - metadata["error_message"] = event.error_message + if not self._omit_harvested_error(): + if event.error_type is not None: + metadata["error_type"] = event.error_type + metadata["error_message"] = event.error_message target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) handle = self.client.retriever( trace_id=target_trace_id, diff --git a/tests/unit/test_langfuse_provider_isolation.py b/tests/unit/test_langfuse_provider_isolation.py new file mode 100644 index 00000000..ef174358 --- /dev/null +++ b/tests/unit/test_langfuse_provider_isolation.py @@ -0,0 +1,265 @@ +"""Langfuse client ownership + TracerProvider isolation (0114 / 0116).""" + +# Spec basis: observability §6 / §8.9 (proposals 0114 + 0116, the payload-leak +# invariant). Behavior ships ahead of the v0.110.0 pin; these unit tests pin the +# credentials-in construction, the per-credential isolated-provider reuse, the +# post-construct isolation classification, and the raise / suppress / opt-out +# arms. The end-to-end provider-leak assertions are conformance fixtures 157/158. + +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from pydantic import SecretStr + +pytest.importorskip("langfuse") + +from openarmature.observability.langfuse import ( # noqa: E402 + LangfuseObserver, + LangfuseProviderIsolationUnavailable, +) +from openarmature.observability.langfuse import adapter as _adapter_mod # noqa: E402 +from openarmature.observability.langfuse.adapter import ( # noqa: E402 + ISOLATION_ISOLATED, + ISOLATION_LEAKED, + ISOLATION_SHARED_ACCEPTED, + ISOLATION_UNDETECTABLE, + LangfuseSDKAdapter, +) + +_ADAPTER = "openarmature.observability.langfuse.adapter" + + +@pytest.fixture(autouse=True) +def _clear_provider_registry() -> Any: # pyright: ignore[reportUnusedFunction] + # The isolated-provider registry is process-wide; clear it around each test so + # reuse assertions and fresh-build assertions do not cross-contaminate. + _adapter_mod._ISOLATED_PROVIDERS.clear() + yield + _adapter_mod._ISOLATED_PROVIDERS.clear() + + +def _langfuse_binding_passed(**kwargs: Any) -> MagicMock: + # OA won the singleton: the SDK bound the provider we handed it. + client = MagicMock() + client._resources.tracer_provider = kwargs["tracer_provider"] + return client + + +def _langfuse_binding_other(**kwargs: Any) -> MagicMock: + # The singleton returned a client bound to someone else's provider. + client = MagicMock() + client._resources.tracer_provider = object() + return client + + +def _langfuse_no_resources(**kwargs: Any) -> MagicMock: + # A future SDK that does not expose the binding at all. + return MagicMock(spec=[]) + + +def _langfuse_tracing_disabled(**kwargs: Any) -> MagicMock: + # Tracing disabled: no provider bound, nothing exports (not a hidden binding). + client = MagicMock() + client._resources.tracer_provider = None + client._tracing_enabled = False + return client + + +def _hook(state: Any) -> Any: + return state + + +# --- adapter: construction + isolation classification ------------------------- + + +def test_isolate_default_binds_dedicated_provider_status_isolated() -> None: + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_passed) as mock_lf: + adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk"), host="h") + assert mock_lf.call_args.kwargs["tracer_provider"] is not None + assert adapter._isolation_status == ISOLATION_ISOLATED + + +def test_singleton_binding_other_provider_status_leaked() -> None: + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_other): + adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert adapter._isolation_status == ISOLATION_LEAKED + + +def test_binding_not_exposed_status_undetectable() -> None: + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_no_resources): + adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert adapter._isolation_status == ISOLATION_UNDETECTABLE + + +def test_accept_shared_provider_binds_ambient_status_shared_accepted() -> None: + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_other) as mock_lf: + adapter = LangfuseSDKAdapter.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), accept_shared_provider=True + ) + assert mock_lf.call_args.kwargs["tracer_provider"] is None + assert adapter._isolation_status == ISOLATION_SHARED_ACCEPTED + + +def test_isolated_provider_reused_per_public_key() -> None: + providers: list[Any] = [] + + def capture(**kwargs: Any) -> MagicMock: + providers.append(kwargs["tracer_provider"]) + client = MagicMock() + client._resources.tracer_provider = kwargs["tracer_provider"] + return client + + with patch(f"{_ADAPTER}.Langfuse", side_effect=capture): + LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + LangfuseSDKAdapter.from_credentials(public_key="other", secret_key=SecretStr("sk")) + assert providers[0] is providers[1] # same key reuses one provider + assert providers[2] is not providers[0] # a different key gets its own + + +def test_managed_tracer_provider_kwarg_rejected() -> None: + from opentelemetry.sdk.trace import TracerProvider + + with patch(f"{_ADAPTER}.Langfuse"): + with pytest.raises(ValueError, match="tracer_provider"): + LangfuseSDKAdapter.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), tracer_provider=TracerProvider() + ) + + +def test_mode_a_supplied_client_is_wrapped_not_replaced() -> None: + sentinel: Any = object() + adapter = LangfuseSDKAdapter(sentinel) + assert adapter._client is sentinel + assert adapter._isolation_status is None # caller owns the provider (mode a) + + +# --- observer: the raise / suppress / opt-out policy -------------------------- + + +def _patched_adapter(status: str) -> Any: + stub = MagicMock() + stub._isolation_status = status + return patch.object(LangfuseSDKAdapter, "from_credentials", return_value=stub) + + +def test_observer_payloads_off_never_raises_even_on_leak() -> None: + # Default disable_provider_payload=True: no payloads, so an un-isolatable + # client is harmless. + with _patched_adapter(ISOLATION_LEAKED): + obs = LangfuseObserver.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert obs.disable_provider_payload is True + + +def test_observer_payloads_on_leak_raises() -> None: + with _patched_adapter(ISOLATION_LEAKED): + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), disable_provider_payload=False + ) + + +def test_observer_payloads_on_isolated_proceeds() -> None: + with _patched_adapter(ISOLATION_ISOLATED): + obs = LangfuseObserver.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), disable_provider_payload=False + ) + assert obs.disable_provider_payload is False + + +def test_observer_undetectable_suppresses_all_channels_and_warns(caplog: Any) -> None: + with _patched_adapter(ISOLATION_UNDETECTABLE): + with caplog.at_level("WARNING", logger="openarmature.observability"): + obs = LangfuseObserver.from_credentials( + public_key="pk", + secret_key=SecretStr("sk"), + disable_provider_payload=False, + disable_state_payload=False, + trace_input_from_state=_hook, + ) + # Suppress-all: every construction-time channel forced off (fail-safe). + assert obs.disable_provider_payload is True + assert obs.disable_state_payload is True + assert obs.trace_input_from_state is None + assert "suppressing all provider and state payloads" in caplog.text + + +def test_observer_accept_shared_provider_warns_and_proceeds(caplog: Any) -> None: + with _patched_adapter(ISOLATION_SHARED_ACCEPTED): + with caplog.at_level("WARNING", logger="openarmature.observability"): + obs = LangfuseObserver.from_credentials( + public_key="pk", + secret_key=SecretStr("sk"), + accept_shared_provider=True, + disable_provider_payload=False, + ) + assert obs.disable_provider_payload is False # proceeds (acknowledged leak) + assert "acknowledged" in caplog.text + + +def test_observer_forwards_langfuse_kwargs_and_observer_kwargs() -> None: + with patch.object(LangfuseSDKAdapter, "from_credentials") as mock_fc: + mock_fc.return_value._isolation_status = ISOLATION_ISOLATED + obs = LangfuseObserver.from_credentials( + public_key="pk", + secret_key=SecretStr("sk"), + langfuse_kwargs={"environment": "prod"}, + disable_llm_spans=True, + ) + assert mock_fc.call_args.kwargs["environment"] == "prod" + assert mock_fc.call_args.kwargs["accept_shared_provider"] is False + assert obs.disable_llm_spans is True + + +def test_state_payload_channel_on_leak_raises() -> None: + # 0117: the state payload is a leak channel independent of the provider knob. + with _patched_adapter(ISOLATION_LEAKED): + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), disable_state_payload=False + ) + + +def test_supplied_hook_on_leak_raises() -> None: + # A supplied trace_input_from_state hook emits regardless of the knob, so a + # supplied hook is a live channel even under the default privacy posture. + with _patched_adapter(ISOLATION_LEAKED): + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), trace_input_from_state=_hook + ) + + +def test_all_channels_off_on_leak_does_not_raise() -> None: + # Fully-locked-down posture: no construction-time channel live, so an + # un-isolatable client is harmless at construction (the error-message channel + # is handled per-emission, verified end-to-end by conformance fixture 158). + with _patched_adapter(ISOLATION_LEAKED): + obs = LangfuseObserver.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert obs.disable_provider_payload is True + + +@pytest.mark.parametrize( + "status,expected", + [ + (ISOLATION_LEAKED, True), + (ISOLATION_UNDETECTABLE, True), + (ISOLATION_ISOLATED, False), + (ISOLATION_SHARED_ACCEPTED, False), + (None, False), + ], +) +def test_omit_harvested_error_gate(status: Any, expected: bool) -> None: + # The per-emission gate: a failed observation's harvested error_message / + # error_type is dropped only when OA has not established isolation. + obs = LangfuseObserver(client=MagicMock(_isolation_status=status)) + assert obs._omit_harvested_error() is expected + + +def test_tracing_disabled_classified_isolated_not_undetectable() -> None: + # A tracing-disabled client has no provider but exports nothing, so it is + # no-leak (ISOLATED), not the suppress floor (UNDETECTABLE). + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_tracing_disabled): + adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert adapter._isolation_status == ISOLATION_ISOLATED From 67024662ec3983fbf55c57bac1d7674e263d49ca Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sun, 9 Aug 2026 20:00:00 -0700 Subject: [PATCH 2/4] Enforce the payload-leak invariant at emission The isolation arms ran only in LangfuseObserver.from_credentials, so building an observer over a LangfuseSDKAdapter.from_credentials client reached emission on a shared provider with nothing raised, warned, or suppressed. They move to __post_init__, which covers every path to an OA-constructed client, and the payload sites now consult the client's isolation status directly so a knob reopened after construction cannot reopen the leak either. Provider binding gets three corrections. The accept-a-shared-provider opt-out resolves the provider the application registered instead of passing None, which made the SDK build and globally register its own and capture OTel's single-assignment slot. A tracing-disabled client is classified before the binding is read, since it exports nothing whatever provider it holds. The opt-out no longer short-circuits classification, so a client the SDK resolved onto OA's own isolated provider is not recorded as shared. Blank credentials are rejected rather than falling through to the SDK's ambient environment fallback, a sample_rate is applied to the isolated provider, a construction that the SDK's per-credential cache will discard says so, and the isolation warnings are no longer nested under the payload check that left them silent by default. A structured_output_invalid error message quotes the model's own output, so it now follows the payload knob as well as the isolation gate. Adds a canary test that plants a sentinel in every harvested input and asserts none of them reach an un-isolated provider. It asserts the invariant rather than the individual emission sites, so a channel nobody enumerated fails the moment it is added; reintroducing either of the two leaks found in review makes it fail. Docs and both examples now lead with from_credentials rather than the un-isolated construction they taught before, and the claim that the failure-isolation message rides record_exception is corrected to the span attribute it actually uses. --- CHANGELOG.md | 4 +- docs/agent/non-obvious-shapes.md | 9 +- docs/concepts/observability.md | 25 ++ docs/examples/langfuse-observability.md | 20 +- docs/examples/production-observability.md | 3 +- examples/langfuse-observability/main.py | 19 +- examples/production-observability/main.py | 10 +- src/openarmature/AGENTS.md | 9 +- .../observability/langfuse/adapter.py | 131 ++++++++-- .../observability/langfuse/errors.py | 11 +- .../observability/langfuse/observer.py | 167 ++++++++----- .../unit/test_langfuse_payload_leak_canary.py | 230 ++++++++++++++++++ .../unit/test_langfuse_provider_isolation.py | 215 +++++++++++++++- 13 files changed, 734 insertions(+), 119 deletions(-) create mode 100644 tests/unit/test_langfuse_payload_leak_canary.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c29587c4..3b23ab54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The - **Langfuse parallel-branches mapping parity** (proposal 0088, observability §8.4.8 / §8.3 / §8.4.2 / §3.4, spec v0.83.0). Brings the Langfuse observer's parallel-branches rendering to parity with the OTel side. The observer already synthesized the three-level Observation tree (the parallel-branches node Span, a per-branch dispatch Span named by the `branch_name`, and the branch's inner observations) and already emitted the dispatch-span `parallel_branches_parent_node_name` and `branch_name`; the two node-span attributes `parallel_branches_branch_count` and `parallel_branches_error_policy` are now flattened onto the node Span's `observation.metadata` (mirroring the `fan_out_*` attributes), the one §8.4.2 row the observer had never mapped. The three `parallel_branches_*` keys join the reserved caller-metadata set (26 to 29), so a caller passing one as invocation metadata is rejected at the `invoke()` boundary rather than shadowing the OA-emitted field. The OTel side was already complete. Conformance fixture 136 (the dedicated three-level-tree pin) is un-deferred; fixture 030's incidental coverage stands. - **Adaptive call-level retry: per-attempt request override** (proposal 0095, llm-provider §7.1, spec v0.91.0). The LLM-completion call-level retry loop gains an opt-in per-attempt request override. A new `LlmRetryConfig` (the llm-provider-scoped superset of the generic `RetryConfig`, exported from `openarmature.llm`) carries a `per_attempt_override`: a schedule of `RuntimeConfig` partials applied to retries. Attempt 0 uses the caller's base `config` unchanged; retry `i` merges `per_attempt_override[i]` onto the base (the override's non-None fields replace; a None or unspecified field inherits the base, per the §6 null-skip semantics), and the last entry carries forward when the schedule is shorter than the retry count. The canonical use is an escalating temperature schedule that breaks the "temperature 0 replays the same output" determinism trap on a retried structured-output call. `complete()` never mutates the caller's `config` (each attempt config is a fresh copy), and a plain `RetryConfig` preserves the existing byte-identical replay. The per-attempt OTel span carries a new `openarmature.llm.retry_reason` attribute (`transient`) on retries, absent on the base attempt. This is the first half of proposal 0095; the structured-output reask half follows. Spec v0.91.0 is beyond the current v0.88.0 pin, so the behavior ships ahead of the pin (unit-tested); the conformance fixtures 061-066 ride the v0.17.0 pin bump. - **Adaptive call-level retry: structured-output reask** (proposal 0095, llm-provider §7.1, spec v0.91.0). The second half of 0095. `LlmRetryConfig` gains an opt-in `reask` builder (`Callable[[StructuredOutputInvalid], str]`). When present, a `structured_output_invalid` failure becomes retryable for that call (a call-level convenience, not a classifier change; without a builder it stays non-transient and raises on the first occurrence). On each such failure the loop appends two messages to a working transcript, the model's raw output as an `assistant` message and the builder's returned correction as a `user` message, so the retry is informed rather than a byte-identical replay. OA authors no prompt of its own (the caller owns every word beyond the model's output); the builder receives the raised `StructuredOutputInvalid` (its `raw_content` and `failure_description`). The transcript accumulates reask pairs across reask retries and consumes the `max_attempts` budget; a transient retry interleaved in a reask loop re-sends the accumulated transcript unchanged. `complete()` never mutates the caller's `messages` (each reask replaces the transcript with a fresh list rather than appending in place). The retry span's `openarmature.llm.retry_reason` is `reask` on a reask retry, `transient` otherwise. A reask always appends the model output as a fresh `assistant` message (never continues a trailing one): §3 requires the last message before a call to be `user`/`tool`, so the transcript never ends in `assistant`. Ships ahead of the pin (unit-tested); fixtures 062-066 ride the pin bump. -- **Langfuse observer: credentials-in construction with tracer-provider isolation** (proposals 0114 + 0116 + 0117, observability §6 / §8.9, spec v0.108.0 / v0.110.0 / v0.111.0). The Langfuse observer gains a second construction mode alongside today's caller-supplied client: `LangfuseObserver.from_credentials(public_key=..., secret_key=..., host=...)` (over the lower-level `LangfuseSDKAdapter.from_credentials(...)`) builds an OA-owned `Langfuse` client on a dedicated `TracerProvider` by default, so its observations no longer bind the global provider and leak onto the application's OTel backend. A Langfuse v4 client constructed with no `tracer_provider=` attaches its span processor to the globally-registered provider, so in any service that registers a global provider (the standard app-tracing setup) attaching the Langfuse observer silently exported every observation, prompts and completions included, to the app backend. Because the Langfuse SDK caches one client per `public_key`, a dedicated provider takes effect only when OA is the first constructor for that credential; OA reuses one isolated provider per credential and reads the actual binding back after construction. The invariant covers every payload OA harvests from the runtime -- the provider payload (`disable_provider_payload`), the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks), and a failed Tool / Embedding / Retriever / LLM observation's `error_message` / `error_type` -- but not the dimensions the caller deliberately attaches (`correlation_id` / `session_id` / `userId` / trace name / caller metadata), which stay verbatim as cross-backend join keys. When any construction-determinable channel is live and OA establishes the client is bound to a provider it did not isolate, construction fails loud with a categorized `LangfuseProviderIsolationUnavailable` before any observation is emitted, rather than leaking payloads to a shared backend; where OA cannot establish the binding at all (a future SDK), it suppresses every channel and logs a warning. The failed-observation error message is gated per-emission (not knowable at construction): on an un-isolatable provider it is omitted, retaining only the error category where one exists (a Tool failure has no category, so it carries no message-derived status either). A single `accept_shared_provider=True` opt-out turns the whole thing into a warn-and-proceed onto the shared provider. With no channel live (the default privacy posture), an un-isolatable client neither raises nor warns. The existing caller-supplied path (mode a) is unchanged and never mutated: a caller who builds their own client stays responsible for isolating its `tracer_provider`, and OA documents the remedy rather than reaching into the supplied client. The `secret_key` is accepted as a `pydantic.SecretStr`, masked in OA's own reprs and logs with the plaintext read only at the SDK call (`public_key` and `host` stay plain strings). Spec v0.108.0 / v0.110.0 / v0.111.0 are beyond the current v0.107.0 pin, so this ships ahead of the pin (unit-tested); the conformance fixtures (157 / 158, proposals 0115 / 0116 / 0117) ride the pin bump. The LLM error-message arm ships ahead of its spec formalization (proposal 0118, in progress at time of writing). +- **Langfuse observer: credentials-in construction with tracer-provider isolation** (proposals 0114 + 0116 + 0117, observability §6 / §8.9, spec v0.108.0 / v0.110.0 / v0.111.0). The Langfuse observer gains a second construction mode alongside today's caller-supplied client: `LangfuseObserver.from_credentials(public_key=..., secret_key=..., host=...)` (over the lower-level `LangfuseSDKAdapter.from_credentials(...)`) builds an OA-owned `Langfuse` client on a dedicated `TracerProvider` by default, so its observations no longer bind the global provider and leak onto the application's OTel backend. A Langfuse v4 client constructed with no `tracer_provider=` attaches its span processor to the globally-registered provider, so in any service that registers a global provider (the standard app-tracing setup) attaching the Langfuse observer silently exported every observation, prompts and completions included, to the app backend. Because the Langfuse SDK caches one client per `public_key`, a dedicated provider takes effect only when OA is the first constructor for that credential; OA reuses one isolated provider per credential and reads the actual binding back after construction. The invariant covers every payload OA harvests from the runtime -- the provider payload (`disable_provider_payload`), the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks), and a failed Tool / Embedding / Retriever / LLM observation's `error_message` / `error_type` -- but not the dimensions the caller deliberately attaches (`correlation_id` / `session_id` / `userId` / trace name / caller metadata), which stay verbatim as cross-backend join keys. When any construction-determinable channel is live and OA establishes the client is bound to a provider it did not isolate, construction fails loud with a categorized `LangfuseProviderIsolationUnavailable` before any observation is emitted, rather than leaking payloads to a shared backend; where OA cannot establish the binding at all (a future SDK), it suppresses every channel and logs a warning. The failed-observation error message is gated per-emission (not knowable at construction): on an un-isolatable provider it is omitted, retaining only the error category where one exists (a Tool failure has no category, so it carries no message-derived status either). A single `accept_shared_provider=True` opt-out turns the whole thing into a warn-and-proceed onto the shared provider. With no channel live (the default privacy posture), an un-isolatable client neither raises nor warns. The existing caller-supplied path (mode a) is unchanged and never mutated: a caller who builds their own client stays responsible for isolating its `tracer_provider`, and OA documents the remedy rather than reaching into the supplied client. The `secret_key` is accepted as a `pydantic.SecretStr`, masked in OA's own reprs and logs with the plaintext read only at the SDK call (`public_key` and `host` stay plain strings), and a blank credential is rejected at the boundary rather than falling through to the SDK's ambient `LANGFUSE_*` environment fallback. A `sample_rate` passed for the client is applied to the isolated provider, since the SDK only honors it on a provider it builds itself. `accept_shared_provider` binds the provider the application already registered rather than letting the SDK construct and globally register one of its own, which would capture OTel's single-assignment global slot. The new `LangfuseProviderIsolationUnavailable` derives from an `ObservabilityError` base, a fourth hierarchy alongside the graph-engine, llm-provider, and checkpoint ones. Spec v0.108.0 / v0.110.0 / v0.111.0 are beyond the current v0.107.0 pin, so this ships ahead of the pin (unit-tested); the conformance fixtures (157 / 158, proposals 0115 / 0116 / 0117) ride the pin bump. The LLM error-message arm ships ahead of its spec formalization (proposal 0118, in progress at time of writing). ### Changed @@ -36,7 +36,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The ### Fixed -- **The Langfuse failure-isolation marker no longer carries the caught exception's message** (proposal 0118, observability §8.4). **Behavioral for the Langfuse mapping.** The `openarmature.failure_isolated` marker span wrote the caught exception's message into `observation.metadata.error_message`, but that span is a graph-mechanism marker no §8.4.x mapping table covers, so writing harvested exception content onto it was non-conforming over-emission: an exception message that can echo application data (PII, tool arguments, an upstream API error body) reached the Langfuse backend under every privacy setting, since no knob gated it. The marker now carries only `error_category` (plus the caller-supplied `failure_isolation_event_name` and the node name), matching the node span's treatment, so an isolated node failure and an ordinary node failure render the same. The full exception is unaffected on the OTel side, where the span still records it via `record_exception` on openarmature's private provider. A sweep of every bundled Langfuse handler against the same rule found no other unmapped harvested-content emission. +- **The Langfuse failure-isolation marker no longer carries the caught exception's message** (proposal 0118, observability §8.4). **Behavioral for the Langfuse mapping.** The `openarmature.failure_isolated` marker span wrote the caught exception's message into `observation.metadata.error_message`, but that span is a graph-mechanism marker no §8.4.x mapping table covers, so writing harvested exception content onto it was non-conforming over-emission: an exception message that can echo application data (PII, tool arguments, an upstream API error body) reached the Langfuse backend under every privacy setting, since no knob gated it. The marker now carries only `error_category` (plus the caller-supplied `failure_isolation_event_name` and the node name), matching the node span's treatment, so an isolated node failure and an ordinary node failure render the same. The full exception is unaffected on the OTel side, where the failure-isolation span still carries it as the `openarmature.failure_isolation.message` attribute. A sweep of every bundled Langfuse handler against the same rule found no other unmapped harvested-content emission. - **The OTel `openarmature.llm.complete` span records the exception event on a failed attempt** (observability §4.2). A failed provider-call span carried `ERROR` status but no exception event; the node and invocation spans already recorded it, the LLM span did not. It now records the OTel semconv exception event (`exception.type` / `exception.message`) on every failed attempt, matching the sibling spans. Surfaced while wiring the proposal 0082 error-span fixtures. ## [0.16.0] — 2026-07-18 diff --git a/docs/agent/non-obvious-shapes.md b/docs/agent/non-obvious-shapes.md index d94aa707..bc99488f 100644 --- a/docs/agent/non-obvious-shapes.md +++ b/docs/agent/non-obvious-shapes.md @@ -119,19 +119,20 @@ Different classes, same OTel-Logs export path. If both are attached against the The trace-side sibling of the `LoggerProvider` log-bridge gotcha above. When OA constructs the Langfuse client for you (`LangfuseObserver.from_credentials(...)` / `LangfuseSDKAdapter.from_credentials(...)`), it binds the client to a dedicated `TracerProvider` so OA's observations do not leak onto a provider shared with the application's OTel backend. But the Langfuse v4 SDK caches ONE client per `public_key` process-wide: if any client for that key was constructed first (the app called `Langfuse()`, used `langfuse.openai` / `@observe`, or an earlier OA call), the SDK returns the cached client and OA's dedicated provider is silently discarded. So isolation only holds when OA is the FIRST constructor for that credential. -Rather than leak silently, OA detects the binding and fails closed. When a harvested-payload channel is live and OA establishes the client landed on a provider it did not isolate, `from_credentials` raises `LangfuseProviderIsolationUnavailable` at construction, surfacing an init-ordering bug you would otherwise never see. The guarded channels are the provider payload (`disable_provider_payload`) and the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks). Harvested error content is emitted only on the four mapped provider observations (LLM, Embedding, Tool, Retriever), where a failed observation's `error_message` / `error_type` is omitted per-emission on a shared provider with the error category retained; a graph-mechanism span (a node span, a failure-isolation marker) carries only the error category, and its exception detail goes to OA's isolated OTel span via `record_exception`, never to Langfuse. Caller-attached dimensions (`correlation_id`, `session_id`, `userId`, trace name, your own metadata) are join keys by design and stay verbatim. +Rather than leak silently, OA detects the binding and fails closed. When a harvested-payload channel is live and OA establishes the client landed on a provider it did not isolate, `from_credentials` raises `LangfuseProviderIsolationUnavailable` at construction, surfacing an init-ordering bug you would otherwise never see. The guarded channels are the provider payload (`disable_provider_payload`) and the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks). Harvested error content is emitted only on the four mapped provider observations (LLM, Embedding, Tool, Retriever), where a failed observation's `error_message` / `error_type` are omitted per-emission whenever OA could not establish that the client is isolated; LLM, Embedding and Retriever keep their error category as the status message, while a Tool failure has no category at all and so renders `ERROR` with a null status message and no error rows. A graph-mechanism span (a node span, a failure-isolation marker) never carries the exception message on any provider; that detail rides the OTel span instead (`openarmature.failure_isolation.message`), never Langfuse. Caller-attached dimensions (`correlation_id`, `session_id`, `userId`, trace name, your own metadata) are join keys by design and stay verbatim. Remedies: construct OA's Langfuse client before any other client for that `public_key`; or pass `accept_shared_provider=True` to acknowledge a shared provider (OA warns and proceeds); or build your own client with an isolated `tracer_provider=` and pass it in via the caller-supplied path (`LangfuseObserver(client=LangfuseSDKAdapter(your_client))`), which OA never mutates. Under OA's default privacy posture (no payload channel live), an un-isolatable client is harmless and neither raises nor warns. -### Three exception hierarchies; know which one your code catches +### Four exception hierarchies; know which one your code catches -`openarmature` exceptions split across three sibling hierarchies: +`openarmature` exceptions split across four sibling hierarchies: - `RuntimeGraphError` (in `openarmature.graph`): node execution failures: `NodeException`, `RoutingError`, `EdgeException`, `ReducerError`, `StateValidationError`. Each has a `category` string matching the spec's canonical error categories. - `CheckpointError` (in `openarmature.checkpoint`): persistence failures: `CheckpointNotFound`, `CheckpointSaveFailed`, `CheckpointRecordInvalid`, `CheckpointStateMigrationMissing`, `CheckpointStateMigrationFailed`, `CheckpointStateMigrationChainAmbiguous`. - `LlmProviderError` (in `openarmature.llm`): provider call failures: `ProviderAuthentication`, `ProviderInvalidRequest`, `ProviderInvalidResponse`, `ProviderInvalidModel`, `ProviderModelNotLoaded`, `ProviderRateLimit`, `ProviderUnavailable`, `ProviderUnsupportedContentBlock`, `StructuredOutputInvalid`. +- `ObservabilityError` (in `openarmature.observability.langfuse`): observability backend wiring failures: `LangfuseProviderIsolationUnavailable`. Raised while constructing or driving a backend, not while running a graph or calling a provider. -Catching `Exception` works but is too broad; catching one hierarchy misses the other two. If you want to branch on category strings (e.g., for retry logic), catch the relevant base: `RuntimeGraphError` covers all five spec runtime categories, `LlmProviderError` covers all nine provider categories, `CheckpointError` covers all six checkpoint categories. The `TRANSIENT_CATEGORIES` frozenset in `openarmature.llm` enumerates which provider categories are retriable. +Catching `Exception` works but is too broad; catching one hierarchy misses the other three. If you want to branch on category strings (e.g., for retry logic), catch the relevant base: `RuntimeGraphError` covers all five spec runtime categories, `LlmProviderError` covers all nine provider categories, `CheckpointError` covers all six checkpoint categories. The `TRANSIENT_CATEGORIES` frozenset in `openarmature.llm` enumerates which provider categories are retriable. ### Filter `openarmature.*`-namespaced events when your observer only cares about user nodes diff --git a/docs/concepts/observability.md b/docs/concepts/observability.md index b1510f15..ea6eef8b 100644 --- a/docs/concepts/observability.md +++ b/docs/concepts/observability.md @@ -1122,8 +1122,32 @@ pip install 'openarmature[langfuse]' Production wire-up: +```python +from pydantic import SecretStr +from openarmature.observability.langfuse import LangfuseObserver + +observer = LangfuseObserver.from_credentials( + public_key="pk-lf-...", + secret_key=SecretStr("sk-lf-..."), + host="https://cloud.langfuse.com", + disable_provider_payload=False, +) +``` + +Prefer `from_credentials`: openarmature builds the Langfuse client on a +dedicated `TracerProvider`, so its observations do not also land on the +provider your application registered globally. A client you build +yourself binds the global provider unless you pass `tracer_provider=`, +which exports every observation, prompts and completions included, to +your application's tracing backend as well. + +If you must build the client yourself (to reuse an existing instance, +say), isolate it explicitly and hand it in; openarmature never mutates +a client you supply: + ```python from langfuse import Langfuse +from opentelemetry.sdk.trace import TracerProvider from openarmature.observability.langfuse import ( LangfuseObserver, LangfuseSDKAdapter, @@ -1133,6 +1157,7 @@ langfuse_client = Langfuse( public_key="pk-lf-...", secret_key="sk-lf-...", host="https://cloud.langfuse.com", + tracer_provider=TracerProvider(), # keep observations off the global provider ) observer = LangfuseObserver( client=LangfuseSDKAdapter(langfuse_client), diff --git a/docs/examples/langfuse-observability.md b/docs/examples/langfuse-observability.md index 00b31f0b..875e538b 100644 --- a/docs/examples/langfuse-observability.md +++ b/docs/examples/langfuse-observability.md @@ -127,23 +127,23 @@ Wrap the SDK client with `LangfuseSDKAdapter` and pass it to the observer: ```python -from langfuse import Langfuse -from openarmature.observability.langfuse import ( - LangfuseObserver, - LangfuseSDKAdapter, -) +from pydantic import SecretStr +from openarmature.observability.langfuse import LangfuseObserver -langfuse_client = Langfuse( +observer = LangfuseObserver.from_credentials( public_key="pk-lf-...", - secret_key="sk-lf-...", + secret_key=SecretStr("sk-lf-..."), host="https://cloud.langfuse.com", -) -observer = LangfuseObserver( - client=LangfuseSDKAdapter(langfuse_client), disable_provider_payload=False, ) ``` +openarmature builds the client on a dedicated `TracerProvider` here, so +its observations stay off the provider your application registered +globally. Building the client yourself binds that global provider +unless you pass `tracer_provider=`, which exports every observation, +prompts and completions included, to your app's tracing backend too. + The adapter bridges `langfuse>=4.6,<5`'s unified `start_observation` API onto OA's four-method `LangfuseClient` Protocol. v4 has no explicit trace creation (traces are auto-created from observations); diff --git a/docs/examples/production-observability.md b/docs/examples/production-observability.md index 60825dd7..7f29709f 100644 --- a/docs/examples/production-observability.md +++ b/docs/examples/production-observability.md @@ -27,7 +27,8 @@ part is the observability wiring: `OTLPSpanExporter` pointed at HyperDX / Honeycomb / Tempo / any OTLP backend). - `LangfuseObserver` attached with an `InMemoryLangfuseClient` - (production swaps for `LangfuseSDKAdapter(Langfuse(...))`). + (production swaps for `LangfuseObserver.from_credentials(...)`, which + isolates the client's `TracerProvider`). - Both observers consume the same `NodeEvent` stream independently; node code never knows there are two backends. - `LangfuseObserver` carries `trace_input_from_state` and diff --git a/examples/langfuse-observability/main.py b/examples/langfuse-observability/main.py index 98fdc75b..9e097daf 100644 --- a/examples/langfuse-observability/main.py +++ b/examples/langfuse-observability/main.py @@ -252,21 +252,22 @@ async def main() -> None: # The bundled in-memory client captures everything the observer # would have sent to Langfuse; Trace, Observations, Generation - # fields; without needing a Langfuse account. For production: + # fields; without needing a Langfuse account. For production, build + # the observer from credentials so OA owns the Langfuse client and + # keeps its observations on a dedicated TracerProvider: # - # from langfuse import Langfuse - # from openarmature.observability.langfuse import LangfuseSDKAdapter + # from pydantic import SecretStr # - # langfuse_client = Langfuse( + # observer = LangfuseObserver.from_credentials( # public_key="pk-lf-...", - # secret_key="sk-lf-...", + # secret_key=SecretStr("sk-lf-..."), # host="https://cloud.langfuse.com", # ) - # client = LangfuseSDKAdapter(langfuse_client) # - # Validated against ``langfuse>=4.6,<5``. The adapter bridges - # langfuse v4's unified ``start_observation`` API onto OA's - # ``LangfuseClient`` Protocol; the observer code doesn't change. + # A Langfuse client you construct yourself binds the globally + # registered TracerProvider unless you pass ``tracer_provider=``, + # which also exports every observation to your app's tracing + # backend. Validated against ``langfuse>=4.6,<5``. client = InMemoryLangfuseClient() # disable_provider_payload=False opts in to capturing the input messages diff --git a/examples/production-observability/main.py b/examples/production-observability/main.py index 6d775687..a30050c8 100644 --- a/examples/production-observability/main.py +++ b/examples/production-observability/main.py @@ -38,9 +38,13 @@ in-process so the demo can print it at the end without needing a real Langfuse account. ``InMemorySpanExporter`` does the symmetric job for OTel. Production code swaps in - ``LangfuseSDKAdapter(Langfuse(...))`` and - ``BatchSpanProcessor(OTLPSpanExporter(...))`` respectively; the - observer call surface doesn't change. + ``LangfuseObserver.from_credentials(public_key=..., secret_key=...)`` + and ``BatchSpanProcessor(OTLPSpanExporter(...))`` respectively; the + observer call surface doesn't change. Prefer ``from_credentials`` + over building the Langfuse client yourself: OA then binds the client + to a dedicated ``TracerProvider``, whereas a client constructed + without ``tracer_provider=`` binds the globally registered one and + exports every observation to your app's tracing backend too. - **Queryable accumulator observer + per-invocation drain.** A third observer (``LlmUsageAccumulator``) rolls up LLM token totals per invocation, including a cache-hit ratio from diff --git a/src/openarmature/AGENTS.md b/src/openarmature/AGENTS.md index cbda2dd5..494b873f 100644 --- a/src/openarmature/AGENTS.md +++ b/src/openarmature/AGENTS.md @@ -1601,19 +1601,20 @@ Different classes, same OTel-Logs export path. If both are attached against the The trace-side sibling of the `LoggerProvider` log-bridge gotcha above. When OA constructs the Langfuse client for you (`LangfuseObserver.from_credentials(...)` / `LangfuseSDKAdapter.from_credentials(...)`), it binds the client to a dedicated `TracerProvider` so OA's observations do not leak onto a provider shared with the application's OTel backend. But the Langfuse v4 SDK caches ONE client per `public_key` process-wide: if any client for that key was constructed first (the app called `Langfuse()`, used `langfuse.openai` / `@observe`, or an earlier OA call), the SDK returns the cached client and OA's dedicated provider is silently discarded. So isolation only holds when OA is the FIRST constructor for that credential. -Rather than leak silently, OA detects the binding and fails closed. When a harvested-payload channel is live and OA establishes the client landed on a provider it did not isolate, `from_credentials` raises `LangfuseProviderIsolationUnavailable` at construction, surfacing an init-ordering bug you would otherwise never see. The guarded channels are the provider payload (`disable_provider_payload`) and the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks). Harvested error content is emitted only on the four mapped provider observations (LLM, Embedding, Tool, Retriever), where a failed observation's `error_message` / `error_type` is omitted per-emission on a shared provider with the error category retained; a graph-mechanism span (a node span, a failure-isolation marker) carries only the error category, and its exception detail goes to OA's isolated OTel span via `record_exception`, never to Langfuse. Caller-attached dimensions (`correlation_id`, `session_id`, `userId`, trace name, your own metadata) are join keys by design and stay verbatim. +Rather than leak silently, OA detects the binding and fails closed. When a harvested-payload channel is live and OA establishes the client landed on a provider it did not isolate, `from_credentials` raises `LangfuseProviderIsolationUnavailable` at construction, surfacing an init-ordering bug you would otherwise never see. The guarded channels are the provider payload (`disable_provider_payload`) and the Trace-level state input/output (`disable_state_payload` and the `trace_input_from_state` / `trace_output_from_state` hooks). Harvested error content is emitted only on the four mapped provider observations (LLM, Embedding, Tool, Retriever), where a failed observation's `error_message` / `error_type` are omitted per-emission whenever OA could not establish that the client is isolated; LLM, Embedding and Retriever keep their error category as the status message, while a Tool failure has no category at all and so renders `ERROR` with a null status message and no error rows. A graph-mechanism span (a node span, a failure-isolation marker) never carries the exception message on any provider; that detail rides the OTel span instead (`openarmature.failure_isolation.message`), never Langfuse. Caller-attached dimensions (`correlation_id`, `session_id`, `userId`, trace name, your own metadata) are join keys by design and stay verbatim. Remedies: construct OA's Langfuse client before any other client for that `public_key`; or pass `accept_shared_provider=True` to acknowledge a shared provider (OA warns and proceeds); or build your own client with an isolated `tracer_provider=` and pass it in via the caller-supplied path (`LangfuseObserver(client=LangfuseSDKAdapter(your_client))`), which OA never mutates. Under OA's default privacy posture (no payload channel live), an un-isolatable client is harmless and neither raises nor warns. -### Three exception hierarchies; know which one your code catches +### Four exception hierarchies; know which one your code catches -`openarmature` exceptions split across three sibling hierarchies: +`openarmature` exceptions split across four sibling hierarchies: - `RuntimeGraphError` (in `openarmature.graph`): node execution failures: `NodeException`, `RoutingError`, `EdgeException`, `ReducerError`, `StateValidationError`. Each has a `category` string matching the spec's canonical error categories. - `CheckpointError` (in `openarmature.checkpoint`): persistence failures: `CheckpointNotFound`, `CheckpointSaveFailed`, `CheckpointRecordInvalid`, `CheckpointStateMigrationMissing`, `CheckpointStateMigrationFailed`, `CheckpointStateMigrationChainAmbiguous`. - `LlmProviderError` (in `openarmature.llm`): provider call failures: `ProviderAuthentication`, `ProviderInvalidRequest`, `ProviderInvalidResponse`, `ProviderInvalidModel`, `ProviderModelNotLoaded`, `ProviderRateLimit`, `ProviderUnavailable`, `ProviderUnsupportedContentBlock`, `StructuredOutputInvalid`. +- `ObservabilityError` (in `openarmature.observability.langfuse`): observability backend wiring failures: `LangfuseProviderIsolationUnavailable`. Raised while constructing or driving a backend, not while running a graph or calling a provider. -Catching `Exception` works but is too broad; catching one hierarchy misses the other two. If you want to branch on category strings (e.g., for retry logic), catch the relevant base: `RuntimeGraphError` covers all five spec runtime categories, `LlmProviderError` covers all nine provider categories, `CheckpointError` covers all six checkpoint categories. The `TRANSIENT_CATEGORIES` frozenset in `openarmature.llm` enumerates which provider categories are retriable. +Catching `Exception` works but is too broad; catching one hierarchy misses the other three. If you want to branch on category strings (e.g., for retry logic), catch the relevant base: `RuntimeGraphError` covers all five spec runtime categories, `LlmProviderError` covers all nine provider categories, `CheckpointError` covers all six checkpoint categories. The `TRANSIENT_CATEGORIES` frozenset in `openarmature.llm` enumerates which provider categories are retriable. ### Filter `openarmature.*`-namespaced events when your observer only cares about user nodes diff --git a/src/openarmature/observability/langfuse/adapter.py b/src/openarmature/observability/langfuse/adapter.py index d0a034bd..ce334531 100644 --- a/src/openarmature/observability/langfuse/adapter.py +++ b/src/openarmature/observability/langfuse/adapter.py @@ -34,7 +34,10 @@ from __future__ import annotations import json +import logging +import os import threading +from collections.abc import Mapping from contextlib import ExitStack from datetime import datetime from typing import Any, cast @@ -89,23 +92,81 @@ def _stringify_metadata(metadata: dict[str, Any] | None) -> dict[str, str]: _ISOLATED_PROVIDERS: dict[str, Any] = {} _ISOLATED_PROVIDERS_LOCK = threading.Lock() +_logger = logging.getLogger("openarmature.observability") -def _reuse_isolated_provider(public_key: str) -> Any: + +def _resolve_sample_rate(langfuse_kwargs: Mapping[str, Any]) -> float: + """Resolve the sampling ratio the caller asked the Langfuse client for. + + The SDK applies its ``sample_rate`` only while building a provider of its + own, which the isolated path bypasses, so the ratio has to be put on OA's + provider instead or sampling silently does nothing. + """ + raw = langfuse_kwargs.get("sample_rate") + if raw is None: + raw = os.environ.get("LANGFUSE_SAMPLE_RATE") + if raw is None: + return 1.0 + try: + rate = float(raw) + except (TypeError, ValueError): + return 1.0 + return min(max(rate, 0.0), 1.0) + + +def _reuse_isolated_provider(public_key: str, sample_rate: float = 1.0) -> Any: """Return the one isolated ``TracerProvider`` OA uses for ``public_key``, - building it on first use. A bare provider is the whole isolation: the SDK - adds its own span processor to whatever provider it is handed and does not - register a handed-in provider globally.""" + building it on first use. A dedicated provider is the whole isolation: the + SDK adds its own span processor to whatever provider it is handed and does + not register a handed-in provider globally. + + The provider is reused per credential, so the first call's sampling ratio + governs -- matching the SDK, which likewise caches one client per key and + ignores a later call's configuration. + """ from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.sampling import TraceIdRatioBased with _ISOLATED_PROVIDERS_LOCK: provider = _ISOLATED_PROVIDERS.get(public_key) if provider is None: - provider = TracerProvider() + if sample_rate >= 1.0: + provider = TracerProvider() + else: + provider = TracerProvider(sampler=TraceIdRatioBased(sample_rate)) _ISOLATED_PROVIDERS[public_key] = provider return provider -def _classify_isolation(client: Any, provider: Any, accept_shared_provider: bool) -> str: +def _resolve_shared_provider() -> Any: + """Resolve the ambient provider for the accept-a-shared-provider opt-out, or + ``None`` when there is nothing registered to share. + + Handing the SDK ``None`` is not the same as sharing: with no global provider + registered yet the SDK builds one and registers it globally, and OTel allows + that only once, so OA's opt-out would capture the process-global slot and the + application's own later registration would be silently ignored. Resolving the + provider here keeps the opt-out to what it says: bind to what is already there. + """ + from opentelemetry import trace as otel_trace + from opentelemetry.trace import ProxyTracerProvider + + provider = otel_trace.get_tracer_provider() + return None if isinstance(provider, ProxyTracerProvider) else provider + + +def _public_key_is_cached(public_key: str) -> bool: + """Whether the SDK already holds a client for this credential, so the + configuration of the call about to be made will be discarded.""" + try: + from langfuse._client.resource_manager import LangfuseResourceManager + except ImportError: # pragma: no cover - SDK internal moved + return False + instances = getattr(LangfuseResourceManager, "_instances", None) + return bool(instances) and public_key in instances + + +def _classify_isolation(client: Any, public_key: str, accept_shared_provider: bool) -> str: """Classify what the SDK actually bound the client's observations to. The SDK caches one resource manager per public_key, so a handed-in provider @@ -114,22 +175,26 @@ def _classify_isolation(client: Any, provider: Any, accept_shared_provider: bool is discarded. """ # Proposal 0116 payload-leak invariant: OA establishes the binding to pick the - # raise / suppress arm. - if accept_shared_provider: - return ISOLATION_SHARED_ACCEPTED + # raise / suppress arm. A client with tracing disabled exports nothing at all, + # so it is checked first -- there is no binding to establish and no leak to + # guard, whatever provider the cached manager happens to hold. + if not getattr(client, "_tracing_enabled", True): + return ISOLATION_ISOLATED # Guarded read of the SDK-internal binding: a future SDK that stops exposing # it leaves us unable to establish isolation, so the observer takes the # portable suppress floor rather than a false all-clear. resources = getattr(client, "_resources", None) bound = getattr(resources, "tracer_provider", None) if resources is not None else None if bound is None: - # A client with tracing disabled legitimately has no provider and exports - # nothing, so there is no leak to guard -- distinct from a future SDK that - # hides the binding (the genuine undetectable case). - if not getattr(client, "_tracing_enabled", True): - return ISOLATION_ISOLATED return ISOLATION_UNDETECTABLE - return ISOLATION_ISOLATED if bound is provider else ISOLATION_LEAKED + # Membership, not identity: a client the SDK resolved onto the provider OA + # established for this credential satisfies the invariant even when the + # caller opted into sharing, so the opt-out is read only if it is needed. + with _ISOLATED_PROVIDERS_LOCK: + established = _ISOLATED_PROVIDERS.get(public_key) + if bound is established: + return ISOLATION_ISOLATED + return ISOLATION_SHARED_ACCEPTED if accept_shared_provider else ISOLATION_LEAKED class _SpanHandle: @@ -307,7 +372,39 @@ def from_credentials( "tracer_provider is managed by from_credentials " "(via `accept_shared_provider`); do not pass it through langfuse_kwargs" ) - provider = None if accept_shared_provider else _reuse_isolated_provider(public_key) + # Validate at the boundary: the SDK falls back to LANGFUSE_* environment + # credentials for a blank value, so an empty config field would silently + # authenticate against whatever ambient project is configured instead of + # failing. Never echo the value itself. + if not public_key or not public_key.strip(): + raise ValueError("public_key is required and must not be blank") + if not secret_key.get_secret_value().strip(): + raise ValueError("secret_key is required and must not be blank") + + sample_rate = _resolve_sample_rate(langfuse_kwargs) + provider = None + if accept_shared_provider: + provider = _resolve_shared_provider() + if provider is None: + # Nothing is registered to share, and passing the SDK None would + # let it capture the process-global slot. Isolate instead. + _logger.warning( + "accept_shared_provider=True but no TracerProvider is registered yet; " + "using an isolated provider rather than letting the Langfuse SDK claim " + "the process-global one" + ) + if provider is None: + provider = _reuse_isolated_provider(public_key, sample_rate) + + # The SDK caches one client per credential, so a second construction for + # the same key returns the first client and drops this call's secret_key / + # host / other options. Say so rather than let the discard pass silently. + if _public_key_is_cached(public_key): + _logger.warning( + "a Langfuse client already exists for this public_key; the SDK returns the " + "cached client, so this call's host and client options are not applied" + ) + client = Langfuse( public_key=public_key, secret_key=secret_key.get_secret_value(), @@ -316,7 +413,7 @@ def from_credentials( **langfuse_kwargs, ) adapter = cls(client) - adapter._isolation_status = _classify_isolation(client, provider, accept_shared_provider) + adapter._isolation_status = _classify_isolation(client, public_key, accept_shared_provider) return adapter def trace( diff --git a/src/openarmature/observability/langfuse/errors.py b/src/openarmature/observability/langfuse/errors.py index 57b46cdb..5b2b9b47 100644 --- a/src/openarmature/observability/langfuse/errors.py +++ b/src/openarmature/observability/langfuse/errors.py @@ -12,7 +12,16 @@ from __future__ import annotations -class LangfuseProviderIsolationUnavailable(Exception): +class ObservabilityError(Exception): + """Base for errors raised by the bundled observability backends. + + Distinct from the graph-engine, llm-provider, and prompt-management + hierarchies: these are raised while wiring or driving an observability + backend, not while running a graph or calling a provider. + """ + + +class LangfuseProviderIsolationUnavailable(ObservabilityError): """OA cannot keep its payload-bearing Langfuse observations off a TracerProvider shared with the application, and the caller has not accepted a shared provider, so construction fails loud rather than leaking. diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index a5231a55..a7c5dbb6 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -450,6 +450,82 @@ def __post_init__(self) -> None: f"payload_byte_cap={self.payload_byte_cap} below the spec §5.5.5 " f"minimum of {_PAYLOAD_MIN_BYTES} bytes" ) + self._apply_isolation_policy() + + def _apply_isolation_policy(self) -> None: + # Proposal 0116 / 0117 payload-leak arms. They live here rather than in + # from_credentials so every observer over an OA-constructed client is + # guarded, including one built by handing a from_credentials adapter to + # the plain constructor. A caller-supplied client (mode a) records no + # status, so this is a no-op there and the caller keeps their provider. + status = getattr(self.client, "_isolation_status", None) + if status is None: + return + if status == ISOLATION_LEAKED and self._construction_channels_live(): + from openarmature.observability.langfuse.errors import ( + LangfuseProviderIsolationUnavailable, + ) + + raise LangfuseProviderIsolationUnavailable( + "OA constructed a Langfuse client whose observations would reach a " + "TracerProvider shared with the application (the SDK caches one client " + "per public_key, and one was already bound to a provider OA did not " + "isolate), and a payload channel is live so payloads would leak. Pass " + "accept_shared_provider=True to proceed onto the shared provider, or " + "construct OA's Langfuse client before any other client for this key." + ) + if status == ISOLATION_UNDETECTABLE: + # Portable floor: the binding cannot be established, so close every + # construction-time channel. Warned unconditionally -- harvested error + # content is suppressed at emission whether or not a payload channel + # is live, so silence would leave that invisible. + _logger.warning( + "cannot establish the Langfuse client's TracerProvider binding; " + "suppressing all provider and state payloads, and omitting failed " + "observations' error messages, to avoid a possible leak to a shared provider" + ) + self.disable_provider_payload = True + self.disable_state_payload = True + self.trace_input_from_state = None + self.trace_output_from_state = None + elif status == ISOLATION_LEAKED: + # No construction-time channel is live, so nothing is refused; the + # error-message channel is still suppressed at emission, which the + # operator would otherwise have no way to notice. + _logger.info( + "OA's Langfuse client is bound to a TracerProvider it did not isolate; " + "failed observations' error messages are omitted to avoid a leak" + ) + elif status == ISOLATION_SHARED_ACCEPTED: + # A provider-binding decision, not a payload one, so it is reported + # whatever the payload knobs say. + _logger.warning( + "accept_shared_provider=True: OA's Langfuse observations may reach a " + "TracerProvider shared with the application (acknowledged)" + ) + + def _construction_channels_live(self) -> bool: + # Any payload channel knowable at construction (0117): the provider + # payload, the Trace state payload, or a supplied state hook, which emits + # regardless of the state knob. + return ( + not self.disable_provider_payload + or not self.disable_state_payload + or self.trace_input_from_state is not None + or self.trace_output_from_state is not None + ) + + def _isolation_blocks_payload(self) -> bool: + # Emission-time half of the invariant. The knobs are public fields on a + # mutable dataclass, so a channel re-opened after construction would + # otherwise escape the arms above; every payload site consults this. + return getattr(self.client, "_isolation_status", None) in ( + ISOLATION_LEAKED, + ISOLATION_UNDETECTABLE, + ) + + def _emits_provider_payload(self) -> bool: + return not self.disable_provider_payload and not self._isolation_blocks_payload() @classmethod def from_credentials( @@ -485,19 +561,6 @@ def from_credentials( ordinary observer fields (``disable_provider_payload`` / ``disable_llm_spans`` / ...). """ - # Any construction-determinable payload channel makes a shared provider a - # leak (0117): the provider payload, the Trace-level state payload, or a - # supplied trace_input/output-from-state hook (a supplied hook emits - # regardless of the knob, so treat *supplied* as potentially-live). The - # failed-observation error-message channel is not construction-knowable, - # so it does not gate here; it is suppressed per-emission via the status. - emits_payloads = ( - not bool(observer_kwargs.get("disable_provider_payload", True)) - or not bool(observer_kwargs.get("disable_state_payload", True)) - or observer_kwargs.get("trace_input_from_state") is not None - or observer_kwargs.get("trace_output_from_state") is not None - ) - from openarmature.observability.langfuse.adapter import LangfuseSDKAdapter client = LangfuseSDKAdapter.from_credentials( @@ -507,42 +570,9 @@ def from_credentials( accept_shared_provider=accept_shared_provider, **(langfuse_kwargs or {}), ) - - if emits_payloads: - status = client._isolation_status - if status == ISOLATION_LEAKED: - from openarmature.observability.langfuse.errors import ( - LangfuseProviderIsolationUnavailable, - ) - - raise LangfuseProviderIsolationUnavailable( - "OA constructed a Langfuse client whose observations would reach a " - "TracerProvider shared with the application (the SDK caches one client " - "per public_key, and one was already bound to a provider OA did not " - "isolate), and a payload channel is live so payloads would leak. Pass " - "accept_shared_provider=True to proceed onto the shared provider, or " - "construct OA's Langfuse client before any other client for this key." - ) - if status == ISOLATION_UNDETECTABLE: - # Portable floor: cannot establish the binding, so suppress EVERY - # construction-time channel (fail-safe) and log a warning. The - # error-message channel is suppressed per-emission via the status. - _logger.warning( - "cannot establish the Langfuse client's TracerProvider binding; " - "suppressing all provider and state payloads to avoid a possible " - "leak to a shared provider" - ) - observer_kwargs["disable_provider_payload"] = True - observer_kwargs["disable_state_payload"] = True - observer_kwargs["trace_input_from_state"] = None - observer_kwargs["trace_output_from_state"] = None - elif status == ISOLATION_SHARED_ACCEPTED: - _logger.warning( - "accept_shared_provider=True with payload channels enabled: OA's " - "Langfuse observations may reach a TracerProvider shared with the " - "application (acknowledged)" - ) - + # The raise / suppress / warn arms run in __post_init__ off the client's + # recorded isolation status, so they apply to this path and equally to an + # observer built by handing a from_credentials adapter to the constructor. return cls(client=client, **observer_kwargs) def _omit_harvested_error(self) -> bool: @@ -550,10 +580,7 @@ def _omit_harvested_error(self) -> bool: # non-detectable suppress floor -- a failed observation's harvested # error_message / error_type must not reach the shared provider. Isolated, # opted-in, and caller-supplied (mode a) clients emit normally. - return getattr(self.client, "_isolation_status", None) in ( - ISOLATION_LEAKED, - ISOLATION_UNDETECTABLE, - ) + return self._isolation_blocks_payload() async def __call__( self, @@ -980,6 +1007,14 @@ def _handle_invocation_completed(self, event: InvocationCompletedEvent) -> None: self.client.update_trace(id=event.invocation_id, output=output_value) def _resolve_trace_input(self, event: InvocationStartedEvent) -> Any: + # Isolation floor: on a provider OA could not establish as isolated, the + # state channel is closed whatever the levers say, so a knob or hook set + # after construction cannot reopen it. + if self._isolation_blocks_payload(): + stub: dict[str, Any] = {"entry_node": event.entry_node} + if event.correlation_id is not None: + stub["correlation_id"] = event.correlation_id + return stub # Lever 1: caller hook. if self.trace_input_from_state is not None: try: @@ -1002,6 +1037,9 @@ def _resolve_trace_input(self, event: InvocationStartedEvent) -> Any: return stub def _resolve_trace_output(self, event: InvocationCompletedEvent) -> Any: + # Isolation floor, as for the input side. + if self._isolation_blocks_payload(): + return {"final_node": event.final_node, "status": event.status} # Lever 1: caller hook. if self.trace_output_from_state is not None: try: @@ -1915,7 +1953,7 @@ def _handle_typed_llm_completion(self, event: LlmCompletionEvent) -> None: model_parameters: dict[str, Any] = dict(event.request_params or {}) input_value: Any = None output_value: Any = None - if not self.disable_provider_payload: + if self._emits_provider_payload(): if event.input_messages: input_value = self._maybe_truncate_for_input(event.input_messages) if event.output_content is not None: @@ -2000,12 +2038,17 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: if not self._omit_harvested_error(): if event.error_type is not None: metadata["error_type"] = event.error_type - metadata["error_message"] = event.error_message + # A structured_output_invalid message quotes the model's own failing + # output, so for that category the message is response-derived payload + # and follows the payload knob as well as the isolation gate. Other + # categories describe the call, not the response. + if not (event.error_category == "structured_output_invalid" and self.disable_provider_payload): + metadata["error_message"] = event.error_message model_parameters: dict[str, Any] = dict(event.request_params or {}) input_value: Any = None output_value: Any = None end_kwargs: dict[str, Any] = {} - if not self.disable_provider_payload: + if self._emits_provider_payload(): if event.input_messages: input_value = self._maybe_truncate_for_input(event.input_messages) if event.request_extras: @@ -2016,7 +2059,7 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: # level + category. metadata.finish_reason is added for this category by # _typed_event_metadata. if event.error_category == "structured_output_invalid": - if not self.disable_provider_payload and event.output_content is not None: + if self._emits_provider_payload() and event.output_content is not None: output_value = self._maybe_truncate_for_output(event.output_content) usage = self._usage_from_typed_event(event) if usage is not None: @@ -2083,7 +2126,7 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: metadata["openarmature_tool_call_id"] = event.tool_call_id input_value: Any = None output_value: Any = None - if not self.disable_provider_payload: + if self._emits_provider_payload(): if event.arguments is not None: input_value = self._maybe_truncate_for_input(event.arguments) if isinstance(event, ToolCallEvent): @@ -2175,7 +2218,7 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non if event.caller_invocation_metadata is not None: _apply_caller_metadata(metadata, event.caller_invocation_metadata) input_value: Any = None - if not self.disable_provider_payload and event.input_strings: + if self._emits_provider_payload() and event.input_strings: input_value = self._maybe_truncate_for_input(event.input_strings) if isinstance(event, EmbeddingEvent): metadata["openarmature_input_count"] = event.input_count @@ -2184,7 +2227,7 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non if event.response_id is not None: metadata["openarmature_response_id"] = event.response_id output_value: Any = None - if not self.disable_provider_payload and event.output_vectors: + if self._emits_provider_payload() and event.output_vectors: output_value = self._maybe_truncate_for_input(event.output_vectors) usage = LangfuseUsage(input=event.usage.input_tokens) if event.usage is not None else None target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) @@ -2290,7 +2333,7 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: if event.top_k is not None: metadata["openarmature_top_k"] = event.top_k input_value: Any = None - if not self.disable_provider_payload: + if self._emits_provider_payload(): input_value = self._maybe_truncate_for_input( {"query": event.query, "documents": list(event.documents)} ) @@ -2300,7 +2343,7 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: if event.response_id is not None: metadata["openarmature_response_id"] = event.response_id output_value: Any = None - if not self.disable_provider_payload and event.output_results: + if self._emits_provider_payload() and event.output_results: output_value = self._maybe_truncate_for_input( [result.model_dump(exclude_none=True) for result in event.output_results] ) diff --git a/tests/unit/test_langfuse_payload_leak_canary.py b/tests/unit/test_langfuse_payload_leak_canary.py new file mode 100644 index 00000000..65ff72df --- /dev/null +++ b/tests/unit/test_langfuse_payload_leak_canary.py @@ -0,0 +1,230 @@ +"""Property test: no harvested content reaches an un-isolated Langfuse provider.""" + +# Spec basis: observability §6 / §8.4.x (proposals 0114 / 0116 / 0117 / 0118, the +# payload-leak invariant). +# +# This asserts the INVARIANT rather than the individual emission sites. Every +# harvested input carries a unique sentinel; the whole captured Langfuse tree is +# then searched for it. A site-by-site test can only cover the channels someone +# already thought of, so a newly added emission (or one the enumeration missed) +# slips through silently; this fails the moment any harvested value reaches a +# provider OA could not establish as isolated, whatever site emitted it. +# +# The isolated case asserts the sentinel IS present per channel, so the test +# cannot pass by driving nothing. + +import json +from typing import Any +from unittest.mock import MagicMock + +import pytest + +pytest.importorskip("langfuse") + +from openarmature.graph.events import ( # noqa: E402 + CaughtException, + FailureIsolatedEvent, + InvocationCompletedEvent, + InvocationStartedEvent, + LlmFailedEvent, + ToolCallFailedEvent, +) +from openarmature.observability.correlation import ( # noqa: E402 + _reset_invocation_id, + _set_invocation_id, +) +from openarmature.observability.langfuse import ( # noqa: E402 + InMemoryLangfuseClient, + LangfuseObserver, +) +from openarmature.observability.langfuse.client import ( # noqa: E402 + ISOLATION_ISOLATED, + ISOLATION_LEAKED, + ISOLATION_UNDETECTABLE, +) + +# One sentinel per harvested channel, so a leak names which channel leaked. +STATE_IN = "CANARY-state-input-8f21" +STATE_OUT = "CANARY-state-output-3b77" +LLM_MSG = "CANARY-llm-error-5c04" +TOOL_ARG = "CANARY-tool-argument-9d13" +TOOL_MSG = "CANARY-tool-error-1a56" +MARKER_MSG = "CANARY-isolated-exception-7e92" + +ALL_SENTINELS = (STATE_IN, STATE_OUT, LLM_MSG, TOOL_ARG, TOOL_MSG, MARKER_MSG) + +_INV = "inv-canary" + + +async def _drive_every_channel(observer: LangfuseObserver) -> None: + """Feed one event per harvested channel through the observer.""" + token = _set_invocation_id(_INV) + try: + await observer( + InvocationStartedEvent( + initial_state={"payload": STATE_IN}, + invocation_id=_INV, + correlation_id=None, + entry_node="start", + ) + ) + await observer( + LlmFailedEvent( + invocation_id=_INV, + correlation_id=None, + node_name="call_llm", + namespace=("call_llm",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="openai", + model="test-model", + latency_ms=1.0, + input_messages=[], + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-llm", + error_category="provider_unavailable", + error_message=LLM_MSG, + ) + ) + await observer( + ToolCallFailedEvent( + invocation_id=_INV, + correlation_id=None, + node_name="run_tool", + namespace=("run_tool",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + call_id="cc-tool", + tool_name="lookup", + tool_call_id="call_1", + arguments={"query": TOOL_ARG}, + latency_ms=1.0, + error_type="ValueError", + error_message=TOOL_MSG, + ) + ) + await observer( + FailureIsolatedEvent( + event_name="node_failed", + namespace=("run_tool",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + pre_state={}, + post_state={}, + caught_exception=CaughtException(category="node_exception", message=MARKER_MSG, chain=()), + ) + ) + await observer( + InvocationCompletedEvent( + final_state={"payload": STATE_OUT}, + status="completed", + final_node="end", + invocation_id=_INV, + correlation_id=None, + ) + ) + finally: + _reset_invocation_id(token) + + +def _captured_text(client: InMemoryLangfuseClient) -> str: + """Everything the client recorded, flattened, so the search covers any field + a future emission might use rather than the ones enumerated today.""" + traces: list[Any] = [] + for trace in client.traces.values(): + traces.append( + { + "trace": {k: v for k, v in vars(trace).items() if k != "observations"}, + "observations": [vars(obs) for obs in trace.observations], + } + ) + return json.dumps(traces, default=str) + + +def _observer_on(status: str | None) -> tuple[LangfuseObserver, InMemoryLangfuseClient]: + client = InMemoryLangfuseClient() + if status is not None: + client._isolation_status = status # type: ignore[attr-defined] + # Every payload channel opened: the strongest configuration a caller can ask + # for, so the isolation floor is what has to hold the line. + observer = LangfuseObserver( + client=client, + disable_provider_payload=False, + disable_state_payload=False, + ) + return observer, client + + +async def test_no_harvested_content_reaches_an_undetectable_provider() -> None: + # UNDETECTABLE is the arm where an observer still exists with channels + # requested: the binding cannot be established, so construction closes every + # channel instead of refusing. Nothing harvested may reach the client. + observer, client = _observer_on(ISOLATION_UNDETECTABLE) + await _drive_every_channel(observer) + captured = _captured_text(client) + leaked = [s for s in ALL_SENTINELS if s in captured] + assert not leaked, f"harvested content reached an un-isolated provider: {leaked}" + + +def test_a_leaked_provider_refuses_construction_rather_than_emitting() -> None: + # On LEAKED the guarantee is stronger than suppression: with any channel live + # the observer never comes into existence, so there is nothing to emit from. + from openarmature.observability.langfuse import LangfuseProviderIsolationUnavailable + + with pytest.raises(LangfuseProviderIsolationUnavailable): + _observer_on(ISOLATION_LEAKED) + + +async def test_marker_exception_never_reaches_langfuse_even_when_isolated() -> None: + # The failure-isolation marker is a graph-mechanism span no mapping table + # covers, so its exception message is prohibited outright rather than gated. + observer, client = _observer_on(ISOLATION_ISOLATED) + await _drive_every_channel(observer) + assert MARKER_MSG not in _captured_text(client) + + +async def test_channels_are_actually_exercised_when_isolated() -> None: + # Non-vacuity: without this, a workload that silently stopped driving a + # channel would make the leak assertions above pass for the wrong reason. + observer, client = _observer_on(ISOLATION_ISOLATED) + await _drive_every_channel(observer) + captured = _captured_text(client) + missing = [s for s in (STATE_IN, STATE_OUT, LLM_MSG, TOOL_ARG, TOOL_MSG) if s not in captured] + assert not missing, f"channel not exercised, so the leak test proves nothing: {missing}" + + +@pytest.mark.parametrize("status", [ISOLATION_LEAKED, ISOLATION_UNDETECTABLE]) +async def test_reopening_channels_after_construction_still_leaks_nothing(status: str) -> None: + # The knobs are public fields on a mutable dataclass; enforcement is at + # emission, so re-opening a channel afterwards must not reopen the leak. + client = InMemoryLangfuseClient() + client._isolation_status = status # type: ignore[attr-defined] + observer = LangfuseObserver(client=client) + observer.disable_provider_payload = False + observer.disable_state_payload = False + observer.trace_input_from_state = lambda state: state + observer.trace_output_from_state = lambda state: state + await _drive_every_channel(observer) + captured = _captured_text(client) + leaked = [s for s in ALL_SENTINELS if s in captured] + assert not leaked, f"a channel re-opened after construction leaked: {leaked}" + + +def test_adapter_built_client_is_guarded_on_every_construction_path() -> None: + # The bypass class: any observer over an OA-constructed client must be + # guarded, not only one built through the observer's own factory. + from openarmature.observability.langfuse import LangfuseProviderIsolationUnavailable + + leaked_client = MagicMock(_isolation_status=ISOLATION_LEAKED) + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver(client=leaked_client, disable_provider_payload=False) + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver(client=leaked_client, disable_state_payload=False) + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver(client=leaked_client, trace_input_from_state=lambda s: s) diff --git a/tests/unit/test_langfuse_provider_isolation.py b/tests/unit/test_langfuse_provider_isolation.py index ef174358..b3fd00c1 100644 --- a/tests/unit/test_langfuse_provider_isolation.py +++ b/tests/unit/test_langfuse_provider_isolation.py @@ -92,15 +92,85 @@ def test_binding_not_exposed_status_undetectable() -> None: assert adapter._isolation_status == ISOLATION_UNDETECTABLE -def test_accept_shared_provider_binds_ambient_status_shared_accepted() -> None: - with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_other) as mock_lf: - adapter = LangfuseSDKAdapter.from_credentials( - public_key="pk", secret_key=SecretStr("sk"), accept_shared_provider=True - ) - assert mock_lf.call_args.kwargs["tracer_provider"] is None +def test_accept_shared_provider_binds_the_ambient_provider() -> None: + # Opt-out with a real ambient provider: OA binds THAT provider (never None, + # which would let the SDK claim the process-global slot). + from opentelemetry.sdk.trace import TracerProvider + + ambient = TracerProvider() + + def bind_ambient(**kwargs: Any) -> MagicMock: + client = MagicMock() + client._resources.tracer_provider = kwargs["tracer_provider"] + return client + + with patch(f"{_ADAPTER}._resolve_shared_provider", return_value=ambient): + with patch(f"{_ADAPTER}.Langfuse", side_effect=bind_ambient) as mock_lf: + adapter = LangfuseSDKAdapter.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), accept_shared_provider=True + ) + assert mock_lf.call_args.kwargs["tracer_provider"] is ambient assert adapter._isolation_status == ISOLATION_SHARED_ACCEPTED +def test_accept_shared_provider_with_no_ambient_provider_isolates_instead(caplog: Any) -> None: + # Nothing registered to share: handing the SDK None would make it construct + # and globally register its own provider, capturing the one-shot global slot, + # so OA isolates instead and says so. + with patch(f"{_ADAPTER}._resolve_shared_provider", return_value=None): + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_passed) as mock_lf: + with caplog.at_level("WARNING", logger="openarmature.observability"): + adapter = LangfuseSDKAdapter.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), accept_shared_provider=True + ) + assert mock_lf.call_args.kwargs["tracer_provider"] is not None + assert adapter._isolation_status == ISOLATION_ISOLATED + assert "no TracerProvider is registered yet" in caplog.text + + +def test_opted_in_but_actually_isolated_is_classified_isolated() -> None: + # Membership, not the flag: when the SDK resolved the client onto OA's own + # isolated provider, the opt-out must not mark it shared (which would emit a + # false warning and disable the error gate). + with patch(f"{_ADAPTER}._resolve_shared_provider", return_value=None): + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_passed): + adapter = LangfuseSDKAdapter.from_credentials( + public_key="pk", secret_key=SecretStr("sk"), accept_shared_provider=True + ) + assert adapter._isolation_status == ISOLATION_ISOLATED + + +def test_blank_credentials_are_rejected_at_the_boundary() -> None: + # The SDK falls back to ambient LANGFUSE_* env credentials for a blank value, + # so a blank field must fail rather than silently authenticate elsewhere. + with patch(f"{_ADAPTER}.Langfuse"): + with pytest.raises(ValueError, match="public_key"): + LangfuseSDKAdapter.from_credentials(public_key=" ", secret_key=SecretStr("sk")) + with pytest.raises(ValueError, match="secret_key"): + LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("")) + + +def test_cached_public_key_warns_that_client_config_is_discarded(caplog: Any) -> None: + # The SDK caches one client per credential, so a second construction drops + # this call's host / options; that discard must not be silent. + with patch(f"{_ADAPTER}._public_key_is_cached", return_value=True): + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_passed): + with caplog.at_level("WARNING", logger="openarmature.observability"): + LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert "cached client" in caplog.text + + +def test_sample_rate_is_applied_to_the_isolated_provider() -> None: + # The SDK applies sample_rate only on a provider it builds, which isolation + # bypasses, so the ratio has to land on OA's provider or sampling is a no-op. + from opentelemetry.sdk.trace.sampling import TraceIdRatioBased + + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_passed) as mock_lf: + LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk"), sample_rate=0.25) + provider = mock_lf.call_args.kwargs["tracer_provider"] + assert isinstance(provider.sampler, TraceIdRatioBased) + + def test_isolated_provider_reused_per_public_key() -> None: providers: list[Any] = [] @@ -263,3 +333,136 @@ def test_tracing_disabled_classified_isolated_not_undetectable() -> None: with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_tracing_disabled): adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) assert adapter._isolation_status == ISOLATION_ISOLATED + + +def test_tracing_disabled_on_a_foreign_provider_is_still_no_leak() -> None: + # Tracing off exports nothing whatever provider the cached manager holds, so + # it must not classify LEAKED and refuse the call. + def disabled_on_foreign(**kwargs: Any) -> MagicMock: + client = MagicMock() + client._resources.tracer_provider = object() + client._tracing_enabled = False + return client + + with patch(f"{_ADAPTER}.Langfuse", side_effect=disabled_on_foreign): + adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert adapter._isolation_status == ISOLATION_ISOLATED + + +# --- the guard applies to every path to an OA-constructed client -------------- + + +def test_adapter_built_client_is_guarded_through_the_plain_constructor() -> None: + # The arms live in __post_init__, so handing a from_credentials adapter to the + # ordinary constructor is guarded exactly like the observer factory. + with patch(f"{_ADAPTER}.Langfuse", side_effect=_langfuse_binding_other): + adapter = LangfuseSDKAdapter.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + with pytest.raises(LangfuseProviderIsolationUnavailable): + LangfuseObserver(client=adapter, disable_provider_payload=False) + + +def test_payload_knob_reopened_after_construction_is_still_blocked() -> None: + # The knobs are public fields on a mutable dataclass; emission consults the + # isolation status so a channel re-opened afterwards cannot leak. + obs = LangfuseObserver(client=MagicMock(_isolation_status=ISOLATION_LEAKED)) + obs.disable_provider_payload = False + assert obs._emits_provider_payload() is False + + +def test_state_channel_reopened_after_construction_falls_back_to_the_stub() -> None: + from openarmature.graph.events import InvocationStartedEvent + + obs = LangfuseObserver(client=MagicMock(_isolation_status=ISOLATION_LEAKED)) + obs.disable_state_payload = False + obs.trace_input_from_state = _hook + resolved = obs._resolve_trace_input( + InvocationStartedEvent( + initial_state={"secret": "pii"}, + invocation_id="inv-1", + correlation_id=None, + entry_node="start", + ) + ) + assert resolved == {"entry_node": "start"} # the minimal stub, no state + + +# --- behavioral: what actually lands on the observation ----------------------- + + +async def test_failed_tool_observation_omits_error_message_on_a_leaked_provider() -> None: + # The gate's effect, not just its predicate: the harvested message and type + # are absent, and Tool has no category so nothing is smuggled into + # statusMessage either. + from openarmature.graph.events import ToolCallFailedEvent + from openarmature.observability.correlation import _reset_invocation_id, _set_invocation_id + from openarmature.observability.langfuse import InMemoryLangfuseClient + + client = InMemoryLangfuseClient() + client._isolation_status = ISOLATION_LEAKED # type: ignore[attr-defined] + observer = LangfuseObserver(client=client) + token = _set_invocation_id("inv-leak") + try: + await observer( + ToolCallFailedEvent( + invocation_id="inv-leak", + correlation_id=None, + node_name="run_tool", + namespace=("run_tool",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + call_id="cc-1", + tool_name="get_weather", + tool_call_id="call_1", + arguments={"city": "Paris"}, + latency_ms=3.0, + error_type="ValueError", + error_message="rejected SSN 123-45-6789", + ) + ) + finally: + _reset_invocation_id(token) + + obs = next(o for o in client.traces["inv-leak"].observations if o.type == "tool") + assert obs.level == "ERROR" + assert "error_message" not in obs.metadata + assert "error_type" not in obs.metadata + assert obs.status_message is None + + +async def test_failed_tool_observation_keeps_error_message_when_isolated() -> None: + # The converse: an isolated client reports errors normally, so the gate does + # not break legitimate error reporting. + from openarmature.graph.events import ToolCallFailedEvent + from openarmature.observability.correlation import _reset_invocation_id, _set_invocation_id + from openarmature.observability.langfuse import InMemoryLangfuseClient + + client = InMemoryLangfuseClient() + client._isolation_status = ISOLATION_ISOLATED # type: ignore[attr-defined] + observer = LangfuseObserver(client=client) + token = _set_invocation_id("inv-ok") + try: + await observer( + ToolCallFailedEvent( + invocation_id="inv-ok", + correlation_id=None, + node_name="run_tool", + namespace=("run_tool",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + call_id="cc-2", + tool_name="get_weather", + tool_call_id="call_2", + arguments={"city": "Paris"}, + latency_ms=3.0, + error_type="TimeoutError", + error_message="tool timed out", + ) + ) + finally: + _reset_invocation_id(token) + + obs = next(o for o in client.traces["inv-ok"].observations if o.type == "tool") + assert obs.metadata.get("error_message") == "tool timed out" + assert obs.status_message == "tool timed out" From 4d93b694c8a1005033eeb1c035d6a4fc1752466b Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sun, 9 Aug 2026 20:59:44 -0700 Subject: [PATCH 3/4] Address PR review on the isolation constants and prose Export the ISOLATION_* constants from client.py's __all__. They are used cross-module by the adapter and the observer, and live in client.py so the observer can read a client's isolation status without importing the SDK-gated adapter, but they were not declared as exports and so read as module-private. Annotate the error category as ClassVar[str], matching the sibling prompt-management hierarchy; it was a bare assignment with no annotation. Correct the four failed-observation handlers' prose, which described error_type and error_message as unconditionally surfaced. They are omitted when isolation could not be established, and a tool failure carries no category so its status message stays null. --- .../observability/langfuse/client.py | 4 +++ .../observability/langfuse/errors.py | 4 ++- .../observability/langfuse/observer.py | 26 +++++++++++++++---- 3 files changed, 28 insertions(+), 6 deletions(-) diff --git a/src/openarmature/observability/langfuse/client.py b/src/openarmature/observability/langfuse/client.py index 84b6f604..98f6718e 100644 --- a/src/openarmature/observability/langfuse/client.py +++ b/src/openarmature/observability/langfuse/client.py @@ -688,6 +688,10 @@ def _get_trace(self, trace_id: str) -> LangfuseTrace: __all__ = [ + "ISOLATION_ISOLATED", + "ISOLATION_LEAKED", + "ISOLATION_SHARED_ACCEPTED", + "ISOLATION_UNDETECTABLE", "InMemoryLangfuseClient", "LangfuseClient", "LangfuseGenerationHandle", diff --git a/src/openarmature/observability/langfuse/errors.py b/src/openarmature/observability/langfuse/errors.py index 5b2b9b47..af292697 100644 --- a/src/openarmature/observability/langfuse/errors.py +++ b/src/openarmature/observability/langfuse/errors.py @@ -11,6 +11,8 @@ from __future__ import annotations +from typing import ClassVar + class ObservabilityError(Exception): """Base for errors raised by the bundled observability backends. @@ -30,4 +32,4 @@ class LangfuseProviderIsolationUnavailable(ObservabilityError): without matching the message. """ - category = "langfuse_provider_isolation_unavailable" + category: ClassVar[str] = "langfuse_provider_isolation_unavailable" diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index a7c5dbb6..6ba4cc38 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -2034,7 +2034,9 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: # message as well as the category-as-statusMessage on the # observation. error_type is null when no impl-side type was # available; the metadata key is omitted in that case so the - # absence-is-meaningful semantic is preserved. + # absence-is-meaningful semantic is preserved. Both harvested rows + # are omitted entirely when OA could not establish that the client's + # provider is isolated; the category still rides as statusMessage. if not self._omit_harvested_error(): if event.error_type is not None: metadata["error_type"] = event.error_type @@ -2093,6 +2095,8 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: ``error_type`` / ``error_message`` in metadata and as the status message) on a ToolCallFailedEvent. ``input`` (arguments) / ``output`` (result) are payload-gated per ``disable_provider_payload``. + The error rows and the status message are omitted when openarmature + could not establish that the client's provider is isolated. """ from openarmature.observability.correlation import ( current_correlation_id, @@ -2139,6 +2143,10 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: status_message: str | None = None if isinstance(event, ToolCallFailedEvent): level = "ERROR" + # Omitted when OA could not establish that the client's provider is + # isolated. A tool failure carries no error category, so nothing is + # left to put in statusMessage: it stays null rather than falling + # back to the message, which would smuggle the harvested string out. if not self._omit_harvested_error(): if event.error_type is not None: metadata["error_type"] = event.error_type @@ -2175,7 +2183,9 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non Failure (``EmbeddingFailedEvent``): ERROR level with the ``error_category`` as the status message and ``error_type`` / - ``error_message`` in metadata, mirroring the tool failure. The + ``error_message`` in metadata, mirroring the tool failure; the two + error rows are omitted when openarmature could not establish that the + client's provider is isolated, and the category still rides. The request-side ``input`` strings are still payload-gated; there is NO ``output`` (no response received). """ @@ -2245,7 +2255,9 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non handle.end(end_time=end_time) return # Failure path: request-side input_count survives; the response-derived - # rows do not. No output. ERROR level + category-as-statusMessage. + # rows do not. No output. ERROR level + category-as-statusMessage. The + # harvested error rows below are omitted when OA could not establish + # that the client's provider is isolated; the category still rides. metadata["openarmature_input_count"] = len(event.input_strings) if not self._omit_harvested_error(): if event.error_type is not None: @@ -2284,7 +2296,9 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: (``{query, documents}``) + ``output`` (scored results). Failure (``RerankFailedEvent``): ERROR level with the ``error_category`` - as the status message and ``error_type`` / ``error_message`` in + as the status message; the ``error_type`` / ``error_message`` rows are + omitted when openarmature could not establish that the client's provider + is isolated. Otherwise ``error_type`` / ``error_message`` ride in metadata, mirroring the tool / embedding failure. The request-side ``input`` is still payload-gated; there is NO ``output`` (no response received). @@ -2370,7 +2384,9 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: handle.end(end_time=end_time) return # Failure path: the request-side metadata survives; the response-derived - # rows do not. No output. ERROR level + category-as-statusMessage. + # rows do not. No output. ERROR level + category-as-statusMessage. The + # harvested error rows below are omitted when OA could not establish + # that the client's provider is isolated; the category still rides. if not self._omit_harvested_error(): if event.error_type is not None: metadata["error_type"] = event.error_type From 26fab8c31e41daeb3125593ac1e502c4b2bd0174 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sun, 9 Aug 2026 21:06:55 -0700 Subject: [PATCH 4/4] Drive every gated handler from the leak canary The canary drove five events, which left three of the six gated handlers unexercised: the LLM completion path, embeddings, and rerank. For those the leak assertion passed vacuously, and the non-vacuity check could not tell, because it only looked for the sentinels the workload happened to plant. The workload now covers every gated site with a distinct sentinel: LLM input and output on the success path, the tool result, embedding inputs and error message, and rerank query, documents and error message. The non-vacuity check derives from the full sentinel set rather than a hand-picked subset, so a channel that stops being driven fails instead of quietly weakening the leak assertion. Verified by mutation: ungating the embedding error rows or the rerank provider payload now fails the canary, neither of which it could detect before. --- .../unit/test_langfuse_payload_leak_canary.py | 188 +++++++++++++++++- 1 file changed, 184 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_langfuse_payload_leak_canary.py b/tests/unit/test_langfuse_payload_leak_canary.py index 65ff72df..a02bb9dd 100644 --- a/tests/unit/test_langfuse_payload_leak_canary.py +++ b/tests/unit/test_langfuse_payload_leak_canary.py @@ -23,10 +23,16 @@ from openarmature.graph.events import ( # noqa: E402 CaughtException, + EmbeddingEvent, + EmbeddingFailedEvent, FailureIsolatedEvent, InvocationCompletedEvent, InvocationStartedEvent, + LlmCompletionEvent, LlmFailedEvent, + RerankEvent, + RerankFailedEvent, + ToolCallEvent, ToolCallFailedEvent, ) from openarmature.observability.correlation import ( # noqa: E402 @@ -42,16 +48,47 @@ ISOLATION_LEAKED, ISOLATION_UNDETECTABLE, ) +from openarmature.retrieval import ScoredDocument # noqa: E402 -# One sentinel per harvested channel, so a leak names which channel leaked. +# One sentinel per harvested channel, so a leak names which channel leaked. Every +# gated emission site has to appear here, or its assertion below passes vacuously. STATE_IN = "CANARY-state-input-8f21" STATE_OUT = "CANARY-state-output-3b77" +LLM_IN = "CANARY-llm-input-2ba8" +LLM_OUT = "CANARY-llm-output-6f30" LLM_MSG = "CANARY-llm-error-5c04" TOOL_ARG = "CANARY-tool-argument-9d13" +TOOL_RESULT = "CANARY-tool-result-4e81" TOOL_MSG = "CANARY-tool-error-1a56" +EMBED_IN = "CANARY-embedding-input-0c95" +EMBED_MSG = "CANARY-embedding-error-7d24" +RERANK_QUERY = "CANARY-rerank-query-3a67" +RERANK_DOC = "CANARY-rerank-document-8b12" +RERANK_MSG = "CANARY-rerank-error-5e49" MARKER_MSG = "CANARY-isolated-exception-7e92" -ALL_SENTINELS = (STATE_IN, STATE_OUT, LLM_MSG, TOOL_ARG, TOOL_MSG, MARKER_MSG) +# Everything harvested. The marker message is prohibited outright rather than +# gated, so it is asserted separately as well. +ALL_SENTINELS = ( + STATE_IN, + STATE_OUT, + LLM_IN, + LLM_OUT, + LLM_MSG, + TOOL_ARG, + TOOL_RESULT, + TOOL_MSG, + EMBED_IN, + EMBED_MSG, + RERANK_QUERY, + RERANK_DOC, + RERANK_MSG, + MARKER_MSG, +) + +# The subset that must reach an isolated client, proving each channel is really +# driven. The marker is excluded: it never rides a Langfuse observation. +GATED_SENTINELS = tuple(s for s in ALL_SENTINELS if s != MARKER_MSG) _INV = "inv-canary" @@ -80,7 +117,7 @@ async def _drive_every_channel(observer: LangfuseObserver) -> None: provider="openai", model="test-model", latency_ms=1.0, - input_messages=[], + input_messages=[{"role": "user", "content": LLM_IN}], request_params={}, request_extras={}, active_prompt=None, @@ -90,6 +127,31 @@ async def _drive_every_channel(observer: LangfuseObserver) -> None: error_message=LLM_MSG, ) ) + await observer( + LlmCompletionEvent( + invocation_id=_INV, + correlation_id=None, + node_name="call_llm", + namespace=("call_llm",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="openai", + model="test-model", + response_id="resp-1", + response_model="test-model", + usage=None, + latency_ms=1.0, + finish_reason="stop", + input_messages=[{"role": "user", "content": LLM_IN}], + output_content=LLM_OUT, + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-llm-ok", + ) + ) await observer( ToolCallFailedEvent( invocation_id=_INV, @@ -108,6 +170,124 @@ async def _drive_every_channel(observer: LangfuseObserver) -> None: error_message=TOOL_MSG, ) ) + await observer( + ToolCallEvent( + invocation_id=_INV, + correlation_id=None, + node_name="run_tool", + namespace=("run_tool",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + call_id="cc-tool-ok", + tool_name="lookup", + tool_call_id="call_2", + arguments={"query": TOOL_ARG}, + result={"answer": TOOL_RESULT}, + latency_ms=1.0, + ) + ) + await observer( + EmbeddingFailedEvent( + invocation_id=_INV, + correlation_id=None, + node_name="embed", + namespace=("embed",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="openai", + model="embed-model", + latency_ms=1.0, + input_strings=[EMBED_IN], + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-embed", + error_category="provider_unavailable", + error_message=EMBED_MSG, + ) + ) + await observer( + EmbeddingEvent( + invocation_id=_INV, + correlation_id=None, + node_name="embed", + namespace=("embed",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="openai", + model="embed-model", + response_id="resp-e", + response_model="embed-model", + usage=None, + latency_ms=1.0, + input_strings=[EMBED_IN], + input_count=1, + dimensions=2, + output_vectors=[[0.1, 0.2]], + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-embed-ok", + ) + ) + await observer( + RerankFailedEvent( + invocation_id=_INV, + correlation_id=None, + node_name="rerank", + namespace=("rerank",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="cohere", + model="rerank-model", + latency_ms=1.0, + query=RERANK_QUERY, + documents=[RERANK_DOC], + document_count=1, + top_k=1, + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-rerank", + error_category="provider_unavailable", + error_message=RERANK_MSG, + ) + ) + await observer( + RerankEvent( + invocation_id=_INV, + correlation_id=None, + node_name="rerank", + namespace=("rerank",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="cohere", + model="rerank-model", + response_id="resp-r", + response_model="rerank-model", + usage=None, + latency_ms=1.0, + query=RERANK_QUERY, + documents=[RERANK_DOC], + document_count=1, + top_k=1, + result_count=1, + output_results=[ScoredDocument(index=0, relevance_score=0.9, document=RERANK_DOC)], + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-rerank-ok", + ) + ) await observer( FailureIsolatedEvent( event_name="node_failed", @@ -195,7 +375,7 @@ async def test_channels_are_actually_exercised_when_isolated() -> None: observer, client = _observer_on(ISOLATION_ISOLATED) await _drive_every_channel(observer) captured = _captured_text(client) - missing = [s for s in (STATE_IN, STATE_OUT, LLM_MSG, TOOL_ARG, TOOL_MSG) if s not in captured] + missing = [s for s in GATED_SENTINELS if s not in captured] assert not missing, f"channel not exercised, so the leak test proves nothing: {missing}"