Activate conformance fixture 158 payload-leak arms

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).

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 `<prefix>.<key>`, 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.<key>`. 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",

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",
 ]

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",
     }
 )