diff --git a/conformance.toml b/conformance.toml index fcccc60..411bb24 100644 --- a/conformance.toml +++ b/conformance.toml @@ -1144,16 +1144,16 @@ note = "Pins the two-mode Langfuse client ownership model. Mode (b): LangfuseObs # Spec v0.109.0 (proposal 0115). Conformance primitives for Langfuse # provider isolation (conformance-adapter §5.5 / §6.4). [proposals."0115"] -status = "partial" +status = "implemented" since = "0.17.0" -note = "The adapter-capability half is implemented: conformance.toml declares langfuse_bound_provider_detection and the harness gates a case on requires_capability, treating a mismatched arm as a recognized skip and an undeclared capability name as an error. The §6.4 provider-faithful double and the langfuse_client construction directive are built for the two modes fixture 157 exercises (mode: supplied uses a protocol-shaped fake; mode: credentials uses a real client with its egress removed, because openarmature wraps its own client in an adapter that drives the SDK's private surface). Fixture 157 runs. Fixture 158 stays deferred on the payload-bearing / payload-free classification, the preexisting_same_key_client and accept_shared_provider sub-directives, expected_construction_error, and the level key on expected.log_records. Of the five leak assertions, the three identity ones are implemented and asserted by fixture 157; the two payload-scoped ones are declared in the expectations model so fixture 158 parses at this pin, and the Langfuse runner fails loudly if an activated fixture reaches for one, so a declared-but-unimplemented assertion cannot read as coverage." +note = "The adapter-capability half is implemented: conformance.toml declares langfuse_bound_provider_detection and the harness gates a case on requires_capability, treating a mismatched arm as a recognized skip and an undeclared capability name as an error. The §6.4 provider-faithful double and the langfuse_client construction directive are built for both modes (mode: supplied uses a protocol-shaped fake; mode: credentials uses a real client with its egress removed, because openarmature wraps its own client in an adapter that drives the SDK's private surface). Fixtures 157 and 158 both run; 158 asserts five of its eleven cases, with five recognized skips and one named per-case deferral (see the 0116 note). All five leak assertions are implemented: the three identity ones by 157, the two payload-scoped ones by 158. The payload-bearing classification follows §6.4 as narrowed by 0118, so the §8.4.1 minimal stub is payload-FREE and error_type does not make an observation payload-bearing." # Spec v0.110.0 (proposal 0116). Fail closed when Langfuse payloads # would reach a shared provider (observability §6). [proposals."0116"] status = "implemented" since = "0.17.0" -note = "The Langfuse v4 SDK caches one client per public_key, so mode (b)'s dedicated provider takes effect only when OA is the first constructor for that credential; otherwise the SDK returns the cached client on its original provider and discards OA's. OA reuses one isolated provider per credential and reads the actual binding back after construction. Where a construction-determinable payload channel is live and OA establishes the client is bound to a provider it did not isolate, construction raises a categorized LangfuseProviderIsolationUnavailable before any observation is emitted; where OA cannot establish the binding at all it suppresses every channel and logs one WARNING; accept_shared_provider=True is the single opt-out, binding the provider the application already registered rather than letting the SDK construct and globally register one. With no channel live (the default posture) an un-isolatable client neither raises nor warns. Fixture 157 runs; fixture 158 stays deferred on the payload-scoped half of the §6.4 machinery (the payload-bearing classification and the singleton / opt-out sub-directives)." +note = "The Langfuse v4 SDK caches one client per public_key, so mode (b)'s dedicated provider takes effect only when OA is the first constructor for that credential; otherwise the SDK returns the cached client on its original provider and discards OA's. OA reuses one isolated provider per credential and reads the actual binding back after construction. Where a construction-determinable payload channel is live and OA establishes the client is bound to a provider it did not isolate, construction raises a categorized LangfuseProviderIsolationUnavailable before any observation is emitted; where OA cannot establish the binding at all it suppresses every channel and logs one WARNING; accept_shared_provider=True is the single opt-out, binding the provider the application already registered rather than letting the SDK construct and globally register one. With no channel live (the default posture) an un-isolatable client neither raises nor warns. Fixtures 157 and 158 both run. Against our detection-capable declaration 158 asserts five of its eleven cases: the four raise arms plus the ungated opt-out. Five suppress-floor cases are recognized skips selected by requires_capability (§5.5), and one case is deferred by name because it drives a calls_rerank node this harness cannot yet build; it rides the retrieval fixture cluster." # Spec v0.111.0 (proposal 0117). Payload-leak invariant broadened to all # harvested-payload channels (observability §6 / §8.4). diff --git a/tests/conformance/harness/langfuse_provider_fake.py b/tests/conformance/harness/langfuse_provider_fake.py index b5bdfda..c9d6e3b 100644 --- a/tests/conformance/harness/langfuse_provider_fake.py +++ b/tests/conformance/harness/langfuse_provider_fake.py @@ -33,7 +33,7 @@ from __future__ import annotations import json -from typing import Any +from typing import Any, cast from opentelemetry import trace as otel_trace @@ -59,6 +59,24 @@ _OBSERVATION_PREFIX = "langfuse.observation." +def _render_metadata(metadata: Any) -> dict[str, Any]: + """Render metadata onto span attributes the way a Langfuse v4 client does.""" + # Mirrors `langfuse._client.attributes._flatten_and_serialize_metadata`: a dict + # becomes one attribute per key under `.`, with str / int values + # passed through unserialized; anything else becomes a single serialized blob + # under the bare prefix. Emitting the blob unconditionally, as this fake first + # did, is exactly the harness-invented summary the module header rules out -- + # and it let the classifier's error-message limb pass here while being a + # constant False against a real client. + if not isinstance(metadata, dict): + return {OBSERVATION_METADATA_ATTR: _encode(metadata)} + rendered: dict[str, Any] = {} + for key, value in cast("dict[str, Any]", metadata).items(): + name = f"{OBSERVATION_METADATA_ATTR}.{key}" + rendered[name] = value if isinstance(value, str | int) else _encode(value) + return rendered + + def _encode(value: Any) -> str: """Render a payload the way the SDK does, as a JSON string attribute.""" if isinstance(value, str): @@ -114,7 +132,8 @@ def emit(self, trace_id: str, observation: LangfuseObservation) -> None: if observation.output is not None: span.set_attribute(OBSERVATION_OUTPUT_ATTR, _encode(observation.output)) if observation.metadata: - span.set_attribute(OBSERVATION_METADATA_ATTR, _encode(observation.metadata)) + for name, value in _render_metadata(observation.metadata).items(): + span.set_attribute(name, value) span.set_attribute(OBSERVATION_LEVEL_ATTR, observation.level) if observation.status_message is not None: span.set_attribute(OBSERVATION_STATUS_MESSAGE_ATTR, observation.status_message) @@ -242,6 +261,84 @@ def force_flush(self, timeout_ms: int = 5000) -> bool: return recorded and self._binding.flush_provider(timeout_ms) +# The §8.4.1 minimal Trace stubs. openarmature writes one of these whenever the +# isolation floor suppresses the state channel, and `_resolve_trace_input` / +# `_resolve_trace_output` NEVER return None, so a classifier that treats any +# non-None trace payload as "payload present" is a constant true and 158's +# payload-free assertions could never hold. +_TRACE_STUB_KEYS = ({"entry_node"}, {"entry_node", "correlation_id"}, {"final_node", "status"}) + + +def _is_trace_stub(rendered: str) -> bool: + try: + value = json.loads(rendered) + except (TypeError, ValueError): + return False + if not isinstance(value, dict): + return False + return set(cast("dict[str, Any]", value)) in _TRACE_STUB_KEYS + + +# The metadata key whose VALUE is harvested exception text. Every other metadata +# row openarmature writes is a classification token or a lineage id. +_ERROR_MESSAGE_KEY = "error_message" + +# How a Langfuse v4 client renders a dict metadata mapping: one flattened +# attribute per key, `langfuse.observation.metadata.`. The bare, unsuffixed +# attribute appears ONLY for non-dict metadata. Reading just the bare name +# therefore misses the error-message channel entirely on a real client, which is +# the only client `mode: credentials` uses. +_FLATTENED_METADATA_PREFIX = OBSERVATION_METADATA_ATTR + "." + + +def _metadata_carries_error_message(attributes: dict[str, Any]) -> bool: + if attributes.get(_FLATTENED_METADATA_PREFIX + _ERROR_MESSAGE_KEY) is not None: + return True + # Non-dict metadata renders as one serialized blob under the bare name. Parse + # it rather than substring-matching: a caller metadata VALUE containing the + # words "error_message" would otherwise classify a payload-free observation + # as bearing. + blob = attributes.get(OBSERVATION_METADATA_ATTR) + if blob is None: + return False + try: + value = json.loads(cast("str", blob)) + except (TypeError, ValueError): + return False + return isinstance(value, dict) and _ERROR_MESSAGE_KEY in cast("dict[str, Any]", value) + + +def span_is_payload_bearing(span: Any) -> bool: + """Whether an exported observation carries harvested payload.""" + # Per §6.4 as narrowed by 0118, harvested payload is provider input / output, a + # Trace-level state input / output, or a failed observation's `error_message`. + # A payload-FREE observation is the §8.4.1 minimal stub, or a failed one + # carrying only its classifications: `error_type` is a classification token, so + # it does not make an observation payload-bearing. + # + # `status_message` is deliberately NOT a channel here. openarmature writes an + # error CATEGORY there on most failure paths, and only duplicates the harvested + # message into it alongside `metadata.error_message`. Treating the attribute + # itself as payload would misclassify every category-only failure as a leak. + # `test_harvested_status_message_never_travels_without_its_metadata_row` pins + # that coupling, so the narrower rule here cannot go quietly stale. + attributes: dict[str, Any] = dict(span.attributes or {}) + if attributes.get(OBSERVATION_INPUT_ATTR) is not None: + return True + if attributes.get(OBSERVATION_OUTPUT_ATTR) is not None: + return True + for name in (TRACE_INPUT_ATTR, TRACE_OUTPUT_ATTR): + rendered = attributes.get(name) + if rendered is not None and not _is_trace_stub(cast("str", rendered)): + return True + return _metadata_carries_error_message(attributes) + + +def payload_bearing_spans(spans: Any) -> list[Any]: + """The payload-bearing Langfuse observations among an exporter's spans.""" + return [s for s in langfuse_observation_spans(spans) if span_is_payload_bearing(s)] + + def langfuse_observation_spans(spans: Any) -> list[Any]: """The openarmature Langfuse observations among an exporter's spans.""" selected: list[Any] = [] @@ -253,6 +350,8 @@ def langfuse_observation_spans(spans: Any) -> list[Any]: __all__ = [ + "payload_bearing_spans", + "span_is_payload_bearing", "OBSERVATION_INPUT_ATTR", "OBSERVATION_METADATA_ATTR", "OBSERVATION_OUTPUT_ATTR", diff --git a/tests/conformance/harness/langfuse_real_client.py b/tests/conformance/harness/langfuse_real_client.py index 1147962..15daede 100644 --- a/tests/conformance/harness/langfuse_real_client.py +++ b/tests/conformance/harness/langfuse_real_client.py @@ -43,9 +43,7 @@ # Nothing listens here. Belt and braces alongside the processor swap below: if a # future SDK grows a second egress path, it fails fast against a closed port # rather than reaching a real project. -# -# `preexisting_same_key_client` (priming the credential before openarmature -# constructs) lands with fixture 158, which is the only fixture that uses it. + CONFORMANCE_HOST = "http://127.0.0.1:9" @@ -101,9 +99,30 @@ def langfuse_sdk_without_egress() -> Iterator[None]: resource_manager.LangfuseSpanProcessor = prior_processor # type: ignore[assignment, misc] +def prime_credential_on(tracer_provider: Any) -> Any: + """Construct a client for the conformance credential first, as an + application would, so openarmature is not the first constructor for it. + + This is `preexisting_same_key_client`. Because the SDK's own per-credential + cache is in play, the discard it models is the real one: openarmature's later + construction is handed THIS client on THIS provider and the isolated provider + it asked for is dropped. That is the mechanism the whole isolation contract + exists to detect, and no fake could have reproduced it faithfully. + """ + from langfuse import Langfuse + + return Langfuse( + public_key=CONFORMANCE_PUBLIC_KEY, + secret_key=CONFORMANCE_SECRET_KEY, + host=CONFORMANCE_HOST, + tracer_provider=tracer_provider, + ) + + __all__ = [ "CONFORMANCE_HOST", "CONFORMANCE_PUBLIC_KEY", "CONFORMANCE_SECRET_KEY", "langfuse_sdk_without_egress", + "prime_credential_on", ] diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 766c6b1..2cdd5a3 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -448,20 +448,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "Proposal 0107 mock_rerank raises sub-directive not yet wired; rides the " "remaining v0.17.0 fixture-wiring PR (observability 144-156)" ), - # 157 / 158 (proposals 0115 / 0116 / 0117 / 0118, spec v0.109.0-v0.112.0). - # The src side shipped ahead of this pin and is unit-tested; what these need - # is harness machinery: the provider-faithful Langfuse fake of - # conformance-adapter 6.4 (records observation content AND emits it through - # its bound TracerProvider, so a leak to a shared provider is catchable), the - # langfuse_client construction directive, expected_construction_error, and - # the four payload-scoped leak assertions. 158 additionally gates arms on the - # requires_capability audience gate. - "158-langfuse-payload-leak-fail-closed": ( - "needs the provider-faithful fake, the langfuse_client singleton sub-directives, " - "expected_construction_error, and the payload-scoped leak assertions; src side is " - "unit-tested meanwhile in tests/unit/test_langfuse_provider_isolation.py and " - "tests/unit/test_langfuse_payload_leak_canary.py" - ), # Spec v0.103.1 conformance coverage (0084 orphan-fallback arms + the # embedding failure-metrics counterpart). "152-otel-parallel-branch-orphan-llm-fallback": ( @@ -517,6 +503,9 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # 157 (proposals 0114 / 0115): provider isolation, driven by the dedicated # runner in the sibling harness. "157-langfuse-provider-isolation", + # 158 (proposals 0116 / 0117 / 0118): the payload-leak arms, driven by the + # same isolation runner in the sibling harness. + "158-langfuse-payload-leak-fail-closed", } ) diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py index f688a47..8c3b248 100644 --- a/tests/conformance/test_observability_langfuse.py +++ b/tests/conformance/test_observability_langfuse.py @@ -15,9 +15,11 @@ from __future__ import annotations +import contextlib import copy import json -from collections.abc import Callable, Mapping, Sequence +import logging +from collections.abc import Callable, Iterator, Mapping, Sequence from contextvars import ContextVar from dataclasses import dataclass from datetime import UTC, datetime @@ -38,6 +40,7 @@ LangfuseObserver, LangfuseTrace, ) +from openarmature.observability.langfuse.errors import ObservabilityError from openarmature.prompts import ( Prompt, PromptManager, @@ -58,12 +61,14 @@ OBSERVATION_TYPE_ATTR, ProviderFaithfulLangfuseClient, langfuse_observation_spans, + payload_bearing_spans, ) from .harness.langfuse_real_client import ( CONFORMANCE_HOST, CONFORMANCE_PUBLIC_KEY, CONFORMANCE_SECRET_KEY, langfuse_sdk_without_egress, + prime_credential_on, ) from .test_observability import _reset_otel_global_tracer_provider @@ -175,6 +180,11 @@ # `credentials` a REAL Langfuse client with its egress removed, so the # adapter path under test is the shipped one. "157-langfuse-provider-isolation", + # 158 (proposals 0116 / 0117 / 0118): the payload-leak invariant's arms. + # Against our detection-capable declaration this runs the four raise cases + # plus the ungated opt-out; the five non-capable suppress-floor cases are + # recognized skips, and one case is deferred by name below. + "158-langfuse-payload-leak-fail-closed", } ) @@ -183,9 +193,21 @@ # ``(fixture_stem, case_name)``. The case-loop in the runner ``continue``s # past matching cases — NOT ``pytest.skip``, which would skip the whole # fixture's test invocation and hide the surrounding cases that DO run. -# Currently empty; the harness covers every activated case. Kept as a named hook -# so future per-case deferrals don't need to re-introduce the pattern. -_DEFERRED_CASES: frozenset[tuple[str, str]] = frozenset() +# Each entry carries its reason, which is reported as a RecognizedSkip so the +# deferral is visible on a PASSING run rather than only inside an assertion +# failure message. +_DEFERRED_CASES: dict[tuple[str, str], str] = { + ( + "158-langfuse-payload-leak-fail-closed", + "locked_down_failure_on_detected_shared_provider_does_not_raise", + ): ( + "drives a `calls_rerank` node, and this harness's node builder has no retrieval " + "support at all (137 / 138 reach retrieval through the OTel runner instead); " + "building rerank node construction here as a side effect of an isolation change " + "is how unexercised harness code gets in, so it rides the retrieval fixture " + "cluster that 150 / 151 also wait on" + ), +} # Mocks the spec fixture 037 references for ``trace_input_from_state`` / @@ -536,19 +558,38 @@ async def fetch( async def test_langfuse_fixture(fixture_path: Path) -> None: spec = _load(fixture_path) fixture_stem = fixture_path.stem - if fixture_stem == _FIXTURE_157: + if fixture_stem in _ISOLATION_FIXTURES: cases_157 = cast("list[dict[str, Any]]", spec["cases"]) excluded_157: dict[str, str] = {} ran_157 = 0 for case in cases_157: case_name = cast("str", case.get("name") or "") + # Gate FIRST, so a deferred case still has its `requires_capability` + # block validated: checking the deferral first let a typo'd capability + # name slip past the declare-it-or-fail guard. gate = capability_skip_reason(case.get("requires_capability")) if gate is not None: excluded_157[case_name] = gate report_recognized_skip(fixture_stem, case_name, gate) continue + deferral = _DEFERRED_CASES.get((fixture_stem, case_name)) + if deferral is not None: + # Reported through the same channel as the capability gate. Writing + # the reason into `excluded_157` alone made the deferral invisible + # on a green run, because that dict is rendered only inside + # `assert_some_case_ran`'s failure message -- so the suite looked + # identical whether this fixture asserted eleven cases or five. + excluded_157[case_name] = f"per-case deferral: {deferral}" + report_recognized_skip(fixture_stem, case_name, f"per-case deferral: {deferral}") + continue ran_157 += 1 - _reject_unknown_directives(fixture_stem, case_name, case, handled_expected=_EXPECTED_157) + _reject_unknown_directives( + fixture_stem, + case_name, + case, + handled_expected=_EXPECTED_ISOLATION, + handled_case=_ISOLATION_CASE_DIRECTIVES, + ) _reject_unimplemented_assertions( fixture_stem, case_name, cast("Mapping[str, Any]", case.get("expected") or {}) ) @@ -2434,35 +2475,28 @@ def _assert_observation_tree( # how the 0118 ``metadata_absent`` directive would have landed. Anything outside # this set fails loudly instead, so a spec-side directive we have not built yet # surfaces as a gap rather than a false pass. -# Assertion keys the expectations model accepts so fixtures 157 / 158 parse at the -# v0.112.0 pin, but which no comparator implements yet (they need the -# provider-faithful fake and the `langfuse_client` construction directive). +# Every leak assertion, all implemented by `_assert_isolation_expectations`: the +# three identity keys are asserted by 157, the two payload-scoped ones by 158. # # Declaring a field to make a fixture parse is how an assertion quietly becomes # dead: the key validates, the comparator never looks at it, and the fixture reads -# like coverage. Both fixtures are deferred, so nothing should reach these today; -# this fails loudly if an activated one ever does, rather than waiting for someone -# to notice the assertion was never running. -# The identity half, implemented by `_assert_isolation_expectations`. +# like coverage. `_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS` below is the loud half +# of that guard, and every key the corpus uses must sit in one set or the other. _IMPLEMENTED_LEAK_ASSERTIONS = frozenset( { "no_langfuse_observations_on_global", "no_langfuse_observations_on_private", "langfuse_observations_on_global", - } -) - -# The payload-scoped half, still declared-for-parsing only: it needs a -# payload-bearing / payload-free classification, which arrives with fixture 158. -# Every leak assertion the corpus uses must be in one set or the other, which is -# what stops a key being neither implemented nor guarded. -_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS = frozenset( - { "no_payload_bearing_langfuse_observations_on_global", "payload_bearing_langfuse_observations_on_global", } ) +# Empty: every leak assertion the corpus uses is now implemented. Kept as the +# named hook for the next declared-but-unimplemented one, because the alternative +# is quietly allowlisting it and letting a fixture read as coverage. +_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS: frozenset[str] = frozenset() + # Case-level and `expected`-level keys this runner implements. Enumerated because # a fixture case is consumed by `if "" in case` chains, so a key nobody @@ -2520,33 +2554,41 @@ def _assert_observation_tree( # Sub-keys of `langfuse_client`, guarded separately: the case-level diff below # does not descend into nested mappings, so without this a sub-directive the # runner drops looks identical to one it honours. -_LANGFUSE_CLIENT_DIRECTIVES = frozenset({"mode", "provider"}) -_DEFERRED_LANGFUSE_CLIENT_DIRECTIVES: dict[str, str] = { - "preexisting_same_key_client": ( - "priming the SDK's per-credential cache lands with fixture 158; 157 declares it false, " - "which `langfuse_sdk_without_egress` already guarantees by clearing the cache" - ), - "accept_shared_provider": "the shared-provider opt-out lands with fixture 158", -} +_LANGFUSE_CLIENT_DIRECTIVES = frozenset( + {"mode", "provider", "preexisting_same_key_client", "accept_shared_provider"} +) +_DEFERRED_LANGFUSE_CLIENT_DIRECTIVES: dict[str, str] = {} -# Derived from the implemented-leak set rather than restating it: the three -# identity keys were spelled out here, in _EXPECTED_157 and in -# _IMPLEMENTED_LEAK_ASSERTIONS, so they could drift apart silently. -_EXPECTED_DIRECTIVES = _IMPLEMENTED_LEAK_ASSERTIONS | frozenset( +# Per-runner, because each runner is granted credit only for what it READS. +# +# The leak assertions and `log_records` live exclusively in +# `_assert_isolation_expectations`, which only the isolation runner calls. Unioning +# `_IMPLEMENTED_LEAK_ASSERTIONS` into the generic set to keep the identity keys DRY +# also handed `_run_case` credit for six keys it never reads, converting a loud +# `_reject_unimplemented_assertions` failure into silent coverage for any future +# fixture on that path. DRY across runners is the wrong axis here: what these sets +# record is which code exists, so they may only be shared where the code is. +_EXPECTED_DIRECTIVES = frozenset( { "detached_trace_count", "invariants", "langfuse_trace", "langfuse_traces", - "no_payload_bearing_langfuse_observations_on_global", - "payload_bearing_langfuse_observations_on_global", } ) # What each hand-built runner actually reads. Narrower than the generic set on -# purpose: 157 and 134 handle their own expectations and must not inherit credit -# for keys only `_run_case` implements. -_EXPECTED_157 = _IMPLEMENTED_LEAK_ASSERTIONS | frozenset({"langfuse_trace"}) +# purpose: the isolation runner and 134 handle their own expectations and must not +# inherit credit for keys only `_run_case` implements, and vice versa. +# Case-level keys only the isolation runner reads. `disable_provider_payload` has a +# nested spelling under `langfuse_observer` that `_run_case` reads and a case-level +# one that only this runner reads; granting the case-level name file-wide would let +# a fixture on the generic path declare it and have it silently dropped. +_ISOLATION_CASE_DIRECTIVES = _CASE_DIRECTIVES | frozenset( + {"disable_provider_payload", "expected_construction_error"} +) + +_EXPECTED_ISOLATION = _IMPLEMENTED_LEAK_ASSERTIONS | frozenset({"langfuse_trace", "log_records"}) _EXPECTED_134 = frozenset({"langfuse_trace", "invariants"}) # Declared by an activated fixture, NOT read by this runner. Each entry is a named @@ -2559,10 +2601,6 @@ def _assert_observation_tree( "`expected.langfuse_trace.metadata.invocation_id`, which IS asserted; the top-level " "spelling is a second, unread copy and wiring it needs a ruling on which is normative" ), - "log_records": ( - "the `level` key on log records arrives with fixture 158, whose suppress-floor cases " - "assert a WARNING; no activated fixture uses it yet" - ), } @@ -2572,16 +2610,19 @@ def _reject_unknown_directives( case: Mapping[str, Any], *, handled_expected: frozenset[str], + handled_case: frozenset[str] = _CASE_DIRECTIVES, ) -> None: """Fail on a case or expected key this runner does not act on.""" - # `handled_expected` is per-RUNNER, not file-global. Three runners live in this - # file with different coverage, so a shared set would let one runner's - # implementation vouch for a key another runner silently drops. - unknown_case = sorted(set(case) - _CASE_DIRECTIVES - _DOCUMENTARY_CASE_KEYS) + # BOTH sets are per-RUNNER, not file-global. Three runners live in this file + # with different coverage, so a shared set would let one runner's + # implementation vouch for a key another runner silently drops. That is not + # hypothetical: `disable_provider_payload` has two spellings, and only the + # isolation runner reads the case-level one. + unknown_case = sorted(set(case) - handled_case - _DOCUMENTARY_CASE_KEYS) assert not unknown_case, ( f"{fixture_stem}::{case_name} declares case-level directives this harness does not " f"read: {unknown_case}. They would otherwise be silently ignored while the case still " - f"reads like coverage; implement them and add them to _CASE_DIRECTIVES." + f"reads like coverage; implement them and add them to the runner's case-directive set." ) client_cfg = case.get("langfuse_client") if isinstance(client_cfg, dict): @@ -2886,6 +2927,8 @@ def _assert_string_or_placeholder(label: str, actual: str | None, expected: Any) # --------------------------------------------------------------------------- _FIXTURE_157 = "157-langfuse-provider-isolation" +_FIXTURE_158 = "158-langfuse-payload-leak-fail-closed" +_ISOLATION_FIXTURES = frozenset({_FIXTURE_157, _FIXTURE_158}) @dataclass @@ -2897,6 +2940,43 @@ class _IsolationCapture: isolated_exporter: Any | None # the provider openarmature built for Langfuse +# The observer method that emits the isolation-decision WARNINGs §6 mandates. +# `from_credentials` emits its own, unrelated, WARNINGs on the same logger, so the +# `log_records` assertion discriminates on this rather than on level alone. +_ISOLATION_DECISION_FUNC = "_apply_isolation_policy" + + +def _assert_isolation_decision_emitter_exists() -> None: + # Renaming the method would leave the filter above matching nothing, which + # fails as "no WARNING was emitted" and would send a reader hunting a + # behaviour change that did not happen. Name the real cause instead. + assert hasattr(LangfuseObserver, _ISOLATION_DECISION_FUNC), ( + f"LangfuseObserver has no {_ISOLATION_DECISION_FUNC}; the log_records assertion " + f"discriminates on that call site, so it must be updated alongside the rename" + ) + + +@contextlib.contextmanager +def _caplog_at_warning() -> Iterator[list[Any]]: + """Capture openarmature's observability WARNING records for the case.""" + records: list[Any] = [] + + class _Sink(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + logger = logging.getLogger("openarmature.observability") + handler = _Sink(level=logging.WARNING) + previous = logger.level + logger.addHandler(handler) + logger.setLevel(logging.WARNING) + try: + yield records + finally: + logger.removeHandler(handler) + logger.setLevel(previous) + + def _attach_exporter(provider: Any) -> Any: from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: PLC0415 from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( # noqa: PLC0415 @@ -2920,22 +3000,34 @@ def _build_isolation_observer( """Build the observer per the case's `langfuse_client` mode.""" client_cfg = cast("dict[str, Any]", case.get("langfuse_client") or {}) if cast("str", client_cfg.get("mode", "supplied")) == "credentials": - # `preexisting_same_key_client` and `accept_shared_provider` are 158's and - # land with it: 157 exercises neither, and a branch no fixture reaches is - # how an assertion ends up shipping unverified. + if bool(client_cfg.get("preexisting_same_key_client", False)): + # The application constructs FIRST, so the SDK's own per-credential + # cache hands openarmature that client on the global provider and + # discards the isolated one it asked for. That is genuinely the SDK's + # cache rather than a simulation, which is what makes this arm mean + # something. + prime_credential_on(global_provider) observer = LangfuseObserver.from_credentials( public_key=CONFORMANCE_PUBLIC_KEY, secret_key=SecretStr(CONFORMANCE_SECRET_KEY), host=CONFORMANCE_HOST, + accept_shared_provider=bool(client_cfg.get("accept_shared_provider", False)), **observer_kwargs, ) # openarmature's own isolated provider, read back so the fixture can prove # the observations went THERE rather than merely not going elsewhere. isolated = _isolated_provider_for(CONFORMANCE_PUBLIC_KEY) - # Fatal rather than "skip the check": if openarmature stops registering an - # isolated provider under this key, the positive half of the isolate case - # would vanish silently and the remaining assertions are satisfied by a run - # that emitted nothing. + if bool(client_cfg.get("accept_shared_provider", False)): + # Under the opt-out openarmature deliberately does NOT build an + # isolated provider: it binds the one the application registered. So + # there is nothing to attach to, and the case's own + # `payload_bearing_langfuse_observations_on_global` is what proves the + # observations went somewhere. + return observer, None + # Otherwise fatal rather than "skip the check": if openarmature stops + # registering an isolated provider under this key, the positive half of the + # isolate case would vanish silently, and what remains is satisfied by a run + # that emitted nothing at all. assert isolated is not None, ( "openarmature registered no isolated TracerProvider for this credential, so the " "non-vacuity half of the isolate case could not be checked" @@ -2959,7 +3051,11 @@ def _build_isolation_observer( def _assert_isolation_expectations( - expected: Mapping[str, Any], capture: _IsolationCapture, private_spans: list[Any] + expected: Mapping[str, Any], + capture: _IsolationCapture, + private_spans: list[Any], + *, + log_records: list[Any] | None = None, ) -> None: """The §5.5 identity leak assertions.""" global_langfuse = langfuse_observation_spans(capture.global_exporter.get_finished_spans()) @@ -2980,11 +3076,54 @@ def _assert_isolation_expectations( f"Langfuse observations reached openarmature's own OTel provider: " f"{[s.name for s in private_langfuse]}" ) + if expected.get("no_payload_bearing_langfuse_observations_on_global") is True: + bearing = payload_bearing_spans(capture.global_exporter.get_finished_spans()) + assert not bearing, ( + f"payload-bearing Langfuse observations reached the caller's global provider: " + f"{[s.name for s in bearing]}" + ) + if expected.get("payload_bearing_langfuse_observations_on_global") is True: + # The opt-out arm, and the non-vacuity partner for every `no_payload_bearing` + # case above: those are satisfied by a run that emitted nothing at all, so + # without a case proving payload-bearing observations CAN reach the global + # provider the set as a whole would assert very little. + bearing = payload_bearing_spans(capture.global_exporter.get_finished_spans()) + assert bearing, ( + "no payload-bearing Langfuse observation reached the caller's global provider; " + "with the shared provider accepted, openarmature proceeds onto it and its payload " + "is expected to land there" + ) if expected.get("langfuse_observations_on_global") is True: assert global_langfuse, ( "no Langfuse observation reached the caller's global provider; a supplied client " "bound to it must keep emitting there, since openarmature must not rebind it" ) + expected_logs = expected.get("log_records") + if isinstance(expected_logs, list): + # §5.5's `level` key. The suppress and opt-out arms mandate a WARNING, and + # this is the only observable half of that obligation: without it those + # cases assert the payload outcome while the "logs one WARNING" MUST goes + # unchecked. + # + # Matched by EMITTING CALL SITE, not by level alone. Every 158 case primes + # the credential, so `from_credentials` always logs its own unrelated + # cached-client WARNING on this same logger inside the capture window. A + # level-only match is satisfied by that incidental record, and downgrading + # the mandated emitter on its own then leaves the case green -- verified as + # a surviving mutant before this was narrowed. + assert log_records is not None, "log_records was declared but nothing captured logs" + _assert_isolation_decision_emitter_exists() + for wanted in cast("list[dict[str, Any]]", expected_logs): + level = cast("str", wanted["level"]) + matched = [ + r for r in log_records if r.levelname == level and r.funcName == _ISOLATION_DECISION_FUNC + ] + assert matched, ( + f"expected a {level} log record from the isolation decision " + f"({_ISOLATION_DECISION_FUNC}); captured " + f"{[(r.levelname, r.funcName, r.getMessage()[:50]) for r in log_records]}" + ) + expected_trace = expected.get("langfuse_trace") if isinstance(expected_trace, dict): # The fixture's OWN declared non-vacuity proof. Substituting a weaker @@ -3042,8 +3181,17 @@ async def _run_langfuse_157(case: Mapping[str, Any]) -> None: observer_cfg = cast("dict[str, Any]", case.get("langfuse_observer") or {}) observer_kwargs: dict[str, Any] = {} - if "disable_provider_payload" in observer_cfg: - observer_kwargs["disable_provider_payload"] = bool(observer_cfg["disable_provider_payload"]) + for flag in ("disable_provider_payload", "disable_state_payload"): + if flag in observer_cfg: + observer_kwargs[flag] = bool(observer_cfg[flag]) + # The state channel and its two hooks are payload channels in their own right + # (0117), and 158's `state_channel_preexists_raises` / `hook_preexists_raises` + # exist precisely to show the raise fires on those with the provider payload + # already suppressed. Reading only `disable_provider_payload` left those cases + # with no live channel, so nothing raised and the arm proved nothing. + for hook in ("trace_input_from_state", "trace_output_from_state"): + if hook in observer_cfg: + observer_kwargs[hook] = _resolve_trace_io_hook(cast("str", observer_cfg[hook])) llm_provider = OpenAIProvider( base_url="http://mock-llm.test", @@ -3087,13 +3235,59 @@ async def _run_langfuse_157(case: Mapping[str, Any]) -> None: if bool(case.get("caller_global_otel_active", False)): _reset_otel_global_tracer_provider(global_provider) graph = builder.compile() - langfuse_observer, isolated_exporter = _build_isolation_observer( - case, observer_kwargs, global_provider - ) - graph.attach_observer(langfuse_observer) # openarmature's OWN OTel provider: the second shared-provider spelling # §6 forbids a Langfuse observation from reaching. - graph.attach_observer(OTelObserver(span_processor=SimpleSpanProcessor(private_exporter))) + # + # Attached BEFORE the Langfuse observer is constructed, because a + # case-level `disable_provider_payload` is the negative control for + # "the two suppressions are independent" -- 158's + # `..._otel_not_suppressing` leaves the OTel side emitting payload and + # still expects the Langfuse construction to raise. Built after that + # construction, as it first was, the flag sat behind the + # `expected_construction_error` early return and never executed for the + # only case that declares it, so the case was indistinguishable from its + # sibling. Deleting the whole block left the suite green. + otel_kwargs: dict[str, Any] = {} + if "disable_provider_payload" in case: + otel_kwargs["disable_provider_payload"] = bool(case["disable_provider_payload"]) + otel_observer = OTelObserver(span_processor=SimpleSpanProcessor(private_exporter), **otel_kwargs) + if "disable_provider_payload" in case: + # Asserted, not merely passed. Under `expected_construction_error` + # the raise happens before anything is emitted, so no conforming run + # can OBSERVE this flag's effect -- which is what a negative control + # is, and why dropping the wiring above left the suite green. Pinning + # the resulting config is the part that can be checked: it proves the + # directive was read and honoured rather than accepted and dropped, + # which is what `_ISOLATION_CASE_DIRECTIVES` vouches for. + assert otel_observer.disable_provider_payload == bool(case["disable_provider_payload"]), ( + "the case-level disable_provider_payload did not reach the OTel " + "observer, so the case cannot show the two suppressions are independent" + ) + graph.attach_observer(otel_observer) + expected_construction = cast("dict[str, Any] | None", case.get("expected_construction_error")) + with _caplog_at_warning() as records: + if expected_construction is not None: + # Setup-scope assertion: construction itself is the behaviour + # under test. It must fail BEFORE an observer exists, so there + # is no run and no observation, which is the point. + with pytest.raises(ObservabilityError) as excinfo: + _build_isolation_observer(case, observer_kwargs, global_provider) + category = getattr(excinfo.value, "category", None) + assert category == expected_construction["category"], ( + f"construction error category: expected " + f"{expected_construction['category']!r}, got {category!r}" + ) + _assert_isolation_expectations( + cast("dict[str, Any]", case.get("expected") or {}), + _IsolationCapture(global_exporter, private_exporter, None), + list(private_exporter.get_finished_spans()), + log_records=records, + ) + return + langfuse_observer, isolated_exporter = _build_isolation_observer( + case, observer_kwargs, global_provider + ) + graph.attach_observer(langfuse_observer) await graph.invoke(state_cls(**cast("dict[str, Any]", case.get("initial_state") or {}))) await graph.drain() @@ -3103,6 +3297,7 @@ async def _run_langfuse_157(case: Mapping[str, Any]) -> None: cast("dict[str, Any]", case.get("expected") or {}), _IsolationCapture(global_exporter, private_exporter, isolated_exporter), list(private_exporter.get_finished_spans()), + log_records=records, ) finally: langfuse_adapter._ISOLATED_PROVIDERS.clear() diff --git a/tests/unit/test_langfuse_provider_fake.py b/tests/unit/test_langfuse_provider_fake.py index 8b1ccd2..53b16bc 100644 --- a/tests/unit/test_langfuse_provider_fake.py +++ b/tests/unit/test_langfuse_provider_fake.py @@ -12,7 +12,9 @@ from __future__ import annotations import json -from typing import Any +from collections.abc import Callable +from types import SimpleNamespace +from typing import Any, cast import pytest from opentelemetry import trace as otel_trace @@ -21,13 +23,17 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from tests.conformance.harness.langfuse_provider_fake import ( + _TRACE_STUB_KEYS, OBSERVATION_INPUT_ATTR, OBSERVATION_METADATA_ATTR, OBSERVATION_OUTPUT_ATTR, + OBSERVATION_STATUS_MESSAGE_ATTR, OBSERVATION_TYPE_ATTR, TRACE_INPUT_ATTR, ProviderFaithfulLangfuseClient, + _is_trace_stub, langfuse_observation_spans, + span_is_payload_bearing, ) @@ -96,7 +102,10 @@ def test_error_message_metadata_reaches_the_span() -> None: ) attrs = _attrs(langfuse_observation_spans(exporter.get_finished_spans())[0]) - assert "LEAKED EXCEPTION TEXT" in attrs[OBSERVATION_METADATA_ATTR] + # Flattened per key, as a real v4 client renders a dict. The bare + # OBSERVATION_METADATA_ATTR name carries nothing for dict metadata. + assert attrs[f"{OBSERVATION_METADATA_ATTR}.error_message"] == "LEAKED EXCEPTION TEXT" + assert OBSERVATION_METADATA_ATTR not in attrs def test_trace_payload_reaches_the_provider() -> None: @@ -228,3 +237,261 @@ def test_recording_is_unaffected_by_the_provider_binding(tracing_enabled: bool) observation = client.traces["t"].observations[0] assert observation.name == "gen" assert observation.output == "hello" + + +# -- payload-bearing classification (§6.4, narrowed by 0118) ---------------- + + +def _one_span(build: Callable[[ProviderFaithfulLangfuseClient], None]) -> Any: + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + build(client) + client.force_flush() + return langfuse_observation_spans(exporter.get_finished_spans())[0] + + +def test_provider_payload_is_payload_bearing() -> None: + span = _one_span(lambda c: c.generation(trace_id="t", name="gen").end(output="LEAKED")) + assert span_is_payload_bearing(span) is True + + +def test_error_message_is_payload_bearing() -> None: + # The third harvested channel (0117): a failed observation's error_message. + span = _one_span( + lambda c: c.tool(trace_id="t", name="run_tool").end( + metadata={"error_type": "ValueError", "error_message": "LEAKED EXCEPTION TEXT"} + ) + ) + assert span_is_payload_bearing(span) is True + + +def test_error_type_alone_is_payload_free() -> None: + # 0118 narrowed the rule to the MESSAGE. error_type is a classification + # token, so a failed observation carrying only classifications is + # payload-free; treating it as bearing would make 158's assertions + # unsatisfiable for any failed observation. + span = _one_span( + lambda c: c.tool(trace_id="t", name="run_tool").end(metadata={"error_type": "ValueError"}) + ) + assert span_is_payload_bearing(span) is False + + +@pytest.mark.parametrize( + "stub", + [ + {"entry_node": "ask"}, + {"entry_node": "ask", "correlation_id": "c-1"}, + {"final_node": "ask", "status": "completed"}, + ], +) +def test_the_minimal_trace_stub_is_payload_free(stub: dict[str, Any]) -> None: + # The load-bearing one. openarmature writes a stub whenever the isolation + # floor suppresses the state channel, and _resolve_trace_input / _output + # never return None, so classifying any non-None trace payload as present + # would be a constant true and 158 could never pass. + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.update_trace(id="t", input=stub) + client.span(trace_id="t", name="node").end() + span = langfuse_observation_spans(exporter.get_finished_spans())[0] + assert span_is_payload_bearing(span) is False + + +def test_raw_state_on_the_trace_is_payload_bearing() -> None: + # Non-vacuity for the stub cases: a real state payload through the same + # channel IS bearing, so the stub result comes from its shape rather than + # from the trace channel being ignored. + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.update_trace(id="t", input={"raw_user_text": "LEAKED STATE"}) + client.span(trace_id="t", name="node").end() + span = langfuse_observation_spans(exporter.get_finished_spans())[0] + assert span_is_payload_bearing(span) is True + + +def _stub_resolving_observer(*, blocked: bool) -> Any: + from openarmature.observability.langfuse import InMemoryLangfuseClient + from openarmature.observability.langfuse.client import ISOLATION_LEAKED + from openarmature.observability.langfuse.observer import LangfuseObserver + + client = InMemoryLangfuseClient() + if blocked: + # The isolation floor, which closes the state channel whatever the knobs + # say. Set dynamically because the observer reads it with getattr, so it + # is not a declared attribute on the client. + cast("Any", client)._isolation_status = ISOLATION_LEAKED # noqa: SLF001 + # disable_state_payload drives the OTHER path to the same stub (lever 3), so + # both spellings get walked below rather than only the isolation one. + return LangfuseObserver(client=client, disable_state_payload=True) + + +@pytest.mark.parametrize("blocked", [True, False]) +@pytest.mark.parametrize("correlation_id", [None, "c-1"]) +def test_every_stub_the_observer_can_emit_is_a_known_stub_shape( + blocked: bool, correlation_id: str | None +) -> None: + # `_is_trace_stub` hardcodes the key sets, so it is correct only while those + # are what openarmature actually emits. Drive the resolvers and read the keys + # off the RESULT: a stub gaining, losing, or renaming a key fails here, which + # a source-substring match could not detect. That matters because the + # conformance cases that would catch the fallout are capability skips for this + # adapter, so the classifier could be silently wrong with a green suite. + observer = _stub_resolving_observer(blocked=blocked) + started = SimpleNamespace(entry_node="ask", correlation_id=correlation_id, initial_state=None) + completed = SimpleNamespace(final_node="ask", status="completed", final_state=None) + + for resolved in ( + observer._resolve_trace_input(cast("Any", started)), # noqa: SLF001 + observer._resolve_trace_output(cast("Any", completed)), # noqa: SLF001 + ): + assert isinstance(resolved, dict) + keys = set(cast("dict[str, Any]", resolved)) + assert keys in _TRACE_STUB_KEYS, ( + f"observer emitted stub keys {sorted(keys)}, which _TRACE_STUB_KEYS does not " + f"recognise; every suppressed trace would reclassify as payload-bearing" + ) + assert _is_trace_stub(json.dumps(resolved, sort_keys=True)) is True + + +_TOOL_ERROR = "CANARY-tool-error-message-4b7c" + + +async def _emit_failed_tool(observer: Any) -> None: + from openarmature.graph.events import InvocationStartedEvent, ToolCallFailedEvent + from openarmature.observability.correlation import _reset_invocation_id, _set_invocation_id + + invocation_id = "inv-status-message" + token = _set_invocation_id(invocation_id) + try: + await observer( + InvocationStartedEvent( + initial_state={}, + invocation_id=invocation_id, + correlation_id=None, + entry_node="run_tool", + ) + ) + await observer( + ToolCallFailedEvent( + invocation_id=invocation_id, + 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={}, + latency_ms=1.0, + error_type="ValueError", + error_message=_TOOL_ERROR, + ) + ) + finally: + _reset_invocation_id(token) + + +def test_the_classifier_agrees_with_a_real_langfuse_client() -> None: + # The decisive one. Every fixture-158 case is `mode: credentials`, i.e. a REAL + # client, and the fake is the only client that ever wrote metadata as a single + # blob. Classifying on that blob made the error-message limb a constant False + # on the path the fixture actually drives, while this file's other tests -- all + # driving the fake -- passed. A double must not be the only witness for the + # behaviour it exists to model, so drive the real SDK here. + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + + from tests.conformance.harness.langfuse_real_client import ( + CONFORMANCE_HOST, + CONFORMANCE_PUBLIC_KEY, + CONFORMANCE_SECRET_KEY, + langfuse_sdk_without_egress, + ) + + with langfuse_sdk_without_egress(): + from langfuse import Langfuse + + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + client = Langfuse( + public_key=CONFORMANCE_PUBLIC_KEY, + secret_key=CONFORMANCE_SECRET_KEY, + host=CONFORMANCE_HOST, + tracer_provider=provider, + ) + # Exactly the shape observer.py renders for a failed observation. + failed = client.start_observation( + name="run_tool", + as_type="span", + metadata={"error_type": "ValueError", "error_message": "LEAKED EXCEPTION TEXT"}, + ) + failed.update(status_message="LEAKED EXCEPTION TEXT", level="ERROR") + failed.end() + # And a category-only failure, which must stay payload-free. + classified = client.start_observation( + name="classified", as_type="span", metadata={"error_type": "ValueError"} + ) + classified.update(status_message="provider_unavailable", level="ERROR") + classified.end() + generation = client.start_observation( + name="gen", as_type="generation", input="PROMPT", output="COMPLETION" + ) + generation.end() + provider.force_flush() + + by_name = {s.name: s for s in langfuse_observation_spans(exporter.get_finished_spans())} + assert set(by_name) == {"run_tool", "classified", "gen"} + assert span_is_payload_bearing(by_name["run_tool"]) is True + assert span_is_payload_bearing(by_name["classified"]) is False + assert span_is_payload_bearing(by_name["gen"]) is True + + +@pytest.mark.parametrize("disable_provider_payload", [False, True]) +@pytest.mark.asyncio +async def test_harvested_status_message_never_travels_without_its_metadata_row( + disable_provider_payload: bool, +) -> None: + # `span_is_payload_bearing` classifies on `metadata.error_message` and treats + # `status_message` as payload-free, because openarmature writes an error + # CATEGORY there on most failure paths and classifying the attribute itself + # would misread every category-only failure as a leak. That narrower rule is + # safe only while the harvested message reaches status_message ONLY alongside + # the metadata row the classifier does read. Drive a real failed observation + # both ways and assert the coupling, so routing the message to status_message + # alone fails here instead of going quiet. + from openarmature.observability.langfuse.observer import LangfuseObserver + + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + observer = LangfuseObserver(client=client, disable_provider_payload=disable_provider_payload) + await _emit_failed_tool(observer) + client.force_flush() + + spans = [ + s + for s in langfuse_observation_spans(exporter.get_finished_spans()) + if _attrs(s).get(OBSERVATION_TYPE_ATTR) == "tool" + ] + assert spans, "the failed tool observation never reached the provider" + attrs = _attrs(spans[0]) + status_message = attrs.get(OBSERVATION_STATUS_MESSAGE_ATTR) + metadata_row = attrs.get(f"{OBSERVATION_METADATA_ATTR}.error_message") + + if status_message == _TOOL_ERROR: + assert metadata_row == _TOOL_ERROR, ( + "the harvested message reached status_message without the " + "metadata.error_message row span_is_payload_bearing reads, so the leak " + "would classify payload-free" + ) + assert span_is_payload_bearing(spans[0]) is True + else: + # Suppressed: status_message carries only the classification token. + assert metadata_row is None + assert span_is_payload_bearing(spans[0]) is False