diff --git a/conformance.toml b/conformance.toml index 9f9a6de..df40a64 100644 --- a/conformance.toml +++ b/conformance.toml @@ -1103,21 +1103,21 @@ note = "0113 pins the malformed case of the §6 Managed-field collision MERGE ar [proposals."0114"] status = "implemented" since = "0.17.0" -note = "Pins the two-mode Langfuse client ownership model. Mode (b): LangfuseObserver.from_credentials(public_key=..., secret_key=..., host=...) (over LangfuseSDKAdapter.from_credentials) builds an OA-owned client on a dedicated TracerProvider, so its observations no longer bind the globally-registered provider and export onto the application's OTel backend. Mode (a), a caller-supplied client, is never mutated: the caller stays responsible for isolating its tracer_provider and OA documents the remedy. secret_key is a pydantic SecretStr, masked in OA's reprs and logs with the plaintext read only at the SDK call; a blank credential is rejected at the boundary rather than falling through to the SDK's ambient LANGFUSE_* environment fallback. Fixture 157 is deferred: it needs the §6.4 provider-faithful fake and the langfuse_client construction directive, and is unit-tested meanwhile in tests/unit/test_langfuse_provider_isolation.py." +note = "Pins the two-mode Langfuse client ownership model. Mode (b): LangfuseObserver.from_credentials(public_key=..., secret_key=..., host=...) (over LangfuseSDKAdapter.from_credentials) builds an OA-owned client on a dedicated TracerProvider, so its observations no longer bind the globally-registered provider and export onto the application's OTel backend. Mode (a), a caller-supplied client, is never mutated: the caller stays responsible for isolating its tracer_provider and OA documents the remedy. secret_key is a pydantic SecretStr, masked in OA's reprs and logs with the plaintext read only at the SDK call; a blank credential is rejected at the boundary rather than falling through to the SDK's ambient LANGFUSE_* environment fallback. Fixture 157 runs: its mode-(a) case drives a protocol-shaped provider-faithful fake and its mode-(b) case drives a REAL Langfuse client with its egress removed, so the construct-from-credentials path under test is the shipped one." # Spec v0.109.0 (proposal 0115). Conformance primitives for Langfuse # provider isolation (conformance-adapter §5.5 / §6.4). [proposals."0115"] status = "partial" 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 Langfuse fake (records observation content AND emits it as OTel spans through its bound TracerProvider, so a leak to a shared provider is catchable) and the langfuse_client construction directive are NOT yet built, so fixtures 157 / 158 stay deferred. The four leak assertions are declared in the expectations model so those fixtures parse 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 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." # 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 158 is deferred on the same §6.4 machinery as 157." +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)." # 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 new file mode 100644 index 0000000..b5bdfda --- /dev/null +++ b/tests/conformance/harness/langfuse_provider_fake.py @@ -0,0 +1,264 @@ +"""Provider-faithful Langfuse client fakes for the conformance harness.""" + +# Spec basis: conformance-adapter §6.4 (proposal 0115, extended by 0116 / 0117 / +# 0118). The bundled InMemoryLangfuseClient records observation content but has no +# provider at all, so a leak onto a shared TracerProvider was not expressible and +# the 0114 isolation obligations shipped unfixtured. +# +# These fakes record content as the in-memory client does AND emit each +# observation as an OTel span through their bound TracerProvider, the way a real +# Langfuse v4 client does. An exporter installed on that provider therefore +# observes any observation that reaches it, which is what the leak assertions +# read: content assertions read the recorded side, leak assertions read the +# provider side. +# +# The emitted spans carry the real `langfuse.*` attribute namespace rather than a +# harness-invented summary. That is deliberate. A double that stamps its own +# verdict, read by an assertion, tests the double's opinion instead of the +# property -- any channel the double's predicate missed would be invisible, and +# the assertion could not be cross-checked against a real client. +# +# SCOPE. This module serves `langfuse_client: {mode: supplied}` only, where the +# CALLER hands openarmature an object satisfying its LangfuseClient protocol. +# +# `mode: credentials` is NOT faked. There openarmature constructs the client +# itself and wraps it in LangfuseSDKAdapter, which drives the real SDK's private +# surface (start_observation, _otel_tracer, _create_remote_parent_span, +# _get_otel_trace_id, and the real langfuse._client.span classes). Faking that is +# chasing a private API, and the provider observations would not land in a +# recorded side anyway, since the adapter builds real SDK span objects for them. +# That mode uses a REAL client with its egress neutralised -- see +# `langfuse_real_client.py`. + +from __future__ import annotations + +import json +from typing import Any + +from opentelemetry import trace as otel_trace + +from openarmature.observability.langfuse.client import ( + InMemoryLangfuseClient, + LangfuseObservation, +) + +# Attribute names a Langfuse v4 client writes on the spans it exports, verified +# against langfuse/_client/attributes.py. The observation NAME is the OTel span +# name, not an attribute. The leak assertions select openarmature's Langfuse +# observations out of an exporter by the `langfuse.observation.` prefix, mirroring +# how the OTel-side check selects openarmature spans. +OBSERVATION_TYPE_ATTR = "langfuse.observation.type" +OBSERVATION_INPUT_ATTR = "langfuse.observation.input" +OBSERVATION_OUTPUT_ATTR = "langfuse.observation.output" +OBSERVATION_METADATA_ATTR = "langfuse.observation.metadata" +OBSERVATION_LEVEL_ATTR = "langfuse.observation.level" +OBSERVATION_STATUS_MESSAGE_ATTR = "langfuse.observation.status_message" +TRACE_INPUT_ATTR = "langfuse.trace.input" +TRACE_OUTPUT_ATTR = "langfuse.trace.output" + +_OBSERVATION_PREFIX = "langfuse.observation." + + +def _encode(value: Any) -> str: + """Render a payload the way the SDK does, as a JSON string attribute.""" + if isinstance(value, str): + return value + try: + return json.dumps(value, default=str, sort_keys=True) + except (TypeError, ValueError): # pragma: no cover - defensive + return repr(value) + + +class _ProviderBinding: + """The shared provider side: which TracerProvider, and how a span is emitted. + + Composed rather than inherited so it can sit under both client shapes without + entangling with the dataclass base one of them already has. + """ + + # `tracer_provider=None` means NONE SUPPLIED, which the SDK resolves to the + # globally-registered provider; that is the priming construction §6.4 + # describes and is a different state from tracing being off. Collapsing the + # two onto one argument would make a primed client both unable to leak and + # classified isolated, so 158's raise arm could never fire. + def __init__(self, tracer_provider: Any = None, *, tracing_enabled: bool = True) -> None: + resolved = None + if tracing_enabled: + resolved = tracer_provider if tracer_provider is not None else otel_trace.get_tracer_provider() + self.tracer_provider = resolved + self.tracing_enabled = tracing_enabled + self.tracer = resolved.get_tracer("langfuse") if resolved is not None else None + # Trace-level input / output ride the trace's first exported span, as the + # SDK writes them onto the span that opens the trace. + self.pending_trace_attrs: dict[str, dict[str, str]] = {} + + @property + def active(self) -> bool: + return self.tracer is not None + + def note_trace_payload(self, trace_id: str, *, input: Any = None, output: Any = None) -> None: + pending = self.pending_trace_attrs.setdefault(trace_id, {}) + if input is not None: + pending[TRACE_INPUT_ATTR] = _encode(input) + if output is not None: + pending[TRACE_OUTPUT_ATTR] = _encode(output) + + def emit(self, trace_id: str, observation: LangfuseObservation) -> None: + """Emit one recorded observation as an OTel span on the bound provider.""" + if self.tracer is None: + return + span = self.tracer.start_span(observation.name or f"langfuse.{observation.type}") + span.set_attribute(OBSERVATION_TYPE_ATTR, observation.type) + if observation.input is not None: + span.set_attribute(OBSERVATION_INPUT_ATTR, _encode(observation.input)) + 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)) + 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) + for name, value in self.pending_trace_attrs.pop(trace_id, {}).items(): + span.set_attribute(name, value) + span.end() + + def drain_orphan_trace_payloads(self) -> None: + """Emit trace payload that never rode an observation. + + A state-channel leak on an invocation that produced no observations would + otherwise never reach the provider, and the assertion would read a clean + exporter. + """ + for trace_id, attrs in list(self.pending_trace_attrs.items()): + if attrs and self.tracer is not None: + span = self.tracer.start_span("langfuse.trace") + # Carries the observation-namespace type too, or the leak selector + # (which matches on `langfuse.observation.`) drops the very span + # this method exists to surface, and a state-channel leak on an + # invocation with no observations reads as a clean exporter. + span.set_attribute(OBSERVATION_TYPE_ATTR, "trace") + for name, value in attrs.items(): + span.set_attribute(name, value) + span.end() + self.pending_trace_attrs.pop(trace_id, None) + + def flush_provider(self, timeout_ms: int) -> bool: + provider: Any = self.tracer_provider + if provider is None or not hasattr(provider, "force_flush"): + return True + return bool(provider.force_flush(timeout_ms)) + + +class _Resources: + """Stands in for the SDK's resource manager, exposing the one attribute + openarmature reads to establish a client's provider binding.""" + + def __init__(self, tracer_provider: Any) -> None: + self.tracer_provider = tracer_provider + + +# --------------------------------------------------------------------------- +# Protocol shape -- what a caller supplies (mode: supplied) +# --------------------------------------------------------------------------- + + +class _ExportingHandle: + """Wraps a recorded-observation handle and exports the span at `end()`.""" + + # The real client keeps its span open for the observation's lifetime and + # exports at end, so anything written by a later `update()` / `end(...)` is on + # the exported span. Exporting at CREATION instead would classify a payload + # that arrives late as absent -- `status_message`, for one, is set from + # harvested exception text after the fact. + + def __init__(self, inner: Any, binding: _ProviderBinding, trace_id: str) -> None: + self._inner = inner + self._binding = binding + self._trace_id = trace_id + + @property + def id(self) -> str: + return str(self._inner.id) + + @property + def observation(self) -> LangfuseObservation: + obs: LangfuseObservation = self._inner.observation + return obs + + def update(self, **fields: Any) -> None: + self._inner.update(**fields) + + def end(self, **kwargs: Any) -> None: + self._inner.end(**kwargs) + self._binding.emit(self._trace_id, self.observation) + + +class ProviderFaithfulLangfuseClient(InMemoryLangfuseClient): + """A protocol-shaped client that also exports through a TracerProvider.""" + + def __init__(self, tracer_provider: Any = None, *, tracing_enabled: bool = True) -> None: + super().__init__() + self._binding = _ProviderBinding(tracer_provider, tracing_enabled=tracing_enabled) + self.tracer_provider = self._binding.tracer_provider + self._tracing_enabled = tracing_enabled + self._resources = _Resources(self._binding.tracer_provider) + + def _wrap(self, handle: Any, trace_id: str) -> Any: + return _ExportingHandle(handle, self._binding, trace_id) if self._binding.active else handle + + def span(self, **kwargs: Any) -> Any: + return self._wrap(super().span(**kwargs), kwargs["trace_id"]) + + def generation(self, **kwargs: Any) -> Any: + return self._wrap(super().generation(**kwargs), kwargs["trace_id"]) + + def tool(self, **kwargs: Any) -> Any: + return self._wrap(super().tool(**kwargs), kwargs["trace_id"]) + + def embedding(self, **kwargs: Any) -> Any: + return self._wrap(super().embedding(**kwargs), kwargs["trace_id"]) + + def retriever(self, **kwargs: Any) -> Any: + return self._wrap(super().retriever(**kwargs), kwargs["trace_id"]) + + def update_trace(self, **kwargs: Any) -> Any: + result = super().update_trace(**kwargs) + self._binding.note_trace_payload(kwargs["id"], input=kwargs.get("input"), output=kwargs.get("output")) + return result + + def force_flush(self, timeout_ms: int = 5000) -> bool: + """Flush the recorded side AND the bound provider.""" + # The inherited implementation returns True without flushing anything, + # documented as "no outbound buffer" -- false for this subclass. A harness + # on a BatchSpanProcessor would otherwise read an empty exporter and every + # leak assertion would pass having observed nothing. + recorded = bool(super().force_flush(timeout_ms)) + self._binding.drain_orphan_trace_payloads() + for trace in self.traces.values(): + for observation in trace.observations: + if not observation.ended: + self._binding.emit(trace.id, observation) + observation.ended = True + return recorded and self._binding.flush_provider(timeout_ms) + + +def langfuse_observation_spans(spans: Any) -> list[Any]: + """The openarmature Langfuse observations among an exporter's spans.""" + selected: list[Any] = [] + for span in spans: + attributes: dict[str, Any] = dict(span.attributes or {}) + if any(name.startswith(_OBSERVATION_PREFIX) for name in attributes): + selected.append(span) + return selected + + +__all__ = [ + "OBSERVATION_INPUT_ATTR", + "OBSERVATION_METADATA_ATTR", + "OBSERVATION_OUTPUT_ATTR", + "OBSERVATION_TYPE_ATTR", + "TRACE_INPUT_ATTR", + "TRACE_OUTPUT_ATTR", + "ProviderFaithfulLangfuseClient", + "langfuse_observation_spans", +] diff --git a/tests/conformance/harness/langfuse_real_client.py b/tests/conformance/harness/langfuse_real_client.py new file mode 100644 index 0000000..1147962 --- /dev/null +++ b/tests/conformance/harness/langfuse_real_client.py @@ -0,0 +1,109 @@ +"""A real Langfuse client for the isolation fixtures, with its egress removed.""" + +# Spec basis: conformance-adapter §6.4 (proposals 0115 / 0116). §6.4 calls for a +# provider-faithful double, and for `mode: supplied` that is the protocol-shaped +# fake in `langfuse_provider_fake.py`. For `mode: credentials` the double is the +# REAL SDK client, because openarmature constructs it itself and wraps it in +# LangfuseSDKAdapter, which drives the SDK's private surface: start_observation, +# _otel_tracer, _create_remote_parent_span, _get_otel_trace_id, and the real +# langfuse._client.span classes for back-dated observations. Faking that surface +# is chasing a private API, and it was tried: each fake internal satisfied +# uncovered another, and provider observations would not have landed in a recorded +# side regardless, because the adapter builds real SDK span objects for them. +# +# So everything stays real except the network. The SDK attaches its own +# LangfuseSpanProcessor -- a batching OTLP exporter -- to whatever TracerProvider +# it is handed, which in a conformance run means retries against an unreachable +# API and background threads. Replacing that one class leaves the client, the +# adapter path, the provider binding, and the per-credential singleton genuinely +# real, and the fixture's own exporter still receives every span because it is a +# separate processor on the same provider. +# +# This is strictly more faithful than a fake, and it is the reason the fixtures +# mean anything: the private-API coupling they exercise is coupling openarmature +# already ships (see `tests/unit/test_langfuse_sdk_internals.py`, which enumerates +# it), not coupling the harness invented. + +from __future__ import annotations + +import contextlib +import os +from collections.abc import Iterator +from typing import Any + +from opentelemetry.sdk.trace import SpanProcessor + +# Credentials the harness constructs with. Any non-blank pair works: nothing +# authenticates, and `from_credentials` rejects blanks at the boundary precisely +# so a blank cannot fall through to the SDK's ambient LANGFUSE_* environment +# credentials. +CONFORMANCE_PUBLIC_KEY = "pk-lf-conformance" +CONFORMANCE_SECRET_KEY = "sk-lf-conformance" + +# 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" + + +class _NoEgressSpanProcessor(SpanProcessor): + """Stands in for LangfuseSpanProcessor, dropping everything.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + +@contextlib.contextmanager +def langfuse_sdk_without_egress() -> Iterator[None]: + """Run with the SDK's exporting span processor replaced, and its cache clean. + + The per-credential client cache is process-global and outlives a case, so it + is cleared on both sides: a stale entry would hand the next case a client + bound to the previous case's provider and silently decide its isolation + verdict. + """ + import unittest.mock + + from langfuse._client import resource_manager + + # Reaching into a private module, deliberately and in one place. The shipped + # adapter already depends on this SDK's internals (enumerated and guarded in + # tests/unit/test_langfuse_sdk_internals.py); the harness adds no new coupling + # beyond swapping the one class that would otherwise talk to the network. + prior_processor = resource_manager.LangfuseSpanProcessor # pyright: ignore[reportPrivateImportUsage] + prior_instances = dict(resource_manager.LangfuseResourceManager._instances) + resource_manager.LangfuseSpanProcessor = _NoEgressSpanProcessor # type: ignore[assignment, misc] + resource_manager.LangfuseResourceManager._instances.clear() + # The SDK resolves its base URL from LANGFUSE_BASE_URL / LANGFUSE_HOST BEFORE + # falling back to the host argument, so an ambient variable outranks the + # unreachable host below and would point the API, score-ingestion and media + # clients at a real project. We keep those variables set for live validation, + # so this is a plausible local state rather than a hypothetical. + cleared = {k: v for k, v in os.environ.items() if k.startswith("LANGFUSE_")} + try: + with unittest.mock.patch.dict(os.environ, {}, clear=False): + for key in cleared: + os.environ.pop(key, None) + yield + finally: + os.environ.update(cleared) + # Shut every client this block created before dropping it. Each resource + # manager starts consumer threads, holds an httpx pool and registers an + # atexit handler; clearing the cache alone orphans all of that. + for manager in list(resource_manager.LangfuseResourceManager._instances.values()): + with contextlib.suppress(Exception): + manager.shutdown() + resource_manager.LangfuseResourceManager._instances.clear() + resource_manager.LangfuseResourceManager._instances.update(prior_instances) + resource_manager.LangfuseSpanProcessor = prior_processor # type: ignore[assignment, misc] + + +__all__ = [ + "CONFORMANCE_HOST", + "CONFORMANCE_PUBLIC_KEY", + "CONFORMANCE_SECRET_KEY", + "langfuse_sdk_without_egress", +] diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 0e8ff03..766c6b1 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -456,11 +456,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # langfuse_client construction directive, expected_construction_error, and # the four payload-scoped leak assertions. 158 additionally gates arms on the # requires_capability audience gate. - "157-langfuse-provider-isolation": ( - "needs the conformance-adapter 6.4 provider-faithful Langfuse fake and the " - "langfuse_client construction directive; src side is unit-tested meanwhile in " - "tests/unit/test_langfuse_provider_isolation.py" - ), "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 " @@ -519,6 +514,9 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # under `disable_provider_payload`. Its langfuse_trace shape lives in the # sibling harness, like 123 and 130. "159-langfuse-llm-failure-error-message", + # 157 (proposals 0114 / 0115): provider isolation, driven by the dedicated + # runner in the sibling harness. + "157-langfuse-provider-isolation", } ) diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py index 6600074..f688a47 100644 --- a/tests/conformance/test_observability_langfuse.py +++ b/tests/conformance/test_observability_langfuse.py @@ -19,6 +19,7 @@ import json from collections.abc import Callable, Mapping, Sequence from contextvars import ContextVar +from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any, cast @@ -26,6 +27,7 @@ import httpx import pytest import yaml +from pydantic import SecretStr from openarmature.graph import END, BranchSpec, ExplicitMapping, GraphBuilder from openarmature.llm import OpenAIProvider @@ -51,6 +53,19 @@ capability_skip_reason, report_recognized_skip, ) +from .harness.langfuse_provider_fake import ( + OBSERVATION_OUTPUT_ATTR, + OBSERVATION_TYPE_ATTR, + ProviderFaithfulLangfuseClient, + langfuse_observation_spans, +) +from .harness.langfuse_real_client import ( + CONFORMANCE_HOST, + CONFORMANCE_PUBLIC_KEY, + CONFORMANCE_SECRET_KEY, + langfuse_sdk_without_egress, +) +from .test_observability import _reset_otel_global_tracer_provider CONFORMANCE_DIR = ( Path(__file__).resolve().parents[2] / "openarmature-spec" / "spec" / "observability" / "conformance" @@ -154,6 +169,12 @@ # non-vacuous: flag on asserts the message absent with the category # retained, flag off asserts it present. "159-langfuse-llm-failure-error-message", + # 157 (proposals 0114 / 0115): the two provider-isolation MUSTs. Driven by + # a dedicated runner: it needs the caller's global TracerProvider, a second + # observer on openarmature's own private provider, and for mode + # `credentials` a REAL Langfuse client with its egress removed, so the + # adapter path under test is the shipped one. + "157-langfuse-provider-isolation", } ) @@ -407,7 +428,27 @@ def _fixture_id(path: Path) -> str: def _load(path: Path) -> dict[str, Any]: - return cast("dict[str, Any]", yaml.safe_load(path.read_text())) + spec = cast("dict[str, Any]", yaml.safe_load(path.read_text())) + for entry in cast("list[Any]", spec.get("cases") or [spec]): + if isinstance(entry, dict): + _normalize_observer_config(cast("dict[str, Any]", entry)) + return spec + + +def _normalize_observer_config(case: dict[str, Any]) -> None: + """Fold the `langfuse_observer_config` spelling onto `langfuse_observer`.""" + # Some fixtures spell the observer block `langfuse_observer_config` (059) and + # others `langfuse_observer`. Only the second was ever read, so 059's two + # cases -- which differ ONLY by this key -- both ran at the default and its + # state-payload-ENABLED case was a duplicate of the default-config one that + # had never exercised the path it names. Normalising here is the same move + # already made for the fixture-level subgraph folds. + config = case.pop("langfuse_observer_config", None) + if config is None: + return + merged = dict(cast("dict[str, Any]", case.get("langfuse_observer") or {})) + merged.update(cast("dict[str, Any]", config)) + case["langfuse_observer"] = merged # --------------------------------------------------------------------------- @@ -495,6 +536,28 @@ 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: + 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 = 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 + ran_157 += 1 + _reject_unknown_directives(fixture_stem, case_name, case, handled_expected=_EXPECTED_157) + _reject_unimplemented_assertions( + fixture_stem, case_name, cast("Mapping[str, Any]", case.get("expected") or {}) + ) + try: + await _run_langfuse_157(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + assert_some_case_ran(fixture_stem, ran_157, excluded_157) + return if fixture_stem == _FIXTURE_134: # 134's nested double fan-out + `calls_llm_from_wrapper` orphan primitive # are not modeled by the generic `_run_case` topology path; each case is @@ -512,6 +575,7 @@ async def test_langfuse_fixture(fixture_path: Path) -> None: ran_134 += 1 # 134 bypasses _run_case, where the guard otherwise lives, so it is # the one dispatch path that has to call it itself. + _reject_unknown_directives(fixture_stem, case_name, case, handled_expected=_EXPECTED_134) _reject_unimplemented_assertions( fixture_stem, case_name, cast("Mapping[str, Any]", case.get("expected") or {}) ) @@ -1066,6 +1130,12 @@ async def _run_case(case: Mapping[str, Any], *, fixture_stem: str | None = None) # funnel through here, so wiring it at the call sites instead would leave a # future dispatch path free to skip it silently. `_run_langfuse_134` bypasses # this function entirely and so carries its own call. + _reject_unknown_directives( + fixture_stem or "", + cast("str", case.get("name") or ""), + case, + handled_expected=_EXPECTED_DIRECTIVES, + ) _reject_unimplemented_assertions( fixture_stem or "", cast("str", case.get("name") or ""), @@ -2373,17 +2443,177 @@ def _assert_observation_tree( # 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. -_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS = frozenset( +# The identity half, implemented by `_assert_isolation_expectations`. +_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", } ) +# 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 +# implements is silently ignored and the case reads like coverage while asserting +# less than it says. This is the sibling of `_OBSERVATION_DIRECTIVES` one and two +# levels up: that set guards an observation entry, these guard the case and its +# expected block. +# +# A new key at a pin bump SHOULD fail here. Implementing a directive means adding +# it to the set, which makes acceptance a deliberate act rather than a default. +_CASE_DIRECTIVES = frozenset( + { + # Graph shape and identity. + "edges", + "entry", + "initial_state", + "inner_subgraphs", + "name", + "nodes", + "render_variables", + "state", + "subgraph", + "subgraphs", + # Assertions. + "expected", + "expected_error", + "invariants", + # Caller-supplied invocation dimensions. + "caller_correlation_id", + "caller_invocation_id", + "caller_metadata", + # Observer / client construction. + "caller_global_otel_active", + "langfuse_client", + "langfuse_observer", + "detached_fan_outs", + "detached_subgraphs", + # Mocks and backends. + "checkpointer", + "mock_llm", + "prompt_backend", + # Resume / multi-run. + "first_run_expected", + "first_run_expected_error", + "resume", + # Audience gate (conformance-adapter §5.5). + "requires_capability", + } +) + +# Prose the fixture carries for a reader. Kept apart from the implemented set so +# nothing inert is mistaken for something the runner acts on. +_DOCUMENTARY_CASE_KEYS = frozenset({"description"}) + +# 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", +} + +# 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( + { + "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"}) +_EXPECTED_134 = frozenset({"langfuse_trace", "invariants"}) + +# Declared by an activated fixture, NOT read by this runner. Each entry is a named +# skip with a reason rather than an indistinguishable member of the implemented +# set, because a reader takes that set as a contract. Deleting an entry without +# implementing the key makes the guard fail, which is the point. +_DEFERRED_EXPECTED_DIRECTIVES: dict[str, str] = { + "invocation_id": ( + "fixtures 035 / 036 also spell the caller-invocation-id assertion under " + "`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" + ), +} + + +def _reject_unknown_directives( + fixture_stem: str, + case_name: str, + case: Mapping[str, Any], + *, + handled_expected: frozenset[str], +) -> 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) + 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." + ) + client_cfg = case.get("langfuse_client") + if isinstance(client_cfg, dict): + unknown_client = sorted( + set(cast("dict[str, Any]", client_cfg)) + - _LANGFUSE_CLIENT_DIRECTIVES + - set(_DEFERRED_LANGFUSE_CLIENT_DIRECTIVES) + ) + assert not unknown_client, ( + f"{fixture_stem}::{case_name} declares langfuse_client sub-directives this harness " + f"does not read: {unknown_client}." + ) + expected = case.get("expected") + if not isinstance(expected, dict): + return + # `_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS` is subtracted here so the more + # specific guard owns those keys: they are KNOWN and deliberately unimplemented, + # which is a different failure from a key nobody recognises, and they carry a + # message naming what has to be built before a fixture may use them. + unknown_expected = sorted( + set(cast("dict[str, Any]", expected)) + - handled_expected + - set(_DEFERRED_EXPECTED_DIRECTIVES) + - _UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS + ) + assert not unknown_expected, ( + f"{fixture_stem}::{case_name} declares expected-block directives this runner does not " + f"read: {unknown_expected}. Same failure shape as an unimplemented observation " + f"directive; implement them, or record them in _DEFERRED_EXPECTED_DIRECTIVES with a reason." + ) + + def _reject_unimplemented_assertions(fixture_stem: str, case_name: str, expected: Mapping[str, Any]) -> None: """Fail if a case reaches for a leak assertion no comparator implements.""" unimplemented = sorted(set(expected) & _UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS) @@ -2649,3 +2879,232 @@ def _assert_string_or_placeholder(label: str, actual: str | None, expected: Any) # Unused helper kept for the SamplingConfig import to avoid a future # F401 when 024's prompt-sampling path lands. _ = SamplingConfig + + +# --------------------------------------------------------------------------- +# Fixture 157 — Langfuse provider isolation (proposals 0114 / 0115) +# --------------------------------------------------------------------------- + +_FIXTURE_157 = "157-langfuse-provider-isolation" + + +@dataclass +class _IsolationCapture: + """The provider surfaces the leak assertions read.""" + + global_exporter: Any # the caller's globally-registered provider + private_exporter: Any # openarmature's own OTel observer provider + isolated_exporter: Any | None # the provider openarmature built for Langfuse + + +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 + InMemorySpanExporter, + ) + + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return exporter + + +def _isolated_provider_for(public_key: str) -> Any: + from openarmature.observability.langfuse import adapter as langfuse_adapter # noqa: PLC0415 + + return langfuse_adapter._ISOLATED_PROVIDERS.get(public_key) + + +def _build_isolation_observer( + case: Mapping[str, Any], observer_kwargs: dict[str, Any], global_provider: Any +) -> tuple[LangfuseObserver, Any]: + """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. + observer = LangfuseObserver.from_credentials( + public_key=CONFORMANCE_PUBLIC_KEY, + secret_key=SecretStr(CONFORMANCE_SECRET_KEY), + host=CONFORMANCE_HOST, + **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. + 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" + ) + return observer, _attach_exporter(isolated) + # mode: supplied -- the CALLER owns the client and the provider it is bound to, + # and openarmature must not rebind it. `provider: global` means the caller + # built it on the globally-registered provider. + provider_spelling = cast("str", client_cfg.get("provider", "global")) + # Mapping an unmodelled spelling onto None would be actively wrong, not merely + # unsupported: the fake reads None as "none supplied" and resolves it back to + # the globally-registered provider, so a future `provider: isolated` case would + # get a client on the global provider and its leak verdict would be inverted. + assert provider_spelling == "global", ( + f"langfuse_client.provider={provider_spelling!r} is not modelled by this runner; " + f"only 'global' is. Construct the intended provider explicitly rather than letting it " + f"fall through to the caller's global one." + ) + client = ProviderFaithfulLangfuseClient(global_provider) + return LangfuseObserver(client=client, **observer_kwargs), None + + +def _assert_isolation_expectations( + expected: Mapping[str, Any], capture: _IsolationCapture, private_spans: list[Any] +) -> None: + """The §5.5 identity leak assertions.""" + global_langfuse = langfuse_observation_spans(capture.global_exporter.get_finished_spans()) + private_langfuse = langfuse_observation_spans(private_spans) + + if expected.get("no_langfuse_observations_on_global") is True: + assert not global_langfuse, ( + f"Langfuse observations reached the caller's global provider: {[s.name for s in global_langfuse]}" + ) + if expected.get("no_langfuse_observations_on_private") is True: + # Non-vacuity: openarmature's own OTel spans DO land here, so an empty + # exporter would mean nothing was recorded rather than that nothing leaked. + assert private_spans, ( + "openarmature's private provider recorded no spans at all, so this assertion " + "would hold whether or not a Langfuse observation leaked onto it" + ) + assert not private_langfuse, ( + f"Langfuse observations reached openarmature's own OTel provider: " + f"{[s.name for s in private_langfuse]}" + ) + 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_trace = expected.get("langfuse_trace") + if isinstance(expected_trace, dict): + # The fixture's OWN declared non-vacuity proof. Substituting a weaker + # "at least one observation exists" check let the node span alone satisfy + # it, so a dropped Generation would have left the case green while the + # fixture read as proving one was emitted and isolated. + _assert_isolated_observations_match(cast("dict[str, Any]", expected_trace), capture) + + +def _assert_isolated_observations_match( + expected_trace: Mapping[str, Any], capture: _IsolationCapture +) -> None: + """Check the fixture's expected observation tree against the isolated provider.""" + # A real Langfuse client has no recorded side, so the expectation is read off + # the spans it exported: name, type and output are what the fixture pins. + assert capture.isolated_exporter is not None, ( + "langfuse_trace was declared but this case has no isolated provider to read it from" + ) + spans = langfuse_observation_spans(capture.isolated_exporter.get_finished_spans()) + by_name = {s.name: dict(s.attributes or {}) for s in spans} + + def _walk(entries: list[Any]) -> None: + for entry in entries: + node = cast("dict[str, Any]", entry) + name = cast("str", node["name"]) + assert name in by_name, ( + f"expected a Langfuse observation named {name!r} on openarmature's isolated " + f"provider; saw {sorted(by_name)}" + ) + attrs = by_name[name] + assert attrs.get(OBSERVATION_TYPE_ATTR) == node["type"], ( + f"observation {name!r} type: expected {node['type']!r}, " + f"got {attrs.get(OBSERVATION_TYPE_ATTR)!r}" + ) + if "output" in node: + assert attrs.get(OBSERVATION_OUTPUT_ATTR) == node["output"], ( + f"observation {name!r} output: expected {node['output']!r}, " + f"got {attrs.get(OBSERVATION_OUTPUT_ATTR)!r}" + ) + _walk(cast("list[Any]", node.get("children") or [])) + + _walk(cast("list[Any]", expected_trace.get("observations") or [])) + + +async def _run_langfuse_157(case: Mapping[str, Any]) -> None: + from opentelemetry import trace as otel_trace # noqa: PLC0415 + from opentelemetry.sdk.trace import TracerProvider # noqa: PLC0415 + from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: PLC0415 + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( # noqa: PLC0415 + InMemorySpanExporter, + ) + + from openarmature.observability.langfuse import adapter as langfuse_adapter # noqa: PLC0415 + from openarmature.observability.otel.observer import OTelObserver # noqa: PLC0415 + + 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"]) + + llm_provider = OpenAIProvider( + base_url="http://mock-llm.test", + model=_resolve_llm_model(case), + api_key="test", + transport=_build_mock_llm_handler(cast("list[dict[str, Any]]", case["mock_llm"])), + ) + state_fields = cast("dict[str, dict[str, Any]]", case["state"]["fields"]) + state_cls = build_state_cls("LangfuseIsolationState", state_fields) + builder = GraphBuilder(state_cls) + for node_name, node_spec in cast("dict[str, Any]", case["nodes"]).items(): + builder.add_node( + node_name, + _build_node_body( + node_name=node_name, + node_spec=cast("dict[str, Any]", node_spec), + provider=llm_provider, + prompt_manager=None, + render_variables={}, + ), + ) + for edge in cast("list[dict[str, str]]", case["edges"]): + target_raw = edge["to"] + builder.add_edge(edge["from"], END if target_raw == "END" else target_raw) + builder.set_entry(cast("str", case["entry"])) + + prior_global = otel_trace.get_tracer_provider() + global_provider = TracerProvider() + global_exporter = _attach_exporter(global_provider) + private_exporter = InMemorySpanExporter() + # openarmature keys its isolated providers in a module global that outlives the + # case, so a stale entry would let the next case inherit this one's verdict. + with langfuse_sdk_without_egress(): + # Snapshotted INSIDE the block whose `finally` restores it. Clearing it + # before entering meant an exception from the context manager's own private + # import would leave the dict empty for the rest of the process, and every + # later from_credentials would build a fresh provider. + prior_isolated = dict(langfuse_adapter._ISOLATED_PROVIDERS) + langfuse_adapter._ISOLATED_PROVIDERS.clear() + try: + 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))) + + await graph.invoke(state_cls(**cast("dict[str, Any]", case.get("initial_state") or {}))) + await graph.drain() + langfuse_observer.force_flush() + + _assert_isolation_expectations( + cast("dict[str, Any]", case.get("expected") or {}), + _IsolationCapture(global_exporter, private_exporter, isolated_exporter), + list(private_exporter.get_finished_spans()), + ) + finally: + langfuse_adapter._ISOLATED_PROVIDERS.clear() + langfuse_adapter._ISOLATED_PROVIDERS.update(prior_isolated) + _reset_otel_global_tracer_provider(prior_global) diff --git a/tests/conformance/test_unimplemented_assertion_guard.py b/tests/conformance/test_unimplemented_assertion_guard.py index 934b850..88f78d3 100644 --- a/tests/conformance/test_unimplemented_assertion_guard.py +++ b/tests/conformance/test_unimplemented_assertion_guard.py @@ -37,12 +37,19 @@ def test_set_covers_every_leak_assertion_the_deferred_fixtures_use() -> None: leak_keys = {k for k in used if "langfuse_observations" in k or "payload_bearing" in k} assert leak_keys, "no leak assertions found in 157 / 158; this check is reading nothing" - unguarded = sorted(leak_keys - langfuse_runner._UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS) - assert not unguarded, ( - f"fixtures 157 / 158 assert {unguarded}, which the expectations model accepts for parsing " - f"but no comparator implements and nothing guards. Add them to " - f"_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS, or implement them." + accounted = ( + langfuse_runner._IMPLEMENTED_LEAK_ASSERTIONS | langfuse_runner._UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS ) + unaccounted = sorted(leak_keys - accounted) + assert not unaccounted, ( + f"fixtures 157 / 158 assert {unaccounted}, which the expectations model accepts for " + f"parsing but which is neither implemented nor guarded. Implement it, or add it to " + f"_UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS so an activated fixture reaching it fails loudly." + ) + overlap = sorted( + langfuse_runner._IMPLEMENTED_LEAK_ASSERTIONS & langfuse_runner._UNIMPLEMENTED_OBSERVABILITY_ASSERTIONS + ) + assert not overlap, f"{overlap} are listed as both implemented and unimplemented" def test_every_unimplemented_key_is_a_real_model_field() -> None: diff --git a/tests/unit/test_langfuse_provider_fake.py b/tests/unit/test_langfuse_provider_fake.py new file mode 100644 index 0000000..8b1ccd2 --- /dev/null +++ b/tests/unit/test_langfuse_provider_fake.py @@ -0,0 +1,230 @@ +"""The provider-faithful Langfuse fake, driven against a real TracerProvider.""" + +# Spec basis: conformance-adapter §6.4 (proposals 0115 / 0116 / 0117 / 0118). +# +# This file exists because the previous version of the fake shipped importing +# nowhere: every claim it made about the Langfuse v4 SDK went unexecuted, and an +# adversarial review found nine defects in it that no test run could have caught. +# So each of its obligations is driven here against a real SDK TracerProvider and +# in-memory exporter, and the observations it records are cross-checked against +# the spans it exports, so the two sides cannot silently disagree. + +from __future__ import annotations + +import json +from typing import Any + +import pytest +from opentelemetry import trace as otel_trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from tests.conformance.harness.langfuse_provider_fake import ( + OBSERVATION_INPUT_ATTR, + OBSERVATION_METADATA_ATTR, + OBSERVATION_OUTPUT_ATTR, + OBSERVATION_TYPE_ATTR, + TRACE_INPUT_ATTR, + ProviderFaithfulLangfuseClient, + langfuse_observation_spans, +) + + +def _provider() -> tuple[TracerProvider, InMemorySpanExporter]: + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return provider, exporter + + +def _attrs(span: Any) -> dict[str, Any]: + return dict(span.attributes or {}) + + +# -- emission --------------------------------------------------------------- + + +def test_an_observation_reaches_the_bound_provider() -> None: + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.generation(trace_id="t", name="openarmature.llm.complete").end() + + spans = langfuse_observation_spans(exporter.get_finished_spans()) + assert len(spans) == 1 + assert spans[0].name == "openarmature.llm.complete" + assert _attrs(spans[0])[OBSERVATION_TYPE_ATTR] == "generation" + + +def test_payload_written_at_end_is_on_the_exported_span() -> None: + # The load-bearing lifecycle property. A fake exporting at CREATION would + # stamp the span before this output exists and report it payload-free, so a + # real leak arriving via `end()` would read as no leak at all. + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + handle = client.generation(trace_id="t", name="gen") + handle.end(output="LEAKED MODEL OUTPUT") + + span = langfuse_observation_spans(exporter.get_finished_spans())[0] + assert _attrs(span)[OBSERVATION_OUTPUT_ATTR] == "LEAKED MODEL OUTPUT" + + +def test_payload_written_by_update_is_on_the_exported_span() -> None: + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + handle = client.span(trace_id="t", name="node") + handle.update(input={"role": "user"}, output="done") + handle.end() + + attrs = _attrs(langfuse_observation_spans(exporter.get_finished_spans())[0]) + assert json.loads(attrs[OBSERVATION_INPUT_ATTR]) == {"role": "user"} + assert attrs[OBSERVATION_OUTPUT_ATTR] == "done" + + +def test_error_message_metadata_reaches_the_span() -> None: + # The third harvested channel (0117): a failed observation's error_message + # rides observation.metadata, so it has to survive onto the exported span or + # a leak through it would be invisible to the provider-side assertions. + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.tool(trace_id="t", name="run_tool").end( + metadata={"error_type": "ValueError", "error_message": "LEAKED EXCEPTION TEXT"} + ) + + attrs = _attrs(langfuse_observation_spans(exporter.get_finished_spans())[0]) + assert "LEAKED EXCEPTION TEXT" in attrs[OBSERVATION_METADATA_ATTR] + + +def test_trace_payload_reaches_the_provider() -> None: + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.update_trace(id="t", input={"raw": "STATE PAYLOAD"}) + client.span(trace_id="t", name="node").end() + + joined = " ".join(json.dumps(_attrs(s)) for s in exporter.get_finished_spans()) + assert "STATE PAYLOAD" in joined + assert TRACE_INPUT_ATTR in joined + + +def test_trace_payload_reaches_the_provider_even_with_no_observations() -> None: + # A state-channel leak on an invocation that produced no observations would + # otherwise never be exported and the assertion would see a clean provider. + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.update_trace(id="t", output={"raw": "STATE PAYLOAD"}) + client.force_flush() + + joined = " ".join(json.dumps(_attrs(s)) for s in exporter.get_finished_spans()) + assert "STATE PAYLOAD" in joined + + +def test_an_unended_observation_still_exports_on_flush() -> None: + provider, exporter = _provider() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.generation(trace_id="t", name="never-ended", output="LEAKED") + assert langfuse_observation_spans(exporter.get_finished_spans()) == [] + + client.force_flush() + spans = langfuse_observation_spans(exporter.get_finished_spans()) + assert [s.name for s in spans] == ["never-ended"] + + +def test_force_flush_flushes_the_provider_not_only_the_recorder() -> None: + # Under a BatchSpanProcessor the exporter stays empty until the PROVIDER is + # flushed. The inherited force_flush returns True having flushed nothing, so + # a harness reading the exporter after it would find no spans and every leak + # assertion would pass having observed nothing. + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(BatchSpanProcessor(exporter)) + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.generation(trace_id="t", name="gen", output="payload").end() + + assert exporter.get_finished_spans() == () + assert client.force_flush() is True + assert len(langfuse_observation_spans(exporter.get_finished_spans())) == 1 + + +# -- binding ---------------------------------------------------------------- + + +def test_no_provider_supplied_binds_the_global_one() -> None: + # §6.4: "a plain priming construction with no provider supplied binds the + # global provider". Treating that as tracing-off instead would make the primed + # client both unable to leak and classified isolated, so 158's raise arm could + # never fire. + provider, exporter = _provider() + previous = otel_trace._TRACER_PROVIDER # type: ignore[attr-defined] # pyright: ignore[reportPrivateUsage] + otel_trace._TRACER_PROVIDER = provider # type: ignore[attr-defined] # pyright: ignore[reportPrivateUsage] + try: + client = ProviderFaithfulLangfuseClient() + assert client.tracer_provider is provider + assert client._resources.tracer_provider is provider + client.trace(id="t") + client.span(trace_id="t", name="node").end() + finally: + otel_trace._TRACER_PROVIDER = previous # type: ignore[attr-defined] # pyright: ignore[reportPrivateUsage] + assert len(langfuse_observation_spans(exporter.get_finished_spans())) == 1 + + +def test_tracing_disabled_is_a_separate_state_from_no_provider() -> None: + client = ProviderFaithfulLangfuseClient(tracing_enabled=False) + assert client.tracer_provider is None + assert client._resources.tracer_provider is None + assert client._tracing_enabled is False + client.trace(id="t") + client.span(trace_id="t", name="node").end() # records, exports nowhere + assert client.traces["t"].observations[0].name == "node" + + +def test_the_bound_provider_is_readable_where_openarmature_looks() -> None: + # adapter._classify_isolation walks client._resources.tracer_provider. If the + # fake does not expose it there, every fixture client classifies undetectable + # and the arms 157 / 158 exist to pin are unreachable. + provider, _ = _provider() + client = ProviderFaithfulLangfuseClient(provider) + resources = getattr(client, "_resources", None) + assert getattr(resources, "tracer_provider", None) is provider + + +# -- per-credential singleton ---------------------------------------------- + + +# -- span selection --------------------------------------------------------- + + +def test_selection_ignores_spans_that_are_not_langfuse_observations() -> None: + # The leak assertions filter an exporter that also carries openarmature's own + # OTel spans, so the selector must not sweep those up. + provider, exporter = _provider() + provider.get_tracer("openarmature").start_span("openarmature.node").end() + client = ProviderFaithfulLangfuseClient(provider) + client.trace(id="t") + client.span(trace_id="t", name="node").end() + + finished = exporter.get_finished_spans() + assert len(finished) == 2 + assert [s.name for s in langfuse_observation_spans(finished)] == ["node"] + + +@pytest.mark.parametrize("tracing_enabled", [True, False]) +def test_recording_is_unaffected_by_the_provider_binding(tracing_enabled: bool) -> None: + # The recorded side is what the content assertions read, and it must behave + # identically whichever provider the client is bound to. + provider, _ = _provider() + client = ProviderFaithfulLangfuseClient( + provider if tracing_enabled else None, tracing_enabled=tracing_enabled + ) + client.trace(id="t") + client.generation(trace_id="t", name="gen", output="hello").end() + observation = client.traces["t"].observations[0] + assert observation.name == "gen" + assert observation.output == "hello" diff --git a/tests/unit/test_langfuse_sdk_internals.py b/tests/unit/test_langfuse_sdk_internals.py new file mode 100644 index 0000000..95fd775 --- /dev/null +++ b/tests/unit/test_langfuse_sdk_internals.py @@ -0,0 +1,114 @@ +"""The private Langfuse SDK surface openarmature's shipped adapter depends on.""" + +# openarmature declares `langfuse>=4.6,<5`, and inside that range its shipped +# adapter reaches for private SDK symbols: the client class itself, the +# `_resources` / `_tracing_enabled` pair the isolation verdict rests on, the +# per-credential cache behind the already-cached warning, the `_otel_tracer` and +# `_create_remote_parent_span` internals behind every back-dated observation, and +# the four `langfuse._client.span` classes it constructs. A minor upgrade inside +# our own declared range could break any of them. +# +# Until now the only guard was the live-account integration test, which CI +# deselects (`addopts = ["-m", "not integration"]`), so in practice nothing +# checked them on a normal run. These tests are the CI-visible half: they do not +# need credentials or a network, and they fail loudly on a rename rather than +# letting a client quietly classify `undetectable` or an observation quietly stop +# being emitted. +# +# openarmature.org/compatibility tracks the SDK version last verified and already +# records the bound-provider read as resting on non-portable internals. This file +# is the executable counterpart to that row. + +from __future__ import annotations + +import inspect + +import pytest + +langfuse = pytest.importorskip("langfuse", reason="the langfuse extra is optional") + + +def test_the_client_class_is_where_the_adapter_binds_it() -> None: + # The adapter binds `Langfuse` at module import, which is why a harness must + # patch the ADAPTER's name to substitute a client. Anything that moves this + # breaks that substitution silently. + from openarmature.observability.langfuse import adapter + + assert adapter.Langfuse is langfuse.Langfuse + + +def test_the_bound_provider_is_readable_through_resources() -> None: + # adapter._classify_isolation walks client._resources.tracer_provider. If it + # goes missing every client classifies `undetectable`, payload silently + # switches off behind one WARNING, and conformance.toml keeps publishing + # langfuse_bound_provider_detection = true with nothing going red. + from langfuse._client.resource_manager import LangfuseResourceManager + + assert "self._resources" in inspect.getsource(langfuse.Langfuse) + assert "self.tracer_provider" in inspect.getsource(LangfuseResourceManager) + + +def test_the_tracing_enabled_flag_classify_isolation_branches_on_exists() -> None: + # _classify_isolation reads client._tracing_enabled FIRST and short-circuits to + # `isolated` when it is false. If it disappears, getattr's default of True keeps + # the code running while the short-circuit silently stops applying. + assert "self._tracing_enabled" in inspect.getsource(langfuse.Langfuse) + + +def test_the_per_credential_cache_is_where_the_adapter_looks() -> None: + # adapter._public_key_is_cached reads this to warn that a second construction + # for a credential silently discards its options. A rename would not fail; the + # warning would simply stop firing. + from langfuse._client.resource_manager import LangfuseResourceManager + + assert isinstance(LangfuseResourceManager._instances, dict) + + +@pytest.mark.parametrize( + ("symbol", "spelling"), + [("_otel_tracer", "self._otel_tracer"), ("_create_remote_parent_span", "def _create_remote_parent_span")], +) +def test_the_back_dated_observation_internals_exist(symbol: str, spelling: str) -> None: + # Every provider observation carrying a start_time goes through + # _start_back_dated_observation, which needs both. Losing either does not + # raise to the caller: the graph observer isolates observer errors, so the + # observation simply stops being emitted and a leak assertion reads clean. + # + # Checked in the source rather than with hasattr, because `_otel_tracer` is an + # INSTANCE attribute and a class-level hasattr would report it missing on a + # perfectly good SDK. + assert spelling in inspect.getsource(langfuse.Langfuse), ( + f"langfuse.Langfuse.{symbol} is gone; adapter._start_back_dated_observation depends " + f"on it, and its absence surfaces only as a swallowed observer warning" + ) + + +@pytest.mark.parametrize( + "name", ["LangfuseGeneration", "LangfuseTool", "LangfuseEmbedding", "LangfuseRetriever"] +) +def test_the_span_classes_the_adapter_constructs_exist(name: str) -> None: + span_module = pytest.importorskip("langfuse._client.span") + assert hasattr(span_module, name) + + +def test_the_installed_version_is_within_the_declared_range() -> None: + # Non-vacuity for everything above: the checks are only meaningful against a + # version we claim to support. This also surfaces drift between what is + # installed and what openarmature.org/compatibility records as verified. + import re + from importlib.metadata import version + + installed = version("langfuse") + # Regex rather than int() on the split parts: a PEP 440 two-component + # pre-release such as 4.7rc1 attaches its suffix to the MINOR, so splitting + # raises ValueError on a version that is inside our declared range. A bare + # ValueError from a guard test also reads identically to a genuine range + # violation, which is the more expensive confusion. + match = re.match(r"^(\d+)\.(\d+)", installed) + assert match is not None, ( + f"cannot parse a major.minor out of langfuse version {installed!r}; this guard cannot " + f"confirm the installed SDK is inside the declared range" + ) + major, minor = int(match.group(1)), int(match.group(2)) + assert (major, minor) >= (4, 6), f"langfuse {installed} is below the declared floor of 4.6" + assert major < 5, f"langfuse {installed} is outside the declared range (<5)" diff --git a/uv.lock b/uv.lock index 9d4f07c..810afe5 100644 --- a/uv.lock +++ b/uv.lock @@ -577,7 +577,7 @@ wheels = [ [[package]] name = "langfuse" -version = "4.7.0" +version = "4.7.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "backoff" }, @@ -589,9 +589,9 @@ dependencies = [ { name = "pydantic" }, { name = "wrapt" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/ab/80/0bd2ed8f781905d8e4c8f69e72fd06d914c907f31a3074449d20f8964c78/langfuse-4.7.0.tar.gz", hash = "sha256:7b592a251777dae44f76db7971871ff16c4c200d490cb5cd4e2f3822a1d593ec", size = 282543, upload-time = "2026-05-27T18:51:01.521Z" } +sdist = { url = "https://files.pythonhosted.org/packages/2c/74/a6f1a99893ee6d1a69439ae7eb92f8fe8806103492dc26531d5942dbd3bf/langfuse-4.7.1.tar.gz", hash = "sha256:f9e262eceedb353b191c1da1f8452d1e8ebf52297ca20e160cda0206608e3a40", size = 320620, upload-time = "2026-05-29T18:06:22.435Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/cf/13/9a2304f546c74bc2bfc4765adb1e6f6a0984ead671fa65c3edcf9d473b58/langfuse-4.7.0-py3-none-any.whl", hash = "sha256:72ca668ce0999d05c787ccbfa313b679b99b3492e268d351549488822dd40c34", size = 482420, upload-time = "2026-05-27T18:50:59.773Z" }, + { url = "https://files.pythonhosted.org/packages/9f/9a/bd3368f46b6c72ee2068b80536826b02ae86df53eff1c79941344503098f/langfuse-4.7.1-py3-none-any.whl", hash = "sha256:a4e59c81ad5e5b16a65d3849f4923ebc3ad6e67ec803ada83d50c0cb66149490", size = 562571, upload-time = "2026-05-29T18:06:20.517Z" }, ] [[package]]