diff --git a/litellm/integrations/langfuse/langfuse.py b/litellm/integrations/langfuse/langfuse.py index ec6b6b1074c..fafc1c2b7a2 100644 --- a/litellm/integrations/langfuse/langfuse.py +++ b/litellm/integrations/langfuse/langfuse.py @@ -284,6 +284,7 @@ class LangFuseLogger: f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\033[0m" ) from e raise_if_unsupported_langfuse_version(self.langfuse_sdk_version) + from litellm.integrations.langfuse.langfuse_sdk import configured_release self.public_key, self.secret_key, self.langfuse_host = resolve_langfuse_credentials( langfuse_public_key=langfuse_public_key, @@ -297,7 +298,7 @@ class LangFuseLogger: self.langfuse_environment: str | None = _env_override else: self.langfuse_environment = self.resolve_deployment_environment() - self.langfuse_release = os.getenv("LANGFUSE_RELEASE") + self.langfuse_release = configured_release() self.langfuse_debug = parse_langfuse_debug(os.getenv("LANGFUSE_DEBUG")) self.langfuse_flush_interval = LangFuseLogger._get_langfuse_flush_interval(flush_interval) @@ -826,15 +827,15 @@ class LangFuseLogger: cache_read_input_tokens=cache_read_input_tokens, ) - generation_name = clean_metadata.pop("generation_name", None) - if generation_name is None: - # if `generation_name` is None, use sensible default values - # If using litellm proxy user `key_alias` if not None - # If `key_alias` is None, just log `litellm-{call_type}` as the generation name - _user_api_key_alias: Final = cast(str | None, clean_metadata.get("user_api_key_alias", None)) - generation_name = f"litellm-{cast(str, kwargs.get('call_type', 'completion'))}" - if _user_api_key_alias is not None: - generation_name = f"litellm:{_user_api_key_alias}" + requested_generation_name: Final = clean_metadata.pop("generation_name", None) + _user_api_key_alias: Final = cast(str | None, clean_metadata.get("user_api_key_alias", None)) + generation_name: Final = ( + str(requested_generation_name) + if requested_generation_name is not None + else f"litellm:{_user_api_key_alias}" + if _user_api_key_alias is not None + else f"litellm-{cast(str, kwargs.get('call_type', 'completion'))}" + ) if response_obj is not None: system_fingerprint = getattr(response_obj, "system_fingerprint", None) diff --git a/litellm/integrations/langfuse/langfuse_prompt_management.py b/litellm/integrations/langfuse/langfuse_prompt_management.py index ec3156ac56f..732f3e59832 100644 --- a/litellm/integrations/langfuse/langfuse_prompt_management.py +++ b/litellm/integrations/langfuse/langfuse_prompt_management.py @@ -2,7 +2,6 @@ Call Hook for LiteLLM Proxy which allows Langfuse prompt management. """ -import os from functools import lru_cache from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast @@ -121,7 +120,7 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge self.langfuse_sdk_version = installed_langfuse_version() raise_if_unsupported_langfuse_version(self.langfuse_sdk_version) - from .langfuse_sdk import acquire_langfuse_tracing + from .langfuse_sdk import acquire_langfuse_tracing, configured_release self.api_client = langfuse_client_init( langfuse_public_key=langfuse_public_key, @@ -139,7 +138,7 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge secret_key=str(self.secret_key), base_url=self.langfuse_host, environment=LangFuseLogger.resolve_deployment_environment(), - release=os.getenv("LANGFUSE_RELEASE"), + release=configured_release(), flush_interval=LangFuseLogger._get_langfuse_flush_interval(flush_interval), # pyright: ignore[reportPrivateUsage] # shared env-fallback helper, not part of the logger's API mock_mode=should_use_langfuse_mock(), ) diff --git a/litellm/integrations/langfuse/langfuse_sdk.py b/litellm/integrations/langfuse/langfuse_sdk.py index b3b9583e2a0..ec88949906d 100644 --- a/litellm/integrations/langfuse/langfuse_sdk.py +++ b/litellm/integrations/langfuse/langfuse_sdk.py @@ -46,7 +46,10 @@ __all__ = ( "build_langfuse_client", "build_langfuse_tracing", "configured_flush_at", + "configured_max_retries", + "configured_release", "configured_sample_rate", + "configured_timeout", "flush_langfuse_tracing", "observation_attributes", "release_langfuse_tracing", @@ -64,6 +67,20 @@ _TRACER_NAME: Final = "langfuse-sdk" _MAX_QUEUE_SIZE: Final = 100_000 _DEFAULT_FLUSH_AT: Final = 512 _CHANNEL_RETIRE_GRACE_SECONDS: Final = 60.0 +_DEFAULT_TIMEOUT_SECONDS: Final = 20.0 +_DEFAULT_MAX_RETRIES: Final = 3 +_COMMON_RELEASE_ENVS: Final = ( + "RENDER_GIT_COMMIT", + "CI_COMMIT_SHA", + "CIRCLE_SHA1", + "SOURCE_VERSION", + "TRAVIS_COMMIT", + "GIT_COMMIT", + "GITHUB_SHA", + "BITBUCKET_COMMIT", + "BUILD_SOURCEVERSION", + "DRONE_COMMIT_SHA", +) _SPAN_LIMITS: Final = SpanLimits( max_attributes=SpanLimits.UNSET, max_events=128, @@ -429,6 +446,34 @@ def configured_sample_rate() -> float: return parsed +def configured_timeout() -> float: + """``LANGFUSE_TIMEOUT`` in seconds for every export and REST call, the v2 SDK's 20 s when unset. + + A value that is not a number raises, as the v2 client did at construction, so a typo is not silently ignored. + """ + return float(os.environ.get("LANGFUSE_TIMEOUT", _DEFAULT_TIMEOUT_SECONDS)) + + +def configured_max_retries() -> int: + """``LANGFUSE_MAX_RETRIES`` as the number of re-sends after a failed export, the v2 SDK's knob and default.""" + raw: Final = os.environ.get("LANGFUSE_MAX_RETRIES") + if raw is None: + return _DEFAULT_MAX_RETRIES + if not raw.strip().isdigit(): + verbose_logger.warning( + "LANGFUSE_MAX_RETRIES=%r is not a whole number; retrying %d times", raw, _DEFAULT_MAX_RETRIES + ) + return _DEFAULT_MAX_RETRIES + return int(raw) + + +def configured_release() -> str | None: + """``LANGFUSE_RELEASE``, else the commit variable of the CI or deploy platform, as both SDK generations resolve it.""" + return os.environ.get("LANGFUSE_RELEASE") or next( + (os.environ[name] for name in _COMMON_RELEASE_ENVS if name in os.environ), None + ) + + def configured_flush_at() -> int: """``LANGFUSE_FLUSH_AT`` as the export batch size, the SDK's own knob, with its default when unset or unusable.""" raw: Final = os.environ.get("LANGFUSE_FLUSH_AT") @@ -485,7 +530,12 @@ class LangfuseSpanExporter(SpanExporter): delays: Sequence[float] = (1.0, 2.0, 4.0) def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: - body: Final = encode_spans(spans).SerializeToString() + body: Final = _encode(spans) + if body is None: + return SpanExportResult.FAILURE + return self._send(body) + + def _send(self, body: bytes) -> SpanExportResult: for delay in self.delays: outcome: _ExportOutcome = self._post(body) if outcome != "retry": @@ -518,8 +568,26 @@ class LangfuseSpanExporter(SpanExporter): return True +def _encode(spans: Sequence[ReadableSpan]) -> bytes | None: + """The OTLP body, or ``None`` when nothing survived: a span the encoder rejects is dropped, not the whole batch.""" + try: + return encode_spans(spans).SerializeToString() + except Exception: # noqa: BLE001 # protobuf raises TypeError or ValueError depending on the field + kept: Final = tuple(span for span in spans if _encodes(span)) + verbose_logger.error("Langfuse export dropped %d span(s) the OTLP encoder rejected", len(spans) - len(kept)) + return encode_spans(kept).SerializeToString() if kept else None + + +def _encodes(span: ReadableSpan) -> bool: + try: + encode_spans((span,)) + except Exception: # noqa: BLE001 # same encoder failure modes as above + return False + return True + + def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> LangfuseSpanExporter: - """Endpoint, headers and timeout mirror the SDK's span processor so the server treats the spans as v4 SDK traffic.""" + """Endpoint, headers, timeout and retries mirror the v2 SDK's so the server treats the spans as SDK traffic.""" export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH") or "/api/public/otel/v1/traces" encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii") return LangfuseSpanExporter( @@ -534,7 +602,8 @@ def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> "x-langfuse-public-key": public_key, } ), - timeout=float(os.getenv("LANGFUSE_TIMEOUT", "5")), + timeout=configured_timeout(), + delays=tuple(2.0**attempt for attempt in range(configured_max_retries())), ) @@ -549,6 +618,26 @@ def _resource(*, environment: str | None, release: str | None) -> Resource: ) +class _ExportLedger(SpanExporter): + """Counts the batches the exporter gave up on, so a flush can report delivery rather than a drained queue.""" + + def __init__(self, exporter: SpanExporter) -> None: + self.exporter: Final = exporter + self.failed_batches = 0 + + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + result: Final = self.exporter.export(spans) + if result is not SpanExportResult.SUCCESS: + self.failed_batches += 1 + return result + + def shutdown(self) -> None: + self.exporter.shutdown() + + def force_flush(self, timeout_millis: int = 30_000) -> bool: + return self.exporter.force_flush(timeout_millis) + + @dataclass(frozen=True, slots=True) class LangfuseTracing: """litellm's own export channel to one Langfuse project: a provider, its tracer and the exporter behind them. @@ -559,9 +648,12 @@ class LangfuseTracing: provider: TracerProvider tracer: Tracer + ledger: _ExportLedger def flush(self, timeout_millis: int = 30_000) -> bool: - return self.provider.force_flush(timeout_millis) + """``True`` only when the queue drained in time and every batch it held was accepted by the destination.""" + failed_before: Final = self.ledger.failed_batches + return self.provider.force_flush(timeout_millis) and self.ledger.failed_batches == failed_before def shutdown(self) -> None: self.provider.shutdown() @@ -584,6 +676,7 @@ class _TracingKey: class _Lease: tracing: LangfuseTracing holders: int + retire: threading.Timer | None = None _TRACING_LOCK: Final = threading.Lock() @@ -619,7 +712,9 @@ def acquire_langfuse_tracing( with _TRACING_LOCK: cached: Final = _TRACING.get(key) if cached is not None: - _TRACING[key] = replace(cached, holders=cached.holders + 1) + if cached.retire is not None: + cached.retire.cancel() + _TRACING[key] = replace(cached, holders=cached.holders + 1, retire=None) return cached.tracing created: Final = build_langfuse_tracing( exporter=DiscardingSpanExporter() @@ -648,27 +743,30 @@ def release_langfuse_tracing(tracing: LangfuseTracing, *, grace_seconds: float = key, lease = held if lease.holders <= 0: return - _TRACING[key] = replace(lease, holders=lease.holders - 1) if lease.holders > 1: + _TRACING[key] = replace(lease, holders=lease.holders - 1) return - if grace_seconds <= 0: - _retire_if_unheld(key, tracing) - return - retire: Final = threading.Timer(grace_seconds, _retire_if_unheld, args=(key, tracing)) - retire.name = "langfuse-retire" - retire.daemon = True - retire.start() - - -def _retire_if_unheld(key: _TracingKey, tracing: LangfuseTracing) -> None: - with _TRACING_LOCK: - lease: Final = _TRACING.get(key) - if lease is None or lease.tracing is not tracing or lease.holders > 0: + if grace_seconds > 0: + retire: Final = threading.Timer(grace_seconds, lambda: _retire_unless_reacquired(key, retire)) + retire.name = "langfuse-retire" + retire.daemon = True + _TRACING[key] = _Lease(tracing=tracing, holders=0, retire=retire) + retire.start() return del _TRACING[key] tracing.shutdown() +def _retire_unless_reacquired(key: _TracingKey, timer: threading.Timer) -> None: + """Only the timer the lease still points at may retire it; a re-acquire cancels and clears the pending one.""" + with _TRACING_LOCK: + lease: Final = _TRACING.get(key) + if lease is None or lease.retire is not timer: + return + del _TRACING[key] + lease.tracing.shutdown() + + class _FlushWorker(threading.Thread): """Daemon, so a channel still blocked at the deadline cannot hold up interpreter exit.""" @@ -725,15 +823,16 @@ def build_langfuse_tracing( id_generator=_RequestedIdGenerator(), span_limits=_SPAN_LIMITS, ) + ledger: Final = _ExportLedger(exporter) provider.add_span_processor( BatchSpanProcessor( - exporter, + ledger, max_queue_size=_MAX_QUEUE_SIZE, max_export_batch_size=flush_at, schedule_delay_millis=flush_interval_millis, ) ) - return LangfuseTracing(provider=provider, tracer=provider.get_tracer(_TRACER_NAME)) + return LangfuseTracing(provider=provider, tracer=provider.get_tracer(_TRACER_NAME), ledger=ledger) @dataclass(frozen=True, slots=True) @@ -758,8 +857,9 @@ class LangfuseApiClient: over ``LangfuseTracing``; nothing here exports spans. Prompts are cached for ``LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS`` (60 by default) as the SDK - does, refreshed on the request that finds them stale; a refresh that fails keeps serving the - stale prompt rather than failing the request, again like the SDK. + does. A stale prompt is served at once and refreshed on a background thread, so the request + that finds it stale, and the event loop it runs on, never wait for the REST round trip; a + refresh that fails keeps serving the stale prompt rather than failing the request, again like the SDK. """ def __init__(self, api: LangfuseAPI, *, prompt_cache_ttl_seconds: float) -> None: @@ -767,6 +867,8 @@ class LangfuseApiClient: self.prompt_cache_ttl_seconds: Final = prompt_cache_ttl_seconds # mutable-ok: per-client prompt cache, guarded by _lock self._prompts: Final[dict[_PromptKey, _CachedPrompt]] = {} + # mutable-ok: keys with a refresh in flight, guarded by _lock + self._refreshing: Final[set[_PromptKey]] = set() self._lock: Final = threading.Lock() def auth_check(self) -> bool: @@ -784,19 +886,35 @@ class LangfuseApiClient: key: Final[_PromptKey] = (name, version, label) with self._lock: cached: Final = self._prompts.get(key) - if cached is not None and monotonic() - cached.fetched_at < self.prompt_cache_ttl_seconds: - return cached.prompt - try: - fetched: Final = _prompt_client(self.api.prompts.get(name, version=version, label=label)) - except Exception as error: - if cached is None: - raise - verbose_logger.warning("Langfuse prompt %r refresh failed, serving the cached version: %s", name, error) - return cached.prompt + if cached is None: + return self._fetch(key) + if monotonic() - cached.fetched_at >= self.prompt_cache_ttl_seconds: + self._refresh_in_background(key) + return cached.prompt + + def _fetch(self, key: _PromptKey) -> PromptClient: + name, version, label = key + fetched: Final = _prompt_client(self.api.prompts.get(name, version=version, label=label)) with self._lock: self._prompts[key] = _CachedPrompt(prompt=fetched, fetched_at=monotonic()) return fetched + def _refresh_in_background(self, key: _PromptKey) -> None: + with self._lock: + if key in self._refreshing: + return + self._refreshing.add(key) + threading.Thread(target=self._refresh, args=(key,), name="langfuse-prompt-refresh", daemon=True).start() + + def _refresh(self, key: _PromptKey) -> None: + try: + self._fetch(key) + except Exception as error: # noqa: BLE001 # a failed refresh keeps the stale prompt in service + verbose_logger.warning("Langfuse prompt %r refresh failed, serving the cached version: %s", key[0], error) + finally: + with self._lock: + self._refreshing.discard(key) + def build_langfuse_client( *, @@ -819,7 +937,7 @@ def build_langfuse_client( x_langfuse_sdk_version=version("langfuse"), x_langfuse_public_key=public_key, httpx_client=httpx_client, - timeout=float(os.getenv("LANGFUSE_TIMEOUT", "5")), + timeout=configured_timeout(), ), prompt_cache_ttl_seconds=float(os.getenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", "60")), ) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 36660c92001..a310d5147f7 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -1058,7 +1058,8 @@ async def proxy_shutdown_event(worker_heartbeat: ProxyWorkerHeartbeat | None = N verbose_proxy_logger.info("Langfuse export channels flushed") else: verbose_proxy_logger.warning( - "Langfuse export did not finish within %dms; remaining spans are left to the background exporter", + "Langfuse shutdown flush incomplete: a channel did not finish within %dms or a batch was rejected " + "(see the export errors above); remaining spans are left to the background exporter", LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS, ) except Exception as e: # noqa: BLE001 # shutdown must continue even if the flush fails diff --git a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py index 36b62bad1f1..f77ec9fc952 100644 --- a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py +++ b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py @@ -11,6 +11,7 @@ import threading import uuid from base64 import b64encode from datetime import datetime, timedelta, timezone +from time import monotonic, sleep from types import MappingProxyType from typing import Final @@ -56,6 +57,12 @@ TRACE_A = "a" * 32 PARENT_C = "c" * 16 +@pytest.fixture(autouse=True) +def _own_channel_registry(monkeypatch: pytest.MonkeyPatch) -> None: + """Channels leaked by other test modules would otherwise take part in every process-wide flush here.""" + monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk._TRACING", {}) + + @pytest.fixture(name="channel") def _channel() -> tuple[LangfuseTracing, InMemorySpanExporter]: exporter = InMemorySpanExporter() @@ -725,6 +732,47 @@ def test_channel_reacquired_within_the_grace_is_kept(monkeypatch: pytest.MonkeyP assert len(exporter.get_finished_spans()) == 1 +def test_retire_timer_of_an_earlier_release_cannot_kill_a_reacquired_channel(monkeypatch: pytest.MonkeyPatch): + """release, re-acquire, release: the first timer used to fire into a channel that a later holder still + counted on for its own grace period, shutting the batch thread down while spans were still queued.""" + tracing, exporter = _acquire_recorded(monkeypatch, "pk-lease-race-test") + + release_langfuse_tracing(tracing, grace_seconds=0.2) + assert _acquire(public_key="pk-lease-race-test", mock_mode=False, flush_interval=600.0) is tracing + release_langfuse_tracing(tracing, grace_seconds=600.0) + + threading.Event().wait(0.5) + assert exporter.shutdowns == 0 + assert _acquire(public_key="pk-lease-race-test", mock_mode=False, flush_interval=600.0) is tracing + tracing.tracer.start_span("generation").end() + assert tracing.flush() is True + assert len(exporter.get_finished_spans()) == 1 + + +class _RejectsEverything(SpanExporter): + def export(self, spans) -> SpanExportResult: + return SpanExportResult.FAILURE + + def shutdown(self) -> None: + return None + + def force_flush(self, timeout_millis: int = 30_000) -> bool: + return True + + +def test_flush_is_false_when_the_destination_rejected_a_batch(monkeypatch: pytest.MonkeyPatch): + """The shutdown hook logs "channels flushed" off this value; a drained queue whose batches all + failed at the destination is a loss, not a flush.""" + monkeypatch.setattr( + "litellm.integrations.langfuse.langfuse_sdk._build_span_exporter", lambda **_: _RejectsEverything() + ) + tracing = _acquire(public_key="pk-flush-truth-test", mock_mode=False, flush_interval=600.0) + tracing.tracer.start_span("generation").end() + + assert tracing.flush() is False + assert flush_langfuse_tracing() is True, "an empty queue after the loss has nothing left to fail" + + def test_release_of_a_channel_the_registry_never_handed_out_is_a_no_op(): exporter = InMemorySpanExporter() tracing = build_langfuse_tracing( @@ -1012,19 +1060,56 @@ def test_exporter_does_not_retry_a_rejected_batch(monkeypatch, status): assert slept == [] +def _finished_span_named(name: object): + provider = TracerProvider() + span = provider.get_tracer("t").start_span("placeholder") + span._name = name # pyright: ignore[reportAttributeAccessIssue, reportPrivateUsage] # the SDK only stores str + span.end() + return span + + +def test_exporter_drops_a_span_the_encoder_rejects_and_still_posts_the_rest(monkeypatch, caplog): + """One span the OTLP encoder cannot serialize used to raise out of ``export`` and lose every span in the batch.""" + from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest + + monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", lambda _: None) + exporter, seen = _exporter_over([200]) + + with caplog.at_level(logging.ERROR, logger="LiteLLM"): + result = exporter.export((_finished_span(), _finished_span_named(12345), _finished_span())) + + assert result is SpanExportResult.SUCCESS + (request,) = seen + decoded = ExportTraceServiceRequest() + decoded.ParseFromString(request.content) + assert [span.name for span in decoded.resource_spans[0].scope_spans[0].spans] == ["generation", "generation"] + assert "dropped 1 span(s)" in caplog.text + + +def test_exporter_reports_failure_when_no_span_of_the_batch_can_be_encoded(monkeypatch): + exporter, seen = _exporter_over([200]) + + assert exporter.export((_finished_span_named(12345),)) is SpanExportResult.FAILURE + assert seen == [] + + def test_built_exporter_uses_the_shared_litellm_handler_and_langfuse_headers(monkeypatch): """No private requests session or TLS adapter: the channel is the same handler the rest of litellm uses.""" from litellm.llms.custom_httpx.http_handler import _get_httpx_client monkeypatch.delenv("LANGFUSE_TIMEOUT", raising=False) + monkeypatch.delenv("LANGFUSE_MAX_RETRIES", raising=False) default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") assert default.handler is _get_httpx_client() - assert default.timeout == 5 + assert default.timeout == 20 + assert len(default.delays) == 3 - monkeypatch.setenv("LANGFUSE_TIMEOUT", "20") + monkeypatch.setenv("LANGFUSE_TIMEOUT", "7.5") + monkeypatch.setenv("LANGFUSE_MAX_RETRIES", "1") exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") assert exporter.endpoint == "https://lf.internal.example/api/public/otel/v1/traces" - assert exporter.timeout == 20 + assert exporter.timeout == 7.5 + assert exporter.delays == (1.0,) assert exporter.headers["Authorization"] == "Basic " + b64encode(b"pk:sk").decode() assert exporter.headers["x-langfuse-public-key"] == "pk" assert exporter.headers["x-langfuse-sdk-version"] == installed_langfuse_version() @@ -1091,6 +1176,78 @@ class _RecordingPromptsApi: ) +class _BlockingPromptsApi(_RecordingPromptsApi): + """Every fetch after the first blocks until the test releases it, and may be told to fail.""" + + def __init__(self) -> None: + super().__init__() + self.release = threading.Event() + self.fail_refresh = False + + def get(self, name: str, *, version: int | None, label: str | None): + is_refresh = bool(self.requests) + prompt = super().get(name, version=version, label=label) + if is_refresh: + assert self.release.wait(5), "refresh was never released" + if self.fail_refresh: + raise RuntimeError("langfuse is down") + return prompt + + +def _wait_until(predicate, timeout: float = 5.0) -> None: + for _ in range(int(timeout / 0.01)): + if predicate(): + return + sleep(0.01) + raise AssertionError("condition not met in time") + + +def test_stale_prompt_is_served_at_once_while_the_refresh_runs_elsewhere(): + """``get_prompt`` runs on the proxy's event loop; a stale entry used to refetch inline and block every + request on the REST round trip. The stale prompt is returned immediately and refreshed off-thread.""" + api = _BlockingPromptsApi() + client = LangfuseApiClient(api, prompt_cache_ttl_seconds=0) # pyright: ignore[reportArgumentType] # duck-typed prompts API + + first = client.get_prompt("greeting") + started = monotonic() + stale = client.get_prompt("greeting") + + assert stale is first, "the stale prompt must come back without waiting on the refresh" + assert monotonic() - started < 1.0, "the stale read waited on the blocked refresh" + _wait_until(lambda: len(api.requests) == 2) + api.release.set() + _wait_until(lambda: client.get_prompt("greeting") is not first) + + +def test_a_failed_background_refresh_keeps_the_stale_prompt_in_service(caplog): + api = _BlockingPromptsApi() + api.fail_refresh = True + client = LangfuseApiClient(api, prompt_cache_ttl_seconds=0) # pyright: ignore[reportArgumentType] # duck-typed prompts API + + first = client.get_prompt("greeting") + with caplog.at_level(logging.WARNING, logger="LiteLLM"): + started = monotonic() + assert client.get_prompt("greeting") is first + assert monotonic() - started < 1.0, "the stale read waited on the blocked refresh" + api.release.set() + _wait_until(lambda: "refresh failed" in caplog.text) + assert client.get_prompt("greeting") is first + + +def test_only_one_refresh_runs_for_a_stale_prompt_under_concurrent_reads(): + api = _BlockingPromptsApi() + client = LangfuseApiClient(api, prompt_cache_ttl_seconds=0.3) # pyright: ignore[reportArgumentType] # duck-typed prompts API + + first = client.get_prompt("greeting") + sleep(0.3) + for _ in range(20): + assert client.get_prompt("greeting") is first + _wait_until(lambda: len(api.requests) == 2) + api.release.set() + _wait_until(lambda: client.get_prompt("greeting") is not first) + assert len(api.requests) == 2 + + def test_prompt_cache_keeps_a_missing_label_apart_from_the_label_named_none(): """A prompt labelled ``"None"`` and the unlabelled default are different prompts in Langfuse and must not answer each other's requests from the cache.""" diff --git a/tests/test_litellm/integrations/test_langfuse.py b/tests/test_litellm/integrations/test_langfuse.py index 7579671d07f..0ab1382ad09 100644 --- a/tests/test_litellm/integrations/test_langfuse.py +++ b/tests/test_litellm/integrations/test_langfuse.py @@ -1339,6 +1339,42 @@ def test_langfuse_environment_falls_back_to_deployment_env_var(monkeypatch): assert _exported_environment(logger) == "deployment-wide" +def _exported_release(logger: LangFuseLogger): + from langfuse import LangfuseOtelSpanAttributes + + return logger.tracing.provider.resource.attributes.get(LangfuseOtelSpanAttributes.RELEASE) + + +@pytest.mark.parametrize("platform_var", ["GITHUB_SHA", "CI_COMMIT_SHA", "RENDER_GIT_COMMIT", "SOURCE_VERSION"]) +def test_release_falls_back_to_the_deploy_platforms_commit_variable(monkeypatch, platform_var): + """Deployments that never set ``LANGFUSE_RELEASE`` still got a release on every trace from the v2 SDK, which + read the CI or hosting platform's commit variable; dropping that silently blanked their release filter.""" + from litellm.integrations.langfuse.langfuse_sdk import _COMMON_RELEASE_ENVS + + for name in ("LANGFUSE_RELEASE", *_COMMON_RELEASE_ENVS): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv(platform_var, "deadbeef") + logger = _build_langfuse_logger(monkeypatch, langfuse_public_key=f"pk-release-{platform_var}") + assert logger.langfuse_release == "deadbeef" + assert _exported_release(logger) == "deadbeef" + + +def test_explicit_langfuse_release_wins_over_the_platform_commit(monkeypatch): + monkeypatch.setenv("LANGFUSE_RELEASE", "v9") + monkeypatch.setenv("GITHUB_SHA", "deadbeef") + logger = _build_langfuse_logger(monkeypatch, langfuse_public_key="pk-release-explicit") + assert _exported_release(logger) == "v9" + + +def test_non_string_generation_name_is_exported_as_its_text(monkeypatch): + """v2 coerced ``generation_name`` through pydantic; a raw int would now fail OTLP encoding and lose the batch.""" + rig = _steering_logger() + + _, _, span = _emit(rig, metadata={"generation_name": 12345}) + + assert span.name == "12345" + + def test_dynamic_langfuse_environment_triggers_dynamic_logger(): from litellm.integrations.langfuse.langfuse_handler import LangFuseHandler from litellm.types.utils import StandardCallbackDynamicParams