diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b23ab5..4f8809a 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), 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). +- **Langfuse observer: credentials-in construction with tracer-provider isolation** (proposals 0114 + 0116 + 0117 + 0118, observability §6 / §8.9 / §8.4, spec v0.108.0 / v0.110.0 / v0.111.0 / v0.112.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` -- 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. A failed observation's `error_message` is harvested exception text, so `disable_provider_payload` governs it for every failure category on all four provider observations: with payloads off it is not rendered, and the error category still rides as the status message where the event carries one. A Tool failure has no category, so its status message is null rather than falling back to the exception string. `error_type` is a classification token rather than harvested content and is never gated, which matters most for a Tool failure where it is the only remaining discriminator; it is optional, so it is emitted only where the failure event supplies one. 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 through v0.112.0 are beyond the current v0.107.0 pin, so this ships ahead of the pin (unit-tested); the conformance fixtures (157 / 158 / 159, proposals 0115 / 0116 / 0117 / 0118) ride the pin bump, as do fixtures 098 / 137 / 138, which are deferred meanwhile because they still assert the pre-0118 shape. ### Changed diff --git a/docs/agent/non-obvious-shapes.md b/docs/agent/non-obvious-shapes.md index bc99488..cf7a97f 100644 --- a/docs/agent/non-obvious-shapes.md +++ b/docs/agent/non-obvious-shapes.md @@ -119,7 +119,7 @@ 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` 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. +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). A failed observation's `error_message` is harvested exception text, so `disable_provider_payload` governs it too, on all four mapped provider observations (LLM, Embedding, Tool, Retriever) and for every failure category: with payloads off it is simply not rendered. The error category still rides as the status message where the event carries one, so you can still see what kind of failure it was; a Tool failure has no category, so its status message is null rather than falling back to the exception string. `error_type` is a classification token, not harvested content, so it is never gated, which matters most for a Tool failure where it is the only discriminator left. 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. diff --git a/docs/concepts/observability.md b/docs/concepts/observability.md index ea6eef8..0e227bf 100644 --- a/docs/concepts/observability.md +++ b/docs/concepts/observability.md @@ -1239,6 +1239,22 @@ serialized form with the §5.5.5 truncation marker parsing back to native shape. The unparseable JSON IS the truncation signal in the Langfuse UI. +`disable_provider_payload` also governs a **failed** observation's +`error_message`, on all four provider observations (Generation, +Embedding, Tool, Retriever) and for every failure category. An +exception string is harvested runtime content: a provider 4xx routinely +quotes the request or the flagged prompt, and a structured-output +failure quotes the model's own output. So with payloads at their +default-off, a failed observation carries no exception text. + +What remains is enough to triage the failure. `error_type` is a +classification token (an exception class name or vendor code), never +gated, and the error category still rides as the observation's status +message wherever the event carries one. A **tool** failure has no +category, so its status message is null rather than falling back to the +exception string. The full exception text is unaffected on the OTel +side. + ### Prompt linkage When a Prompt's source backend exposes a Langfuse Prompt entity diff --git a/src/openarmature/AGENTS.md b/src/openarmature/AGENTS.md index 494b873..08b242d 100644 --- a/src/openarmature/AGENTS.md +++ b/src/openarmature/AGENTS.md @@ -1601,7 +1601,7 @@ 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` 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. +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). A failed observation's `error_message` is harvested exception text, so `disable_provider_payload` governs it too, on all four mapped provider observations (LLM, Embedding, Tool, Retriever) and for every failure category: with payloads off it is simply not rendered. The error category still rides as the status message where the event carries one, so you can still see what kind of failure it was; a Tool failure has no category, so its status message is null rather than falling back to the exception string. `error_type` is a classification token, not harvested content, so it is never gated, which matters most for a Tool failure where it is the only discriminator left. 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. diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 6ba4cc3..8fac4e3 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -476,25 +476,34 @@ def _apply_isolation_policy(self) -> None: ) 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" - ) + # construction-time channel. Warned only when a channel was actually + # live: with the default posture nothing is being taken away, and a + # warning there would contradict the documented "an un-isolatable + # client is harmless and neither raises nor warns". + if self._construction_channels_live(): + _logger.warning( + "cannot establish the Langfuse client's TracerProvider binding; " + "suppressing the provider and state payloads you enabled, 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. + # Reachable only when no construction-time channel is live, so nothing + # is refused and nothing is currently leaking. Reported because the + # binding is a latent problem, and the two ways of enabling a channel + # fail closed differently: re-opening a knob on THIS observer is caught + # at emission by _isolation_blocks_payload() and the payload is + # withheld, while constructing a NEW observer over the same client with + # a channel live raises in __post_init__. Deliberately silent about the + # error message, which disable_provider_payload governs, not this status. _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" + "no payload channel is enabled, so nothing is being exported to it; " + "enabling one fails closed (the payload is withheld, or construction is " + "refused) until OA's client is constructed before any other for this key" ) elif status == ISOLATION_SHARED_ACCEPTED: # A provider-binding decision, not a payload one, so it is reported @@ -575,12 +584,27 @@ def from_credentials( # observer built by handing a from_credentials adapter to the constructor. 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 self._isolation_blocks_payload() + # NOTE: an emitted error_message is written verbatim, not through the + # payload_byte_cap truncation every other payload-classified field uses. The + # cap is not applied because 0118 classifies the field for GATING without + # saying it is subject to §5.5.5 truncation, and fixtures 150/151 exist to + # assert the message LITERALLY, which truncation would contradict. A provider + # that returns a very large exception string therefore renders it in full. + # Raised for the batched spec review rather than changed unilaterally. + def _emits_harvested_error_message(self) -> bool: + # A failed observation's error_message is harvested exception text, so the + # provider-payload flag governs it (0118) for every failure category and on + # every provider observation -- the category does not tell you what the + # string contains, and a provider 4xx routinely quotes the request or the + # flagged prompt. + # + # There is no separate isolation check: once the flag covers the field the + # §6 arms already decide every configuration. Flag on, nothing is emitted + # to suppress; flag off on an isolated provider, it emits legitimately; + # flag off on a detected shared provider, construction raised; flag off + # where isolation could not be established, suppress-all set the flag; flag + # off with the caller opted in, it emits as an acknowledged leak. + return self._emits_provider_payload() async def __call__( self, @@ -2030,22 +2054,15 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: calling_branch_name_chain=event.branch_name_chain, ) metadata = self._typed_event_metadata(event, correlation_id) - # Failure-specific metadata rows: surface error_type + error_ - # 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. 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 - # 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 + # error_type is a classification token, not harvested content, so it is + # never gated; it is optional and absent when no impl-side type was + # available, and the metadata key is omitted in that case so the + # absence-is-meaningful semantic is preserved. The category rides as + # statusMessage either way. + if event.error_type is not None: + metadata["error_type"] = event.error_type + if self._emits_harvested_error_message(): + metadata["error_message"] = event.error_message model_parameters: dict[str, Any] = dict(event.request_params or {}) input_value: Any = None output_value: Any = None @@ -2095,8 +2112,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. + The ``error_message`` row and the status message follow + ``disable_provider_payload``; ``error_type`` is never gated. """ from openarmature.observability.correlation import ( current_correlation_id, @@ -2143,13 +2160,15 @@ 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 + # error_type is a classification token and is never gated, which + # matters most here: a tool failure carries no error category, so it + # is the only failure discriminator the observation has. + if event.error_type is not None: + metadata["error_type"] = event.error_type + # A tool failure has no category to fall back on, so when the message + # is withheld statusMessage stays null rather than taking the message + # instead, which would smuggle the harvested string out. + if self._emits_harvested_error_message(): 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) @@ -2182,12 +2201,11 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non strings + ``output`` vectors. Failure (``EmbeddingFailedEvent``): ERROR level with the - ``error_category`` as the status message and ``error_type`` / - ``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). + ``error_category`` as the status message; the ``error_message`` row + follows ``disable_provider_payload`` while ``error_type`` is never + gated, mirroring the tool / rerank failure. The request-side ``input`` + strings are still payload-gated; there is NO ``output`` (no response + received). """ from openarmature.observability.correlation import ( current_correlation_id, @@ -2255,13 +2273,13 @@ 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. The - # harvested error rows below are omitted when OA could not establish - # that the client's provider is isolated; the category still rides. + # rows do not. No output. ERROR level + category-as-statusMessage. metadata["openarmature_input_count"] = len(event.input_strings) - if not self._omit_harvested_error(): - if event.error_type is not None: - metadata["error_type"] = event.error_type + # error_type is a classification token and is never gated; the category + # rides as statusMessage whether or not the message is emitted. + if event.error_type is not None: + metadata["error_type"] = event.error_type + if self._emits_harvested_error_message(): 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( @@ -2296,10 +2314,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; 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 + as the status message; the ``error_message`` row follows + ``disable_provider_payload`` while ``error_type`` is never gated, + mirroring the tool / embedding failure. The request-side ``input`` is still payload-gated; there is NO ``output`` (no response received). """ @@ -2384,12 +2401,12 @@ 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. 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 + # rows do not. No output. ERROR level + category-as-statusMessage. + # error_type is a classification token and is never gated; the category + # rides as statusMessage whether or not the message is emitted. + if event.error_type is not None: + metadata["error_type"] = event.error_type + if self._emits_harvested_error_message(): 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( diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 76340be..e63185d 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -259,7 +259,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "095-tool-call-id-links-to-llm-request", "096-tool-call-payload-gating", "097-otel-tool-span-attributes", - "098-langfuse-tool-observation", # v0.16.0 — proposal 0059 embedding observability (0059b). A # calls_embed node awaits OpenAIEmbeddingProvider.embed() inside the # node body; the typed EmbeddingEvent / EmbeddingFailedEvent drive the @@ -279,7 +278,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "081-embedding-event-active-prompt-populated", "082-otel-embedding-span-attributes", "083-langfuse-embedding-observation", - "137-langfuse-embedding-failure-observation", "139-otel-embedding-no-usage-input-tokens-omitted", "140-langfuse-embedding-no-usage-usagedetails-omitted", # proposal 0067 §11 embedding metrics: token.usage (input only) + @@ -306,7 +304,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "106-rerank-event-active-prompt-populated", "107-otel-rerank-span-attributes", "108-langfuse-rerank-observation", - "138-langfuse-rerank-failure-observation", "141-otel-rerank-no-usage-attributes-omitted", "142-langfuse-rerank-no-usage-usagedetails-omitted", "109-rerank-metrics-token-and-duration", @@ -326,6 +323,33 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # # _DEFERRED_FIXTURES — not run because the capability is unimplemented. _DEFERRED_FIXTURES: dict[str, str] = { + # Proposal 0118 (spec v0.112.0) gates a failed observation's error_message on + # disable_provider_payload. The behavior ships ahead of the pin, so at the + # pinned v0.107.0 these three still assert the message PRESENT under the + # default posture, which 0118 forbids. Spec reconciles all three at v0.112.0 + # (137/138 assert it absent with error_type retained; 098 gains the tool + # anti-smuggling case). + # + # Un-deferring needs BOTH the pin bump AND the `metadata_absent` directive in + # _assert_langfuse_observation_tree: 137/138 express the withholding through + # that directive alone, so re-listing them while it is unimplemented restores + # a vacuous pass rather than coverage. 098 is unaffected (it asserts a null + # statusMessage, which is implemented). The directive landed in #267, so the + # pin bump is now the only remaining prerequisite; the source behavior here is + # covered meanwhile by the default-posture unit tests in + # tests/unit/test_langfuse_provider_isolation.py. + "098-langfuse-tool-observation": ( + "Proposal 0118 error-message gating: fixture asserts the pre-0118 shape; " + "reconciled at the v0.112.0 pin bump" + ), + "137-langfuse-embedding-failure-observation": ( + "Proposal 0118 error-message gating: fixture asserts the pre-0118 shape; " + "reconciled at the v0.112.0 pin bump" + ), + "138-langfuse-rerank-failure-observation": ( + "Proposal 0118 error-message gating: fixture asserts the pre-0118 shape; " + "reconciled at the v0.112.0 pin bump" + ), # Proposal 0045 IS implemented (v0.11.0), but the nested-case Langfuse # fixture stays deferred: it needs runtime-state item-list lookup for # nested fan-outs plus an augment_metadata_from_outer_item directive @@ -415,10 +439,16 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # Proposal 0107 (spec v0.102.0) mock_embedding / mock_rerank raises # sub-directive -> literal error-field assertion. "150-langfuse-embedding-failure-literal-error-fields": ( - "Proposal 0107 mock-raises literal error fields; harness wiring rides the v0.17.0 fixture-wiring PR" + "Proposal 0107 mock-raises literal error fields; harness wiring rides the v0.17.0 " + "fixture-wiring PR. Also asserts the pre-0118 shape at this pin: spec moves both " + "cases to disable_provider_payload=false at v0.112.0, since asserting the message " + "literally is their purpose" ), "151-langfuse-rerank-failure-literal-error-fields": ( - "Proposal 0107 mock-raises literal error fields; harness wiring rides the v0.17.0 fixture-wiring PR" + "Proposal 0107 mock-raises literal error fields; harness wiring rides the v0.17.0 " + "fixture-wiring PR. Also asserts the pre-0118 shape at this pin: spec moves both " + "cases to disable_provider_payload=false at v0.112.0, since asserting the message " + "literally is their purpose" ), # Spec v0.103.1 conformance coverage (0084 orphan-fallback arms + the # embedding failure-metrics counterpart). @@ -5036,6 +5066,12 @@ async def _run_tool_case(case: Mapping[str, Any]) -> None: lf_kwargs: dict[str, Any] = {"client": langfuse_client} if "disable_provider_payload" in case: lf_kwargs["disable_provider_payload"] = bool(case["disable_provider_payload"]) + # A per-observer block overrides the shared top-level flag, so a fixture + # can drive the two observers at different postures; mirrors the + # embedding and rerank runners. + lf_cfg = cast("dict[str, Any] | None", case.get("langfuse_observer")) or {} + if "disable_provider_payload" in lf_cfg: + lf_kwargs["disable_provider_payload"] = bool(lf_cfg["disable_provider_payload"]) graph.attach_observer(LangfuseObserver(**lf_kwargs)) try: diff --git a/tests/unit/test_langfuse_payload_leak_canary.py b/tests/unit/test_langfuse_payload_leak_canary.py index a02bb9d..e0fb493 100644 --- a/tests/unit/test_langfuse_payload_leak_canary.py +++ b/tests/unit/test_langfuse_payload_leak_canary.py @@ -408,3 +408,17 @@ def test_adapter_built_client_is_guarded_on_every_construction_path() -> None: LangfuseObserver(client=leaked_client, disable_state_payload=False) with pytest.raises(LangfuseProviderIsolationUnavailable): LangfuseObserver(client=leaked_client, trace_input_from_state=lambda s: s) + + +async def test_payloads_off_withholds_every_harvested_channel_on_an_isolated_client() -> None: + # The case that separates the payload flag from the isolation status: the + # provider IS isolated, so an isolation-only gate would emit everything, but + # payloads are off so nothing harvested may be rendered. Without this, a site + # reverted to the retired isolation predicate passes the whole suite. + client = InMemoryLangfuseClient() + client._isolation_status = ISOLATION_ISOLATED # type: ignore[attr-defined] + observer = LangfuseObserver(client=client) # every payload knob at its default + await _drive_every_channel(observer) + captured = _captured_text(client) + leaked = [s for s in ALL_SENTINELS if s in captured] + assert not leaked, f"payloads are disabled but harvested content was rendered: {leaked}" diff --git a/tests/unit/test_langfuse_provider_isolation.py b/tests/unit/test_langfuse_provider_isolation.py index b3fd00c..6de435d 100644 --- a/tests/unit/test_langfuse_provider_isolation.py +++ b/tests/unit/test_langfuse_provider_isolation.py @@ -252,7 +252,7 @@ def test_observer_undetectable_suppresses_all_channels_and_warns(caplog: Any) -> 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 + assert "suppressing the provider and state payloads you enabled" in caplog.text def test_observer_accept_shared_provider_warns_and_proceeds(caplog: Any) -> None: @@ -310,21 +310,30 @@ def test_all_channels_off_on_leak_does_not_raise() -> None: 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. +def test_error_message_follows_the_payload_flag_not_the_isolation_status() -> None: + # 0118: the flag governs the harvested error message. With payloads off it is + # never emitted, whatever the provider turned out to be -- including a plain + # caller-supplied client, which records no isolation status at all. + for status in (ISOLATION_ISOLATED, ISOLATION_SHARED_ACCEPTED, None): + obs = LangfuseObserver(client=MagicMock(_isolation_status=status)) + assert obs.disable_provider_payload is True # the default posture + assert obs._emits_harvested_error_message() is False + + +def test_error_message_emits_with_payloads_on_and_an_isolated_provider() -> None: + obs = LangfuseObserver( + client=MagicMock(_isolation_status=ISOLATION_ISOLATED), disable_provider_payload=False + ) + assert obs._emits_harvested_error_message() is True + + +@pytest.mark.parametrize("status", [ISOLATION_LEAKED, ISOLATION_UNDETECTABLE]) +def test_error_message_withheld_on_a_provider_not_established_as_isolated(status: str) -> None: + # Not a second gate: the §6 arms already closed the flag for these statuses + # (suppress-all), or refused construction outright, so the message cannot ride. obs = LangfuseObserver(client=MagicMock(_isolation_status=status)) - assert obs._omit_harvested_error() is expected + obs.disable_provider_payload = False # a caller reopening it post-construction + assert obs._emits_harvested_error_message() is False def test_tracing_disabled_classified_isolated_not_undetectable() -> None: @@ -389,17 +398,17 @@ def test_state_channel_reopened_after_construction_falls_back_to_the_stub() -> N # --- 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. +async def test_failed_tool_observation_omits_message_under_the_default_posture() -> None: + # The gate's effect, not just its predicate. error_type is NOT gated (0118) -- + # it is the only failure discriminator a tool observation has, since a tool + # failure carries no error category -- but the message is withheld and must + # not be smuggled into statusMessage in its place. 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) + observer = LangfuseObserver(client=client) # disable_provider_payload defaults True token = _set_invocation_id("inv-leak") try: await observer( @@ -426,20 +435,19 @@ async def test_failed_tool_observation_omits_error_message_on_a_leaked_provider( 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.metadata.get("error_type") == "ValueError" # ungated classification token 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. +async def test_failed_tool_observation_keeps_error_message_with_payloads_on() -> None: + # The converse: with payloads enabled the message reports normally, so the + # gate does not break legitimate error triage. 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) + observer = LangfuseObserver(client=client, disable_provider_payload=False) token = _set_invocation_id("inv-ok") try: await observer( @@ -466,3 +474,140 @@ async def test_failed_tool_observation_keeps_error_message_when_isolated() -> No 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" + + +# The Tool test above covers one of the four gated sites. These cover the other +# three: without them, reverting any of the LLM / Embedding / Retriever handlers +# to the retired isolation predicate passes the whole suite. + + +async def _one_observation(observer: LangfuseObserver, client: Any, event: Any, obs_type: str) -> Any: + from openarmature.observability.correlation import _reset_invocation_id, _set_invocation_id + + token = _set_invocation_id(_INV_DEFAULT) + try: + await observer(event) + finally: + _reset_invocation_id(token) + return next(o for o in client.traces[_INV_DEFAULT].observations if o.type == obs_type) + + +_INV_DEFAULT = "inv-default-posture" + + +async def test_failed_llm_generation_omits_message_under_the_default_posture() -> None: + from openarmature.graph.events import LlmFailedEvent + from openarmature.observability.langfuse import InMemoryLangfuseClient + + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client) # disable_provider_payload defaults True + obs = await _one_observation( + observer, + client, + LlmFailedEvent( + invocation_id=_INV_DEFAULT, + correlation_id=None, + node_name="call_llm", + namespace=("call_llm",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="openai", + model="m", + latency_ms=1.0, + input_messages=[], + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-1", + error_category="provider_unavailable", + error_message="upstream said: prompt was 'secret'", + ), + "generation", + ) + assert "error_message" not in obs.metadata + assert obs.status_message == "provider_unavailable" # the category still rides + + +async def test_failed_embedding_omits_message_under_the_default_posture() -> None: + from openarmature.graph.events import EmbeddingFailedEvent + from openarmature.observability.langfuse import InMemoryLangfuseClient + + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client) + obs = await _one_observation( + observer, + client, + EmbeddingFailedEvent( + invocation_id=_INV_DEFAULT, + correlation_id=None, + node_name="embed", + namespace=("embed",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="openai", + model="m", + latency_ms=1.0, + input_strings=["x"], + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-2", + error_category="provider_unavailable", + error_message="upstream said: input was 'secret'", + ), + "embedding", + ) + assert "error_message" not in obs.metadata + assert obs.status_message == "provider_unavailable" + + +async def test_failed_rerank_omits_message_under_the_default_posture() -> None: + from openarmature.graph.events import RerankFailedEvent + from openarmature.observability.langfuse import InMemoryLangfuseClient + + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client) + obs = await _one_observation( + observer, + client, + RerankFailedEvent( + invocation_id=_INV_DEFAULT, + correlation_id=None, + node_name="rerank", + namespace=("rerank",), + attempt_index=0, + fan_out_index=None, + branch_name=None, + provider="cohere", + model="m", + latency_ms=1.0, + query="q", + documents=["d"], + document_count=1, + top_k=1, + request_params={}, + request_extras={}, + active_prompt=None, + active_prompt_group=None, + call_id="cc-3", + error_category="provider_unavailable", + error_message="upstream said: query was 'secret'", + ), + "retriever", + ) + assert "error_message" not in obs.metadata + assert obs.status_message == "provider_unavailable" + + +def test_undetectable_under_the_default_posture_neither_raises_nor_warns(caplog: Any) -> None: + # With no channel live the suppress arm takes nothing away, so warning there + # would contradict the documented "an un-isolatable client is harmless". + with _patched_adapter(ISOLATION_UNDETECTABLE): + with caplog.at_level("WARNING", logger="openarmature.observability"): + obs = LangfuseObserver.from_credentials(public_key="pk", secret_key=SecretStr("sk")) + assert obs.disable_provider_payload is True + assert "cannot establish" not in caplog.text