From 64ed88b9b68a2ff0210359515a64b3e4589492d6 Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Tue, 1 Sep 2026 12:59:34 -0700 Subject: [PATCH] fix(langfuse): harden client lifecycle and v4 value coercion per review - hold the SDK registry lock across evict+construct; reuse the provider for a cached key (atexit pin leak); tear down only litellm-built providers; unpublish the registry entry before the blocking shutdown - rebuild litellm's TLS material (CA bundle / client cert) onto the OTLP exporter, which no longer flows through the injected httpx client - coerce and cap propagated trace attributes (v4 drops non-str and >200 chars) - claim trace root on guardrail/grounding sibling spans like the SDK does - return the exported observation id as generation_id - reject langfuse v5 prereleases in the version gate - pin langfuse==4.15.1 in Dockerfile.build_from_pip; drop dead version sniffs; warn once on ignored UPSTREAM_LANGFUSE_* vars --- .../build_from_pip/Dockerfile.build_from_pip | 2 +- litellm/integrations/langfuse/langfuse.py | 145 +++++++++------- .../langfuse/langfuse_prompt_management.py | 22 +-- litellm/integrations/langfuse/langfuse_sdk.py | 164 +++++++++++++++--- .../test_langfuse_prompt_management.py | 8 +- .../langfuse/test_langfuse_sdk.py | 58 ++++++- .../integrations/test_langfuse.py | 73 +++++++- 7 files changed, 350 insertions(+), 122 deletions(-) diff --git a/docker/build_from_pip/Dockerfile.build_from_pip b/docker/build_from_pip/Dockerfile.build_from_pip index a7758b79fe3..07a36fd8431 100644 --- a/docker/build_from_pip/Dockerfile.build_from_pip +++ b/docker/build_from_pip/Dockerfile.build_from_pip @@ -32,7 +32,7 @@ RUN uv venv --python python && \ "anthropic[vertex]==0.84.0" \ "grpcio==1.78.0" \ "prometheus-client==0.20.0" \ - "langfuse>=4.7,<5.0" \ + "langfuse==4.15.1" \ "opentelemetry-api==1.33.1" \ "opentelemetry-sdk==1.33.1" \ "opentelemetry-exporter-otlp==1.33.1" \ diff --git a/litellm/integrations/langfuse/langfuse.py b/litellm/integrations/langfuse/langfuse.py index 507f6c0fb92..c15789cf903 100644 --- a/litellm/integrations/langfuse/langfuse.py +++ b/litellm/integrations/langfuse/langfuse.py @@ -2,7 +2,7 @@ # On success, logs events to Langfuse import os import traceback -from collections.abc import Callable, Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping, Sequence from datetime import datetime from functools import lru_cache from importlib.metadata import version @@ -141,7 +141,9 @@ def raise_if_unsupported_langfuse_version(installed_version: str) -> None: `propagate_attributes` raises inside the per-request handler and the broad except there turns it into silent total data loss. """ - if Version(MINIMUM_LANGFUSE_VERSION) <= Version(installed_version) < Version(UNSUPPORTED_LANGFUSE_VERSION): + installed: Final = Version(installed_version) + # compare majors, not versions: "5.0.0rc1" sorts below "5" but is just as unsupported + if Version(MINIMUM_LANGFUSE_VERSION) <= installed and installed.major < Version(UNSUPPORTED_LANGFUSE_VERSION).major: return raise ImportError( f"\033[91mlitellm requires langfuse>={MINIMUM_LANGFUSE_VERSION},<{UNSUPPORTED_LANGFUSE_VERSION} for the " @@ -159,17 +161,37 @@ _GENERATION_ONLY_KEYS: Final = frozenset( ) +_PROPAGATED_VALUE_MAX_CHARS: Final = 200 + + +def _coerce_propagated_value(value: object) -> str | Sequence[str]: + """v4 silently drops non-string or >200-char propagated values; v2's pydantic coerced them.""" + if isinstance(value, (list, tuple)): + return [str(item)[:_PROPAGATED_VALUE_MAX_CHARS] for item in value] + return str(value)[:_PROPAGATED_VALUE_MAX_CHARS] + + def _trace_attributes_for_propagation(trace_params: Mapping[str, object]) -> Mapping[str, object]: - """Trace-level fields in v4 are propagated onto the observations, not set on a trace object.""" + """Trace-level fields in v4 are propagated onto the observations, not set on a trace object. + + Values are coerced and capped up front: the SDK drops offenders with only a + warning, and a dropped ``version`` would vanish from the generation too, + because ``_generation_attributes`` already stripped it as propagated. + """ return MappingProxyType( { - propagated: trace_params[key] + propagated: _coerce_propagated_value(trace_params[key]) for key, propagated in _PROPAGATED_TRACE_KEYS.items() if trace_params.get(key) is not None } ) +def _optional_str(value: object) -> str | None: + """v4 sets attribute values raw; a non-string version would be dropped by the server.""" + return str(value) if value is not None else None + + def _trace_public_flag(value: object) -> bool | None: """``trace_public`` reaches here as a bool from metadata or a string from a ``langfuse_*`` header.""" if value is None: @@ -290,8 +312,8 @@ class LangFuseLogger: "debug": self.langfuse_debug, "flush_interval": self.langfuse_flush_interval, # flush interval in seconds "httpx_client": self.langfuse_client, + "environment": self.langfuse_environment, } - parameters["environment"] = self.langfuse_environment self.Langfuse: Langfuse = self.safe_init_langfuse_client(parameters) # set the current langfuse project id in the environ @@ -307,6 +329,10 @@ class LangFuseLogger: verbose_logger.debug("Langfuse project id unavailable, alerting links will omit it") if os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY") is not None: + verbose_logger.warning( + "UPSTREAM_LANGFUSE_* is no longer supported: the langfuse callback moved to SDK v4, " + "which has no second ingestion client. The values are ignored." + ) self.upstream_langfuse_secret_key = os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY") self.upstream_langfuse_public_key = os.getenv("UPSTREAM_LANGFUSE_PUBLIC_KEY") self.upstream_langfuse_host = os.getenv("UPSTREAM_LANGFUSE_HOST") @@ -321,33 +347,20 @@ class LangFuseLogger: - Langfuse initializes 1 thread everytime a client is initialized. - We've had an incident in the past where we reached 100% cpu utilization because Langfuse was initialized several times. """ - from langfuse import Langfuse - if litellm.initialized_langfuse_clients >= MAX_LANGFUSE_INITIALIZED_CLIENTS: raise Exception( f"Max langfuse clients reached: {litellm.initialized_langfuse_clients} is greater than {MAX_LANGFUSE_INITIALIZED_CLIENTS}" ) - from litellm.integrations.langfuse.langfuse_sdk import ( - DiscardingSpanExporter, - build_isolated_tracer_provider, - evict_stale_langfuse_resources, - register_langfuse_client, - ) + from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client - evict_stale_langfuse_resources( - public_key=parameters.get("public_key"), - secret_key=parameters.get("secret_key"), - base_url=parameters.get("base_url"), + environment_param: Final = cast(str | None, parameters.get("environment")) # cast-ok: untyped dict + release_param: Final = cast(str | None, parameters.get("release")) # cast-ok: untyped dict + langfuse_client: Final = acquire_langfuse_client( + parameters=parameters, + environment=environment_param, + release=release_param, + mock_mode=self.is_mock_mode, ) - langfuse_client: Final = Langfuse( - **parameters, - tracer_provider=build_isolated_tracer_provider( - environment=parameters.get("environment"), - release=parameters.get("release"), - ), - span_exporter=DiscardingSpanExporter() if self.is_mock_mode else None, - ) - register_langfuse_client(langfuse_client) litellm.initialized_langfuse_clients += 1 verbose_logger.debug("Created langfuse client number %s", litellm.initialized_langfuse_clients) return langfuse_client @@ -446,8 +459,6 @@ class LangFuseLogger: status_message=status_message, ) verbose_logger.debug("OUTPUT IN LANGFUSE: %s; original: %s", output, response_obj) - trace_id = None - generation_id = None trace_id, generation_id = self._log_langfuse_v2( user_id=user_id, metadata=metadata, @@ -579,11 +590,7 @@ class LangFuseLogger: StandardLoggingPayload | None, kwargs.get("standard_logging_object", None), ) - tags = ( - self._get_langfuse_tags(standard_logging_object=standard_logging_object) - if self._supports_tags() - else [] - ) + tags = self._get_langfuse_tags(standard_logging_object=standard_logging_object) allowlisted_metadata: Final[StandardLoggingMetadata | Mapping[str, object]] = ( standard_logging_object["metadata"] if standard_logging_object is not None else _NO_METADATA @@ -739,17 +746,16 @@ class LangFuseLogger: ("api_base", api_base, bool(api_base)), ("vertex_location", vertex_location, bool(vertex_location)), ("aws_region_name", aws_region_name, bool(aws_region_name)), - ("cache_hit", kwargs.get("cache_hit") or False, self._supports_tags() and "cache_hit" in kwargs), + ("cache_hit", kwargs.get("cache_hit") or False, "cache_hit" in kwargs), ) enrichments: Final[Mapping[str, object]] = { key: value for key, value, include in candidate_enrichments if include } - if self._supports_tags(): - if "cache_hit" in kwargs and kwargs["cache_hit"] is None: - kwargs["cache_hit"] = False # rebind-ok: pre-existing normalization other integrations rely on - if existing_trace_id is None: - trace_params.update({"tags": tags}) + if "cache_hit" in kwargs and kwargs["cache_hit"] is None: + kwargs["cache_hit"] = False # rebind-ok: pre-existing normalization other integrations rely on + if existing_trace_id is None: + trace_params.update({"tags": tags}) proxy_server_request: Final = litellm_params.get("proxy_server_request", None) if proxy_server_request: @@ -837,20 +843,19 @@ class LangFuseLogger: **enrichments, }, "level": level, - "version": clean_metadata.pop("version", None), + "version": _optional_str(clean_metadata.pop("version", None)), } parent_observation_id: Final = metadata.get("parent_observation_id", None) if parent_observation_id is not None: generation_params["parent_observation_id"] = parent_observation_id - if self._supports_prompt(): - generation_params = _add_prompt_to_generation_params( - generation_params=generation_params, - clean_metadata=clean_metadata, - prompt_management_metadata=prompt_management_metadata, - langfuse_client=self.Langfuse, - ) + generation_params = _add_prompt_to_generation_params( + generation_params=generation_params, + clean_metadata=clean_metadata, + prompt_management_metadata=prompt_management_metadata, + langfuse_client=self.Langfuse, + ) if masked_output is not None and isinstance(masked_output, str) and level == "ERROR": generation_params["status_message"] = masked_output @@ -876,14 +881,18 @@ class LangFuseLogger: parent_observation_id=resolve_observation_id(parent_observation_id), # pyright: ignore[reportArgumentType] # metadata value, str or None at runtime ) log_provider_specific_information_as_span( - client=self.Langfuse, context=trace_context, enrichments=enrichments + client=self.Langfuse, + context=trace_context, + enrichments=enrichments, + claim_trace_root=claim_trace_root, ) self._log_guardrail_information_as_span( client=self.Langfuse, context=trace_context, standard_logging_object=standard_logging_object, + claim_trace_root=claim_trace_root, ) - start_generation( + generation: Final = start_generation( client=self.Langfuse, context=trace_context, name=generation_params["name"], # pyright: ignore[reportArgumentType] # always the str set a few lines up @@ -892,10 +901,13 @@ class LangFuseLogger: release=trace_params.get("release"), public=_trace_public_flag(trace_params.get("public")), attributes=_generation_attributes(generation_params, propagated=propagated_trace_attributes), - ).end(end_time=to_unix_nanos(end_time)) + ) + generation.end(end_time=to_unix_nanos(end_time)) - # log_event_on_langfuse tuple-unpacks this and re-wraps it in the dict callers cache - return resolved_trace_id, generation_id + # log_event_on_langfuse tuple-unpacks this and re-wraps it in the dict callers cache. + # The wrapper's id is the exported observation id; the pre-computed generation_id would + # name nothing in langfuse, because v4 derives observation ids from the OTel span. + return resolved_trace_id, generation.id except Exception: verbose_logger.error("Langfuse Layer Error - %s", traceback.format_exc()) return None, None @@ -969,14 +981,6 @@ class LangFuseLogger: tags.append(f"cache_key:{_cache_key}") return tags - def _supports_tags(self): - """Check if current langfuse version supports tags""" - return Version(self.langfuse_sdk_version) >= Version("2.6.3") - - def _supports_prompt(self): - """Check if current langfuse version supports prompt""" - return Version(self.langfuse_sdk_version) >= Version("2.7.3") - @staticmethod def _apply_masking_function(data: object, masking_function: Callable[[object], object]) -> object: """ @@ -1044,6 +1048,7 @@ class LangFuseLogger: client: "Langfuse", context: "Context", standard_logging_object: StandardLoggingPayload | None, + claim_trace_root: bool, ): """ Log guardrail information as a span @@ -1079,6 +1084,7 @@ class LangFuseLogger: context=context, name="guardrail", start_time=guardrail_entry.get("start_time", None), + claim_trace_root=claim_trace_root, attributes={ # mutable-ok: langfuse serializes this payload, a proxy is not json-encodable "input": guardrail_entry.get("guardrail_request", None), "output": guardrail_entry.get("guardrail_response", None), @@ -1172,6 +1178,7 @@ def log_provider_specific_information_as_span( client: "Langfuse", context: "Context", enrichments: Mapping[str, Any], + claim_trace_root: bool, ): """ Logs provider-specific information as spans. @@ -1195,19 +1202,30 @@ def log_provider_specific_information_as_span( for elem in vertex_ai_grounding_metadata: if isinstance(elem, dict): for key, value in elem.items(): - _end_grounding_span(client=client, context=context, name=key, value=value) + _end_grounding_span( + client=client, context=context, name=key, value=value, claim_trace_root=claim_trace_root + ) else: - _end_grounding_span(client=client, context=context, name="vertex_ai_grounding_metadata", value=elem) + _end_grounding_span( + client=client, + context=context, + name="vertex_ai_grounding_metadata", + value=elem, + claim_trace_root=claim_trace_root, + ) else: _end_grounding_span( client=client, context=context, name="vertex_ai_grounding_metadata", value=vertex_ai_grounding_metadata, + claim_trace_root=claim_trace_root, ) -def _end_grounding_span(*, client: "Langfuse", context: "Context", name: str, value: object) -> None: +def _end_grounding_span( + *, client: "Langfuse", context: "Context", name: str, value: object, claim_trace_root: bool +) -> None: from litellm.integrations.langfuse.langfuse_sdk import start_child_span start_child_span( @@ -1215,6 +1233,7 @@ def _end_grounding_span(*, client: "Langfuse", context: "Context", name: str, va context=context, name=name, start_time=None, + claim_trace_root=claim_trace_root, attributes={"input": value}, # mutable-ok: langfuse serializes this payload ).end() diff --git a/litellm/integrations/langfuse/langfuse_prompt_management.py b/litellm/integrations/langfuse/langfuse_prompt_management.py index 7dadffa353d..48ba9d45e36 100644 --- a/litellm/integrations/langfuse/langfuse_prompt_management.py +++ b/litellm/integrations/langfuse/langfuse_prompt_management.py @@ -68,7 +68,9 @@ def langfuse_client_init( Exception: If langfuse package is not installed """ try: - from langfuse import Langfuse + from langfuse import ( + Langfuse, # noqa: F401 # the import is the install probe; construction moved to acquire_langfuse_client + ) except Exception as e: raise Exception( f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\n\033[0m" @@ -118,20 +120,14 @@ def langfuse_client_init( parameters["environment"] = LangFuseLogger.resolve_deployment_environment() - from .langfuse_sdk import ( - DiscardingSpanExporter, - build_isolated_tracer_provider, - evict_stale_langfuse_resources, - register_langfuse_client, - ) + from .langfuse_sdk import acquire_langfuse_client - evict_stale_langfuse_resources(public_key=public_key, secret_key=secret_key, base_url=langfuse_host) - client: Final = Langfuse( - **parameters, # pyright: ignore[reportArgumentType] # kwargs-ok: dict mirrors the typed ctor, values resolved above - tracer_provider=build_isolated_tracer_provider(environment=parameters["environment"], release=langfuse_release), - span_exporter=DiscardingSpanExporter() if is_mock_mode else None, + client: Final = acquire_langfuse_client( + parameters=parameters, + environment=parameters["environment"], + release=langfuse_release, + mock_mode=is_mock_mode, ) - register_langfuse_client(client) return client diff --git a/litellm/integrations/langfuse/langfuse_sdk.py b/litellm/integrations/langfuse/langfuse_sdk.py index 3cad3b14547..63a1e821604 100644 --- a/litellm/integrations/langfuse/langfuse_sdk.py +++ b/litellm/integrations/langfuse/langfuse_sdk.py @@ -1,7 +1,9 @@ from __future__ import annotations +import os import re import threading +from base64 import b64encode from collections.abc import Mapping from datetime import datetime from hashlib import sha256 @@ -22,6 +24,7 @@ __all__ = ( "PUBLIC_ATTRIBUTE", "RELEASE_ATTRIBUTE", "DiscardingSpanExporter", + "acquire_langfuse_client", "build_isolated_tracer_provider", "evict_stale_langfuse_resources", "open_trace_context", @@ -135,17 +138,28 @@ def start_child_span( context: Context, name: str, start_time: datetime | float | None, + claim_trace_root: bool, attributes: Mapping[str, object], ) -> LangfuseSpan: - """Create a sibling observation inside the same trace, keeping its own window.""" + """Create a sibling observation inside the same trace, keeping its own window. + + When the shared parent is the fabricated remote span, every observation must + claim trace root itself — the SDK's own remote-parent paths stamp each span — + or it exports with a parent id that is never exported. + """ otel_span: Final = client._otel_tracer.start_span( # pyright: ignore[reportPrivateUsage] # only route to a historical start time name=name, context=context, start_time=to_unix_nanos(start_time) ) + if claim_trace_root: + otel_span.set_attribute(AS_ROOT_ATTRIBUTE, True) return LangfuseSpan(otel_span=otel_span, langfuse_client=client, **attributes) # pyright: ignore[reportArgumentType] # kwargs-ok: callback-built params, v2 accepted the same shapes _ENVIRONMENT_ATTRIBUTE: Final = "langfuse.environment" -_RELEASE_ATTRIBUTE: Final = "langfuse.release" + +# providers litellm itself constructed; a bundle adopted from user code may hold the +# process-global provider, which litellm must never shut down. +_litellm_built_providers: Final[WeakSet] = WeakSet() def build_isolated_tracer_provider(*, environment: str | None, release: str | None) -> TracerProvider: @@ -162,11 +176,13 @@ def build_isolated_tracer_provider(*, environment: str | None, release: str | No attributes: Final = MappingProxyType( { key: value - for key, value in ((_ENVIRONMENT_ATTRIBUTE, environment), (_RELEASE_ATTRIBUTE, release)) + for key, value in ((_ENVIRONMENT_ATTRIBUTE, environment), (RELEASE_ATTRIBUTE, release)) if value is not None } ) - return TracerProvider(resource=Resource.create(dict(attributes))) + provider: Final = TracerProvider(resource=Resource.create(dict(attributes))) + _litellm_built_providers.add(provider) + return provider class DiscardingSpanExporter(SpanExporter): @@ -187,23 +203,122 @@ class DiscardingSpanExporter(SpanExporter): return True -def evict_stale_langfuse_resources(*, public_key: str | None, secret_key: str | None, base_url: str | None) -> None: - """Drop a cached client whose credentials no longer match the ones being requested. +def _evict_if_stale_locked( + *, public_key: object, secret_key: object, base_url: object +) -> LangfuseResourceManager | None: + """Assumes ``LangfuseResourceManager._lock`` is held; returns the still-valid cached bundle, if any. langfuse keys its client registry on the public key alone, so a rotated secret or a moved host silently keeps exporting with the original values. Only the one stale entry is removed; the SDK's own reset would shut down - every other tenant in the process. + every other tenant in the process. A stale bundle no live litellm client + still holds also has its export thread stopped — but only when litellm + built the provider, because a bundle adopted from user code may share the + process-global provider. """ if not public_key: - return + return None + cached: Final = LangfuseResourceManager._instances.get(public_key) # pyright: ignore[reportPrivateUsage] # registry has no public accessor + if cached is None: + return None + if getattr(cached, "secret_key", None) == secret_key and getattr(cached, "base_url", None) == base_url: + return cached + LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor + with _LIVE_CLIENTS_LOCK: + holders: Final = _live_clients.get(cached) + abandoned: Final = holders is None or len(holders) == 0 + provider: Final = getattr(cached, "tracer_provider", None) + if abandoned and provider is not None and provider in _litellm_built_providers: + provider.shutdown() + return None + + +def evict_stale_langfuse_resources(*, public_key: str | None, secret_key: str | None, base_url: str | None) -> None: + """Drop a cached client whose credentials no longer match the ones being requested.""" with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor - cached: Final = LangfuseResourceManager._instances.get(public_key) # pyright: ignore[reportPrivateUsage] # registry has no public accessor - if cached is None: - return - if getattr(cached, "secret_key", None) == secret_key and getattr(cached, "base_url", None) == base_url: - return - LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor + _evict_if_stale_locked(public_key=public_key, secret_key=secret_key, base_url=base_url) + + +def _build_verified_span_exporter(*, public_key: object, secret_key: object, base_url: object) -> SpanExporter | None: + """Rebuild litellm's TLS material onto the export channel. + + v2 ingested through the injected httpx client, which carried litellm's CA + bundle and client certificate; v4 ships every observation through its own + OTLP exporter, so a private-CA deployment would fail TLS on every export in + a background thread while ``auth_check`` (still on the httpx client) stays + green. Only built when custom TLS material is configured; endpoint and + headers mirror ``langfuse._client.span_processor``. + """ + import litellm + + ca_bundle: Final = litellm.ssl_verify if isinstance(litellm.ssl_verify, str) else None + configured_certificate: Final = os.getenv("SSL_CERTIFICATE") or litellm.ssl_certificate + client_certificate: Final = configured_certificate if isinstance(configured_certificate, str) else None + if ca_bundle is None and client_certificate is None: + return None + import langfuse as langfuse_package + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + + langfuse_version: Final = getattr(langfuse_package, "__version__", "unknown") + + export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH") + endpoint: Final = f"{base_url}/{export_path}" if export_path else f"{base_url}/api/public/otel/v1/traces" + encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii") + return OTLPSpanExporter( + endpoint=endpoint, + headers={ # mutable-ok: the exporter copies these into its session headers + "Authorization": "Basic " + encoded_auth, + "x-langfuse-sdk-name": "python", + "x-langfuse-sdk-version": langfuse_version, + "x-langfuse-public-key": str(public_key), + }, + certificate_file=ca_bundle, + client_certificate_file=client_certificate, + ) + + +def acquire_langfuse_client( + *, + parameters: Mapping[str, object], + environment: str | None, + release: str | None, + mock_mode: bool, +) -> Langfuse: + """Evict-check, construct, and register a client as one atomic step. + + The SDK registry lock is held across the whole sequence: released between + eviction and construction, two concurrent inits for the same public key + with different secrets can bind one tenant's logger to the other tenant's + exporter. The isolated provider is only built when the registry does not + already hold the key — a discarded ``TracerProvider`` stays pinned forever + by its atexit hook, so building one per health probe or alerting lookup + would leak a provider each time. + """ + public_key: Final = parameters.get("public_key") + span_exporter: Final = ( + DiscardingSpanExporter() + if mock_mode + else _build_verified_span_exporter( + public_key=public_key, + secret_key=parameters.get("secret_key"), + base_url=parameters.get("base_url"), + ) + ) + with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor + cached: Final = _evict_if_stale_locked( + public_key=public_key, + secret_key=parameters.get("secret_key"), + base_url=parameters.get("base_url"), + ) + client: Final = Langfuse( + **parameters, # pyright: ignore[reportArgumentType] # kwargs-ok: dict mirrors the typed ctor, values resolved by the callers + tracer_provider=None + if cached is not None + else build_isolated_tracer_provider(environment=environment, release=release), + span_exporter=span_exporter, + ) + register_langfuse_client(client) + return client _LIVE_CLIENTS_LOCK: Final = threading.Lock() @@ -256,21 +371,24 @@ def shutdown_langfuse_client(client: Langfuse) -> None: A client that shares its resources with another live client only flushes: shutting the shared provider down here would silence the other client for the rest of its life, as it did before the reference count existed. + + The registry entry is removed before the blocking shutdown so a concurrent + construct builds a fresh bundle instead of adopting a dying one, and the + provider is only shut down when litellm built it: a bundle adopted from + user code may share the process-global provider. """ resources: Final = getattr(client, "_resources", None) client.flush() if resources is None: client.shutdown() return - if not _release_langfuse_resources(resources, client): - return + public_key: Final = getattr(resources, "public_key", None) + with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor + if not _release_langfuse_resources(resources, client): + return + if public_key is not None and LangfuseResourceManager._instances.get(public_key) is resources: # pyright: ignore[reportPrivateUsage] # registry has no public accessor + LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor client.shutdown() provider: Final = getattr(resources, "tracer_provider", None) - if provider is not None and not isinstance(provider, otel_trace.ProxyTracerProvider): + if provider is not None and provider in _litellm_built_providers: provider.shutdown() - public_key: Final = getattr(resources, "public_key", None) - if public_key is None: - return - with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor - if LangfuseResourceManager._instances.get(public_key) is resources: # pyright: ignore[reportPrivateUsage] # registry has no public accessor - LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor diff --git a/tests/test_litellm/integrations/langfuse/test_langfuse_prompt_management.py b/tests/test_litellm/integrations/langfuse/test_langfuse_prompt_management.py index f9009737ca5..ef8fa34b3f5 100644 --- a/tests/test_litellm/integrations/langfuse/test_langfuse_prompt_management.py +++ b/tests/test_litellm/integrations/langfuse/test_langfuse_prompt_management.py @@ -91,13 +91,12 @@ class TestLangfusePromptManagement: "litellm.integrations.langfuse.langfuse_prompt_management.LangFuseLogger._get_langfuse_flush_interval", return_value=1, ), - patch.dict("sys.modules", {"langfuse": self._mock_langfuse}), + patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", mock_langfuse_class), # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads patch( "litellm.llms.custom_httpx.http_handler.get_ssl_configuration", return_value=False, ) as mock_get_ssl, ): - self._mock_langfuse.Langfuse = mock_langfuse_class langfuse_client_init( langfuse_public_key="pk-1234", @@ -128,15 +127,12 @@ class _RecordingLangfuseForEnv: (("Production", "default"), ("production ", "production"), ("prod", "prod")), ) def test_langfuse_client_init_resolves_deployment_environment(monkeypatch, env_value, expected): - mock_langfuse_module: Final = MagicMock() - mock_langfuse_module.version.__version__ = "2.60.0" - mock_langfuse_module.Langfuse = _RecordingLangfuseForEnv monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") monkeypatch.setenv("LANGFUSE_HOST", "https://test.langfuse.com") monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", env_value) monkeypatch.setattr(_RecordingLangfuseForEnv, "last_environment", None) - with patch.dict("sys.modules", MappingProxyType({"langfuse": mock_langfuse_module})): + with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuseForEnv): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads langfuse_client_init.cache_clear() langfuse_client_init() langfuse_client_init.cache_clear() diff --git a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py index 629240d83a1..80b05512f8d 100644 --- a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py +++ b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py @@ -23,6 +23,7 @@ from litellm.integrations.langfuse.langfuse import ( ) from litellm.integrations.langfuse.langfuse_sdk import ( AS_ROOT_ATTRIBUTE, + _litellm_built_providers, PUBLIC_ATTRIBUTE, RELEASE_ATTRIBUTE, build_isolated_tracer_provider, @@ -100,9 +101,9 @@ def test_guardrail_span_with_float_timestamps_does_not_break_the_generation(clie lf, exporter = client context, claim_root = open_trace_context(client=lf, trace_id="9" * 32, parent_observation_id=None) guardrail_start = 1709294400.0 - start_child_span(client=lf, context=context, name="guardrail", start_time=guardrail_start, attributes={}).end( - end_time=to_unix_nanos(guardrail_start + 2) - ) + start_child_span( + client=lf, context=context, name="guardrail", start_time=guardrail_start, claim_trace_root=claim_root, attributes={} + ).end(end_time=to_unix_nanos(guardrail_start + 2)) start_generation( client=lf, context=context, name="gen", start_time=CALL_START, claim_trace_root=claim_root, attributes={} ).end(end_time=to_unix_nanos(CALL_END)) @@ -141,9 +142,9 @@ def test_child_span_keeps_its_own_window_and_stays_a_sibling(client): lf, exporter = client context, claim_root = open_trace_context(client=lf, trace_id="d" * 32, parent_observation_id=None) guardrail_start = CALL_START + timedelta(seconds=1) - start_child_span(client=lf, context=context, name="guardrail", start_time=guardrail_start, attributes={}).end( - end_time=to_unix_nanos(guardrail_start + timedelta(seconds=2)) - ) + start_child_span( + client=lf, context=context, name="guardrail", start_time=guardrail_start, claim_trace_root=claim_root, attributes={} + ).end(end_time=to_unix_nanos(guardrail_start + timedelta(seconds=2))) start_generation( client=lf, context=context, name="gen", start_time=CALL_START, claim_trace_root=claim_root, attributes={} ).end(end_time=to_unix_nanos(CALL_END)) @@ -152,8 +153,10 @@ def test_child_span_keeps_its_own_window_and_stays_a_sibling(client): guardrail = _only_span(exporter, "guardrail") generation = _only_span(exporter, "gen") assert (guardrail.end_time - guardrail.start_time) == 2 * 1_000_000_000 - assert guardrail.parent.span_id == generation.parent.span_id assert guardrail.context.trace_id == generation.context.trace_id + # the shared remote parent is fabricated and never exported, so both must claim trace root + assert guardrail.attributes.get(AS_ROOT_ATTRIBUTE) is True + assert generation.attributes.get(AS_ROOT_ATTRIBUTE) is True def test_release_is_carried_on_the_root_observation(client): @@ -423,6 +426,7 @@ def _shared_resources_pair(): exporter = InMemorySpanExporter() provider = TracerProvider() provider.add_span_processor(SimpleSpanProcessor(exporter)) + _litellm_built_providers.add(provider) first = Langfuse( public_key=PUBLIC_KEY, secret_key="sk-original", @@ -502,3 +506,43 @@ def test_shutdown_of_a_stale_client_does_not_deregister_the_live_one(): assert stale_resources is not live._resources assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is live._resources + + +def test_ssl_exporter_is_only_built_with_custom_tls_material(monkeypatch, tmp_path): + """v4 exports over its own OTLP channel, so litellm's CA bundle must be rebuilt onto it.""" + import litellm + from litellm.integrations.langfuse.langfuse_sdk import _build_verified_span_exporter + + monkeypatch.delenv("SSL_CERTIFICATE", raising=False) + monkeypatch.setattr(litellm, "ssl_verify", True) + monkeypatch.setattr(litellm, "ssl_certificate", None) + assert ( + _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") is None + ) + + ca_path = tmp_path / "private-ca.pem" + ca_path.write_text("dummy") + monkeypatch.setattr(litellm, "ssl_verify", str(ca_path)) + exporter = _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") + assert exporter is not None + assert exporter._endpoint == "https://lf.internal.example/api/public/otel/v1/traces" + assert exporter._certificate_file == str(ca_path) + assert exporter._headers["x-langfuse-public-key"] == "pk" + + +def test_second_client_on_the_same_key_does_not_build_another_provider(): + """A discarded TracerProvider is pinned forever by its atexit hook.""" + from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client + + pk = "pk-provider-reuse-test" + LangfuseResourceManager._instances.pop(pk, None) + parameters = {"public_key": pk, "secret_key": "sk-reuse", "base_url": "http://127.0.0.1:1"} + try: + first = acquire_langfuse_client(parameters=parameters, environment=None, release=None, mock_mode=True) + providers_after_first = len(_litellm_built_providers) + second = acquire_langfuse_client(parameters=parameters, environment=None, release=None, mock_mode=True) + + assert second._resources is first._resources + assert len(_litellm_built_providers) == providers_after_first + finally: + LangfuseResourceManager._instances.pop(pk, None) diff --git a/tests/test_litellm/integrations/test_langfuse.py b/tests/test_litellm/integrations/test_langfuse.py index 424e847d044..221866de729 100644 --- a/tests/test_litellm/integrations/test_langfuse.py +++ b/tests/test_litellm/integrations/test_langfuse.py @@ -309,7 +309,6 @@ class TestLangfuseUsageDetails(unittest.TestCase): side_effect=lambda generation_params, **kwargs: generation_params, create=True, ) as mock_add_prompt_params, - patch.object(self.logger, "_supports_prompt", return_value=True), ): # Create a mock response object with usage information containing None values response_obj = MagicMock() @@ -980,7 +979,7 @@ def test_failure_handler_langfuse_kwargs_excludes_original_response(): try: # Mock LangFuseHandler to return our capturing mock logger - with patch("litellm.litellm_core_utils.litellm_logging.LangFuseHandler") as mock_handler_class: + with patch("litellm.litellm_core_utils.litellm_logging.LangFuseHandler") as mock_handler_class: # test-quality-ok: route the request to the capturing logger; the real handler builds live clients mock_handler_class.get_langfuse_logger_for_request.return_value = mock_langfuse_logger # Call the actual failure_handler @@ -1047,7 +1046,7 @@ async def test_async_log_failure_event_logs_to_langfuse(): "generation_id": "mock-gen", } - with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler: + with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler: # test-quality-ok: route the request to the capturing logger; the real handler builds live clients mock_handler.get_langfuse_logger_for_request.return_value = mock_logger kwargs = { @@ -1112,7 +1111,7 @@ async def test_async_log_failure_event_works_without_standard_logging_object(): "generation_id": "mock-gen", } - with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler: + with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler: # test-quality-ok: route the request to the capturing logger; the real handler builds live clients mock_handler.get_langfuse_logger_for_request.return_value = mock_logger kwargs = { @@ -1263,7 +1262,7 @@ class _RecordingLangfuse: def _build_langfuse_logger(monkeypatch) -> LangFuseLogger: monkeypatch.setenv("LANGFUSE_MOCK", "false") monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0) - with patch("langfuse.Langfuse", _RecordingLangfuse): + with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads return LangFuseLogger( langfuse_public_key="pk-lit5228", langfuse_secret="sk-lit5228", @@ -1275,7 +1274,7 @@ def test_langfuse_environment_is_passed_to_sdk_client(monkeypatch): monkeypatch.setenv("LANGFUSE_MOCK", "false") monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False) monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0) - with patch("langfuse.Langfuse", _RecordingLangfuse): + with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads logger = LangFuseLogger( langfuse_public_key="pk-env", langfuse_secret="sk-env", @@ -1290,7 +1289,7 @@ def test_langfuse_environment_falls_back_to_deployment_env_var(monkeypatch): monkeypatch.setenv("LANGFUSE_MOCK", "false") monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", "deployment-wide") monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0) - with patch("langfuse.Langfuse", _RecordingLangfuse): + with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads logger = LangFuseLogger( langfuse_public_key="pk-env", langfuse_secret="sk-env", @@ -1577,7 +1576,7 @@ def test_langfuse_environment_is_coerced_and_validated(monkeypatch): monkeypatch.setenv("LANGFUSE_MOCK", "false") monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False) monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0) - with patch("langfuse.Langfuse", _RecordingLangfuse): + with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads logger = LangFuseLogger( langfuse_public_key="pk-env", langfuse_secret="sk-env", @@ -1604,7 +1603,7 @@ def test_langfuse_empty_environment_falls_back_and_is_not_dynamic(monkeypatch): # '' falls back to the deployment env var at init monkeypatch.setenv("LANGFUSE_MOCK", "false") monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0) - with patch("langfuse.Langfuse", _RecordingLangfuse): + with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads logger = LangFuseLogger( langfuse_public_key="pk-env", langfuse_secret="sk-env", @@ -1769,3 +1768,59 @@ def test_resolve_credentials_falls_back_to_langfuse_base_url(monkeypatch): _, _, host = langfuse_module.resolve_langfuse_credentials(langfuse_host="https://explicit.example") assert host == "https://explicit.example" + + +def test_version_gate_rejects_v5_prereleases(): + """"5.0.0rc1" sorts below "5", so a plain version comparison would admit it.""" + langfuse_module.raise_if_unsupported_langfuse_version("4.7") + with pytest.raises(ImportError): + langfuse_module.raise_if_unsupported_langfuse_version("5.0.0rc1") + with pytest.raises(ImportError): + langfuse_module.raise_if_unsupported_langfuse_version("5.0.0") + + +def test_int_steering_values_survive_v4_propagation(): + """v4 drops non-string propagated values outright; v2's pydantic coerced them.""" + rig = _steering_logger() + + _, _, span = _emit( + rig, metadata={"trace_user_id": 12345, "session_id": 67, "trace_version": 3, "tags": ["ok", 99]} + ) + + # the SDK validates AFTER litellm's coercion: a surviving attribute proves the value was a str + assert span.attributes["user.id"] == "12345" + assert span.attributes["session.id"] == "67" + assert span.attributes["langfuse.version"] == "3" + # tags reach propagation as a list; non-str entries must be coerced item-wise + assert langfuse_module._coerce_propagated_value(["ok", 99]) == ["ok", "99"] + + +def test_long_steering_values_are_capped_not_dropped(): + """The SDK drops any propagated value over 200 characters with only a warning.""" + rig = _steering_logger() + long_user: Final = "u" * 250 + + _, _, span = _emit(rig, metadata={"trace_user_id": long_user}) + + assert span.attributes["user.id"] == "u" * 200 + + +def test_returned_generation_id_names_the_exported_observation(): + """v4 derives observation ids from the OTel span, so a pre-computed id would name nothing.""" + logger, exporter = _steering_logger() + + returned = logger.log_event_on_langfuse( + kwargs={ + "call_type": "completion", + "litellm_params": {"metadata": {"trace_id": "d" * 32}}, + "messages": [{"role": "user", "content": "the-input"}], + "optional_params": {}, + }, + response_obj=litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "the-output"}}]), + start_time=datetime.datetime.now(), + end_time=datetime.datetime.now(), + ) + logger.Langfuse.flush() + + span = exporter.get_finished_spans()[-1] + assert returned["generation_id"] == format(span.context.span_id, "016x")