diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index 8779153959f..c02ea6032a0 100644 --- a/litellm/integrations/otel/emitter.py +++ b/litellm/integrations/otel/emitter.py @@ -13,6 +13,7 @@ from opentelemetry.trace.status import Status, StatusCode from litellm.integrations.otel.mappers import resolve_mappers from litellm.integrations.otel.mappers.base import AttributeMapper, AttrValue, SpanData +from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES from litellm.integrations.otel.mappers.openinference import fit_indexed_messages from litellm.integrations.otel.model.config import OpenTelemetryV2Config from litellm.integrations.otel.model.payloads import ( @@ -300,9 +301,11 @@ class SpanEmitter: stamped_later: Final = error_attributes(error) if error else _NO_ATTRIBUTES reserved: Final = len(stamped_later.keys() - mapped.keys()) overwrites: Final = carried_keys(span) & mapped.keys() + # The Langfuse request metadata sheds first, so a span at its attribute limit loses nothing else to it. fitted: Final = fit_indexed_messages( MappingProxyType({key: value for key, value in mapped.items() if key not in overwrites}), attribute_budget(span, reserved), + LANGFUSE_REQUEST_METADATA_ATTRIBUTES, ) for key, value in mapped.items(): if key in overwrites or key in fitted: diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 5092b248eac..e09226ac9cd 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -74,6 +74,7 @@ from litellm.integrations.otel.plumbing.metrics import ( create_genai_metrics, ) from litellm.integrations.otel.plumbing.providers import ( + LITELLM_TRACER_NAME, attach_tenant_fan_out, build_tracer_provider, get_event_logger, @@ -97,7 +98,6 @@ if TYPE_CHECKING: StandardLoggingPayload, ) -LITELLM_TRACER_NAME: Final = "litellm" _published_v2_provider: ApiTracerProvider | None = None diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index e60ea539188..babc123efd2 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -13,6 +13,7 @@ the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace contr import json from collections.abc import Callable from datetime import datetime, timezone +from types import MappingProxyType from typing import Final from litellm.integrations.langfuse.langfuse_otel_attributes import serialize_langfuse_datetime @@ -37,6 +38,23 @@ LANGFUSE_TRACE_NAME: Final = "langfuse.trace.name" LANGFUSE_TRACE_USER_ID: Final = "user.id" LANGFUSE_TRACE_SESSION_ID: Final = "session.id" LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags" +LANGFUSE_OBSERVATION_METADATA: Final = "langfuse.observation.metadata" +LANGFUSE_TRACE_METADATA_PREFIX: Final = "langfuse.trace.metadata." +TRACE_IDENTITY_FIELDS: Final = ( + "user_api_key_alias", + "user_api_key_user_id", + "user_api_key_end_user_id", + "user_api_key_team_id", + "user_api_key_team_alias", +) +# The request metadata attributes only a Langfuse destination receives. +LANGFUSE_REQUEST_METADATA_ATTRIBUTES: Final[frozenset[str]] = frozenset( + {LANGFUSE_OBSERVATION_METADATA, *(f"{LANGFUSE_TRACE_METADATA_PREFIX}{name}" for name in TRACE_IDENTITY_FIELDS)} +) + + +def _identity_field(name: str) -> Callable[[LLMCallSpanData], AttrValue | None]: + return lambda d: d.request_metadata.get(name) class LangfuseMapper: @@ -52,6 +70,9 @@ class LangfuseMapper: if d.completion_start_seconds is not None else None ), + **MappingProxyType( + {f"{LANGFUSE_TRACE_METADATA_PREFIX}{name}": _identity_field(name) for name in TRACE_IDENTITY_FIELDS} + ), } # Sub-tables folded into their respective JSON blobs. @@ -75,6 +96,7 @@ class LangfuseMapper: # JSON-payload attributes: each builder returns the serialized blob or None. _BLOB_ATTRS: dict[str, Callable[[LLMCallSpanData], AttrValue | None]] = { + LANGFUSE_OBSERVATION_METADATA: lambda d: json_if(dict(d.request_metadata)), "langfuse.observation.model.parameters": lambda d: json_if( collect(LangfuseMapper._MODEL_PARAMS, d.request_params) ), diff --git a/litellm/integrations/otel/mappers/openinference.py b/litellm/integrations/otel/mappers/openinference.py index 8d7b4d7cf4a..0c465bbe5cb 100644 --- a/litellm/integrations/otel/mappers/openinference.py +++ b/litellm/integrations/otel/mappers/openinference.py @@ -153,7 +153,9 @@ def _shed_order(groups: Mapping[tuple[str, int, int], tuple[str, ...]]) -> tuple _METADATA_KEY: Final = "metadata" -def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> Mapping[str, AttrValue]: +def fit_indexed_messages( + attrs: Mapping[str, AttrValue], budget: int | None, shed_first: frozenset[str] = frozenset() +) -> Mapping[str, AttrValue]: """``attrs`` with indexed message attributes shed, least valuable first, until at most ``budget`` keys remain. ``None`` means the span has no attribute count limit. Every message still rides the ``input.value`` and @@ -161,12 +163,14 @@ def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> after every indexed message attribute: message attributes are the indexed, queryable view (the blobs carry no per-index keys), so a count-squeezed span keeps them and the single metadata key absorbs only the residual shortfall. Shedding happens here, before ``span.set_attribute``, so the fit is exact and - the SDK's dropped-attributes counter never silently masks the choice. + the SDK's dropped-attributes counter never silently masks the choice. ``shed_first`` keys shed one at a + time before anything else, so they never cost the span any other attribute. """ if budget is None or len(attrs) <= budget: return attrs groups: Final = _message_key_groups(attrs) - sheddable: Final = (*(groups[group] for group in _shed_order(groups)),) + leading: Final = tuple((key,) for key in attrs if key in shed_first) + sheddable: Final = (*leading, *(groups[group] for group in _shed_order(groups))) flex: Final = (_METADATA_KEY,) if _METADATA_KEY in attrs else () candidates: Final = (*sheddable, flex) if flex else sheddable running: Final = tuple(accumulate(len(candidate) for candidate in candidates)) diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index 5d37894bb8f..573ba3fd3e7 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -59,6 +59,17 @@ if TYPE_CHECKING: _INTERNAL_CALL_ORIGINS: Final[frozenset[str]] = frozenset(get_args(InternalCallOrigin)) _AUTH_ATTRIBUTE: Final = TypeAdapter(object) +_EXPORTED_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, str]]] = TypeAdapter(Mapping[str, str]) +# The caller identity the Langfuse metadata keys export; the rest of the request metadata stays in the bare +# ``metadata`` attribute. +IDENTITY_METADATA_FIELDS: Final = ( + "user_api_key_alias", + "user_api_key_user_id", + "user_api_key_end_user_id", + "user_api_key_team_id", + "user_api_key_team_alias", + "user_api_key_org_id", +) REQUESTER_METADATA_KEY: Final = "requester_metadata" REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}." @@ -471,6 +482,23 @@ def _model_info_id(model_info: object) -> str | None: return None +def exported_request_metadata(metadata: Mapping[str, object]) -> Mapping[str, str]: + """The non-empty string ``IDENTITY_METADATA_FIELDS`` of ``metadata``, with ``user_api_key_*`` dropped when + ``litellm.redact_user_api_key_info`` is on, as the ``langfuse`` callback does.""" + from litellm.litellm_core_utils.redact_messages import ( + redact_user_api_key_info as _redact_user_api_key_info, # pyright: ignore[reportUnknownVariableType] # helper uses bare dict + ) + + redact_metadata: Final = cast( # cast-ok: adapter validates the helper's output below + Callable[[dict[str, str]], object], + _redact_user_api_key_info, + ) + present: Final = { + field: value for field in IDENTITY_METADATA_FIELDS if isinstance(value := metadata.get(field), str) and value + } + return MappingProxyType(dict(_EXPORTED_METADATA_ADAPTER.validate_python(redact_metadata(present)))) + + def _team_metadata_dict(value: object) -> Mapping[str, object] | None: """The team's free-form metadata as a raw mapping, or ``None`` when missing or empty. diff --git a/litellm/integrations/otel/model/payloads.py b/litellm/integrations/otel/model/payloads.py index d11505ce026..13bf4b53a83 100644 --- a/litellm/integrations/otel/model/payloads.py +++ b/litellm/integrations/otel/model/payloads.py @@ -12,7 +12,12 @@ from urllib.parse import urlsplit from typing_extensions import ReadOnly, TypedDict -from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity, allowlisted_metadata +from litellm.integrations.otel.model.metadata import ( + RequestContext, + RequestIdentity, + allowlisted_metadata, + exported_request_metadata, +) from litellm.integrations.otel.model.semconv import ( GenAIOperation, GenAIOutputType, @@ -417,6 +422,8 @@ class LLMCallSpanData: response_cost: float | None server: ServerInfo | None identity: RequestIdentity + # The caller identity among the promoted metadata plus the team (see ``exported_request_metadata``). + request_metadata: Mapping[str, str] = field(default_factory=lambda: _EMPTY_METADATA) is_streaming: bool | None = None cost: LLMCost = field(default_factory=LLMCost) tools: tuple[ToolDefinition, ...] = () @@ -467,6 +474,7 @@ class LLMCallSpanData: context: Final = RequestContext.from_standard_logging_payload(payload) metadata: Final = payload.get("metadata") decision: Final = metadata["routing_decision"] if metadata and "routing_decision" in metadata else None + promoted_metadata: Final = allowlisted_metadata(context.identity.metadata, metadata_keys) # Normalize ``response`` to a dict once so the content/id reads below are a # plain ``.get`` — no repeated ``isinstance`` guards. raw_response: Final = payload.get("response") @@ -498,7 +506,14 @@ class LLMCallSpanData: cost=LLMCost.from_breakdown(cast("Mapping[str, object] | None", payload.get("cost_breakdown"))), server=None if cache_hit else ServerInfo.from_api_base(context.api_base), identity=context.identity, - promoted_metadata=allowlisted_metadata(context.identity.metadata, metadata_keys), + request_metadata=exported_request_metadata( + { + "user_api_key_team_id": context.identity.team_id, + "user_api_key_team_alias": context.identity.team_alias, + **promoted_metadata, + } + ), + promoted_metadata=promoted_metadata, is_streaming=as_bool(payload.get("stream")), tools=_extract_tools(params), messages_in=_dicts(payload.get("messages")) if capture_content else (), diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 898eab5715b..db56da033a5 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -43,6 +43,7 @@ from litellm._version import version as litellm_version from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_OBSERVATION_INPUT, LANGFUSE_OBSERVATION_OUTPUT, + LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LANGFUSE_TRACE_NAME, ) from litellm.integrations.otel.model.config import ( @@ -373,6 +374,7 @@ class _DrainPool: self._backlog -= 1 +LITELLM_TRACER_NAME: Final = "litellm" _NO_ATTRIBUTES: Final[Mapping[str, AttributeValue]] = MappingProxyType({}) _DB_SYSTEM_KEYS: Final = frozenset({DB.SYSTEM_NAME, DB.SYSTEM_LEGACY}) # Keys on a database span that describe the proxy's own datastore: its host, its @@ -459,6 +461,48 @@ def without_message_content(span: ReadableSpan) -> ReadableSpan: return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent) +class _CountedSpanView(_SpanView): + """A ``_SpanView`` of ``inner`` with other attributes that reports the dropped counts ``inner`` reports.""" + + def __init__(self, inner: ReadableSpan, attributes: Attributes) -> None: + super().__init__(inner, inner.resource, attributes, inner.events, inner.status, parent=inner.parent) + self._dropped_counts: Final = (inner.dropped_attributes, inner.dropped_events, inner.dropped_links) + + @property + def dropped_attributes(self) -> int: + return self._dropped_counts[0] + + @property + def dropped_events(self) -> int: + return self._dropped_counts[1] + + @property + def dropped_links(self) -> int: + return self._dropped_counts[2] + + +def _is_litellm_span(span: ReadableSpan) -> bool: + """Whether LiteLLM's own tracer wrote ``span``, rather than application or instrumentation code.""" + scope: Final = span.instrumentation_scope + return scope is not None and scope.name == LITELLM_TRACER_NAME + + +def without_langfuse_request_metadata(span: ReadableSpan) -> ReadableSpan: + """``span`` without the request metadata the Langfuse mapper writes for a Langfuse destination. + + Only LiteLLM's own spans lose it: any other span keeps every key its own code set. + """ + attributes: Final = span.attributes or _NO_ATTRIBUTES + if not (_is_litellm_span(span) and any(key in LANGFUSE_REQUEST_METADATA_ATTRIBUTES for key in attributes)): + return span + return _CountedSpanView( + span, + MappingProxyType( + {key: value for key, value in attributes.items() if key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES} + ), + ) + + def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool: return any(key in attributes for key in _DB_SYSTEM_KEYS) @@ -614,12 +658,16 @@ def _for_destination( captures on the server span, whose request side holds the caller's bearer token, and the query string of the request URL, which can hold the same key. A destination whose effective setting disables span content loses the prompt and - response content too. The span itself stays, so the tenant still gets the whole - trace tree. + response content too, and a destination for any backend but Langfuse loses the + request metadata the Langfuse mapper writes on LiteLLM's own spans. The span itself + stays, so the tenant still gets the whole trace tree. """ extra: Final = destination.resource_attributes defaults: Final = destination.resource_defaults - attributes: Final = span.attributes or _NO_ATTRIBUTES + source: Final = ( + span if destination.callback_name == ExporterOwner.LANGFUSE_OTEL else without_langfuse_request_metadata(span) + ) + attributes: Final = source.attributes or _NO_ATTRIBUTES database: Final = _is_database_span(attributes) owned: Final = _is_tenant_owned_span(attributes) unreachable: Final = _guardrail_unreachable(attributes) @@ -631,14 +679,14 @@ def _for_destination( if _tenant_visible(key, database, owned, unreachable) and not (redacted and _is_message_content(key)) } ) - recorded: Final = span.events + recorded: Final = source.events events: Final = tuple(_without_stack_trace(event) for event in recorded) if owned else () unchanged: Final = owned and _same_attributes(kept, attributes) and all(a is b for a, b in zip(events, recorded)) if not extra and not defaults and unchanged: - return span - resource: Final = _tenant_resource(span.resource, defaults, extra) - status: Final = span.status if owned else Status(span.status.status_code) - return _SpanView(span, resource, kept, events, status, parent=span.parent) + return source + resource: Final = _tenant_resource(source.resource, defaults, extra) + status: Final = source.status if owned else Status(source.status.status_code) + return _SpanView(source, resource, kept, events, status, parent=source.parent) class TenantFanOutSpanProcessor(SpanProcessor): @@ -1106,6 +1154,25 @@ class _OverriddenBackendFilter(SpanProcessor): return self._inner.force_flush(timeout_millis) +class _LangfuseRequestMetadataFilter(SpanProcessor): + """Strip the Langfuse request metadata from LiteLLM's own spans before ``inner`` exports them.""" + + def __init__(self, inner: SpanProcessor) -> None: + self._inner: Final = inner + + def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: + self._inner.on_start(span, parent_context) + + def on_end(self, span: ReadableSpan) -> None: + self._inner.on_end(without_langfuse_request_metadata(span)) + + def shutdown(self) -> None: + self._inner.shutdown() + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return self._inner.force_flush(timeout_millis) + + class _MessageContentFilter(SpanProcessor): """Strip prompt and response content from every span before ``inner`` exports it.""" @@ -1377,13 +1444,17 @@ def build_resource(config: OpenTelemetryV2Config) -> Resource: def _spec_processor( - spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool + spec: ExporterSpec, + use_simple_processor: bool | None, + strip_message_content: bool, + strip_langfuse_request_metadata: bool, ) -> SpanProcessor: exporting: Final = _processor_for( _exporter_from_spec(spec), (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor), ) - return _MessageContentFilter(exporting) if strip_message_content else exporting + scoped: Final = _LangfuseRequestMetadataFilter(exporting) if strip_langfuse_request_metadata else exporting + return _MessageContentFilter(scoped) if strip_message_content else scoped def build_tracer_provider( @@ -1411,6 +1482,8 @@ def build_tracer_provider( ``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel`` alone; a collector or any other backend in the same config keeps the full tree. + In a config that writes the Langfuse vocabulary, every exporter but ``langfuse_otel``'s + goes without the request metadata the Langfuse mapper writes on LiteLLM's own spans. ``strip_message_content`` removes prompt and response content before every exporter, for a routed provider whose spans can carry content another destination asked for. @@ -1426,10 +1499,18 @@ def build_tracer_provider( # ``config._normalize`` guarantees at least one spec (it folds the top-level # ``exporter``/``endpoint``/``headers`` fields in when ``exporters`` is empty). + writes_langfuse: Final = "langfuse" in config.mapper_names or any( + spec.owner is ExporterOwner.LANGFUSE_OTEL for spec in config.exporters + ) for spec in config.exporters: if spec.requires_headers and not spec.headers: continue - processor = _spec_processor(spec, use_simple_processor, strip_message_content) + processor = _spec_processor( + spec, + use_simple_processor, + strip_message_content, + strip_langfuse_request_metadata=writes_langfuse and spec.owner is not ExporterOwner.LANGFUSE_OTEL, + ) owner = spec.owner.value if tenant_overrides and spec.owner is not None else None scope = _operator_scope(config, spec) sink = _sink_key(spec.endpoint, parse_headers(spec.headers)) if _exports_to_the_wire(spec) else None @@ -1577,7 +1658,7 @@ def _attached_processors(provider: trace.TracerProvider) -> "tuple[SpanProcessor return tuple(getattr(multi, "_span_processors", ())) -def get_tracer(provider: TracerProvider, name: str = "litellm") -> Tracer: +def get_tracer(provider: TracerProvider, name: str = LITELLM_TRACER_NAME) -> Tracer: # Stamp the instrumentation scope with the LiteLLM package version so every # emitted span carries a deterministic ``scope.version`` (the standard OTel # location for the emitting library's version) for downstream consumers. diff --git a/tests/integration/observability/test_langfuse_otel_metadata.py b/tests/integration/observability/test_langfuse_otel_metadata.py new file mode 100644 index 00000000000..46fca44535f --- /dev/null +++ b/tests/integration/observability/test_langfuse_otel_metadata.py @@ -0,0 +1,1492 @@ +import base64 +import binascii +import json +import threading +import uuid +from collections import Counter +from collections.abc import Callable, Iterator, Mapping, Sequence +from concurrent.futures import ThreadPoolExecutor, wait +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import httpx +import pytest +import yaml +from anthropic import Anthropic, AsyncAnthropic +from integration._support.client import Gateway, Scenario, eventually +from integration._support.process import OwnedProxy, owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from openai import AsyncOpenAI, OpenAI +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest + +_IDENTITIES: Final = ( + "user_api_key_alias", + "user_api_key_user_id", + "user_api_key_end_user_id", + "user_api_key_team_id", + "user_api_key_team_alias", +) +# The only request metadata the Langfuse metadata keys carry; everything else stays in the bare ``metadata``. +_EXPORTED_FIELDS: Final = frozenset({*_IDENTITIES, "user_api_key_org_id"}) + + +@dataclass(frozen=True, slots=True) +class _Rig: + candidate: Gateway + provider: Wire + collector: Wire + proxy: OwnedProxy + + +def _span_attributes(body: bytes) -> tuple[dict[str, object], ...]: + request: Final = ExportTraceServiceRequest.FromString(body) + return tuple( + {attribute.key: getattr(attribute.value, attribute.value.WhichOneof("value")) for attribute in span.attributes} + for resource in request.resource_spans + for scope in resource.scope_spans + for span in scope.spans + ) + + +def _sse_chat(response_id: str) -> tuple[bytes, ...]: + chunk: Final = { + "id": response_id, + "object": "chat.completion.chunk", + "created": 1, + "model": "gpt-4o-mini", + } + return ( + f"data: {json.dumps({**chunk, 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': 'langfuse-echo'}, 'finish_reason': None}]})}\n\n".encode(), + f"data: {json.dumps({**chunk, 'choices': [{'index': 0, 'delta': {}, 'finish_reason': 'stop'}]})}\n\n".encode(), + f"data: {json.dumps({**chunk, 'choices': [], 'usage': {'prompt_tokens': 3, 'completion_tokens': 2, 'total_tokens': 5}})}\n\n".encode(), + b"data: [DONE]\n\n", + ) + + +def _sse_messages(response_id: str) -> tuple[bytes, ...]: + def event(name: str, payload: Mapping[str, object]) -> bytes: + return f"event: {name}\ndata: {json.dumps(payload)}\n\n".encode() + + message: Final = { + "id": response_id, + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-6", + "content": [], + "stop_reason": None, + "usage": {"input_tokens": 3, "output_tokens": 1}, + } + return ( + event("message_start", {"type": "message_start", "message": message}), + event( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + event( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "langfuse-echo"}}, + ), + event("content_block_stop", {"type": "content_block_stop", "index": 0}), + event( + "message_delta", + {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 2}}, + ), + event("message_stop", {"type": "message_stop"}), + ) + + +def _sse_responses(response_id: str) -> tuple[bytes, ...]: + def event(name: str, payload: Mapping[str, object]) -> bytes: + return f"event: {name}\ndata: {json.dumps(payload)}\n\n".encode() + + response: Final = { + "id": response_id, + "object": "response", + "model": "gpt-4o-mini", + "status": "completed", + "output": [ + { + "type": "message", + "id": f"item-{response_id}", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "langfuse-echo"}], + } + ], + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5}, + } + return ( + event( + "response.created", + {"type": "response.created", "response": {**response, "status": "in_progress", "output": []}}, + ), + event( + "response.output_text.delta", + { + "type": "response.output_text.delta", + "response_id": response_id, + "item_id": f"item-{response_id}", + "output_index": 0, + "content_index": 0, + "delta": "langfuse-echo", + }, + ), + event("response.completed", {"type": "response.completed", "response": response}), + ) + + +def _chat_body(response_id: str) -> dict[str, object]: + return { + "id": response_id, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": "langfuse-echo"}, "finish_reason": "stop"} + ], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}, + } + + +def _message_body(response_id: str) -> dict[str, object]: + return { + "id": response_id, + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-6", + "content": [{"type": "text", "text": "langfuse-echo"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 3, "output_tokens": 2}, + } + + +_RESPONSE_FIELDS: Final = { + "created_at": 1, + "error": None, + "incomplete_details": None, + "instructions": None, + "metadata": {}, + "parallel_tool_calls": True, + "temperature": 1.0, + "tool_choice": "auto", + "tools": [], + "top_p": 1.0, + "max_output_tokens": None, + "previous_response_id": None, + "store": True, + "truncation": "disabled", + "user": None, +} + + +def _response_body(response_id: str) -> dict[str, object]: + return { + **_RESPONSE_FIELDS, + "id": response_id, + "object": "response", + "model": "gpt-4o-mini", + "status": "completed", + "output": [ + { + "type": "message", + "id": f"item-{response_id}", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "langfuse-echo"}], + } + ], + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5}, + } + + +@contextmanager +def _langfuse_rig( + gateway: Gateway, + tmp_path: Path, + settings: Mapping[str, object], + marker: str, + *, + sink_reply: Callable[[Request], Reply] | None = None, + overrides: Mapping[str, str] | None = None, + model_list: Callable[[str], Sequence[Mapping[str, object]]] | None = None, +) -> Iterator[_Rig]: + def upstream(request: Request) -> Reply: + if request.method == "GET": + return Reply(body=b'{"data":[]}') + body: Final = json.loads(request.body or b"{}") + suffix: Final = uuid.uuid4().hex[:8] + if "-fail-" in json.dumps(body): + return Reply( + status=500, body=b'{"error": {"message": "scripted provider failure", "type": "server_error"}}' + ) + if request.target.endswith("/chat/completions"): + chat_id: Final = f"chatcmpl-{marker}-{suffix}" + if body.get("stream"): + return Reply(content_type="text/event-stream", chunks=_sse_chat(chat_id)) + return Reply(body=json.dumps(_chat_body(chat_id)).encode()) + if request.target.endswith("/messages"): + message_id: Final = f"msg-{marker}-{suffix}" + if body.get("stream"): + return Reply(content_type="text/event-stream", chunks=_sse_messages(message_id)) + return Reply(body=json.dumps(_message_body(message_id)).encode()) + if request.target.endswith("/responses"): + resp_id: Final = f"resp-{marker}-{suffix}" + if body.get("stream"): + return Reply(content_type="text/event-stream", chunks=_sse_responses(resp_id)) + return Reply(body=json.dumps(_response_body(resp_id)).encode()) + raise AssertionError(request.target) + + def sink(request: Request) -> Reply: + return sink_reply(request) if sink_reply is not None else Reply() + + with wire_server(upstream) as provider, wire_server(sink) as collector: + raw_config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config: Final = { + **raw_config, + "model_list": [*raw_config.get("model_list", []), *(model_list(provider.url) if model_list else ())], + "litellm_settings": {**raw_config["litellm_settings"], "callbacks": ["langfuse_otel"], **settings}, + } + path: Final = tmp_path / "langfuse_otel.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy_process( + gateway, + tmp_path, + { + "LANGFUSE_PUBLIC_KEY": "pk-lf-integration", + "LANGFUSE_SECRET_KEY": "sk-lf-integration", + "LANGFUSE_HOST": collector.url, + "LITELLM_OTEL_V2": "1", + **{ + name: (collector.url if value == "__collector__" else value) + for name, value in dict(overrides or {}).items() + }, + }, + config=path, + workers=2, + ) as owned: + yield _Rig(owned.gateway, provider, collector, owned) + + +def _response_id(value: object) -> str | None: + if not isinstance(value, str): + return None + if value.startswith("resp_"): + try: + decoded: Final = base64.b64decode(value[5:] + "===").decode() + except (binascii.Error, UnicodeDecodeError): + return value + _, _, suffix = decoded.partition("response_id:") + if suffix: + return suffix.split(";", 1)[0] + return value + + +def _span_response_id(span: Mapping[str, object]) -> str | None: + return _response_id(span.get("gen_ai.response.id")) + + +def _watched_spans(collector: Wire) -> Callable[[], tuple[dict[str, object], ...]]: + drained: list[ + dict[str, object] + ] = [] # mutable-ok: drain() consumes the wire queue, so observed spans must accumulate across eventually polls + + def snapshot() -> tuple[dict[str, object], ...]: + drained.extend(attributes for batch in collector.drain() for attributes in _span_attributes(batch.body)) + return tuple(drained) + + return snapshot + + +def _assert_key_alias(spans: Sequence[Mapping[str, object]], key_alias: str | None) -> None: + assert spans, "no span landed" + for span in spans: + assert json.loads(str(span["langfuse.observation.metadata"]))["user_api_key_alias"] == key_alias, span + + +def _v2_llm_span(collector: Wire, response_id: str) -> dict[str, object]: + def matches(span: Mapping[str, object]) -> bool: + return _response_id(span.get("gen_ai.response.id")) == _response_id(response_id) + + observed: Final = eventually( + _watched_spans(collector), lambda spans: sum(matches(span) for span in spans) == 1, seconds=25 + ) + return next(span for span in observed if matches(span)) + + +def _scenario_identity( + scenario: Scenario, + provider: Wire, + *, + spend_logs_metadata: Mapping[str, object] | None = None, + team: bool = True, +) -> tuple[str, dict[str, str | None]]: + team_alias: Final = ("lf-team-" + uuid.uuid4().hex) if team else None + team_id: Final = scenario.team(team_alias=team_alias) if team else None + key_alias: Final = "lf-key-" + uuid.uuid4().hex + key: Final = scenario.key( + **({"team_id": team_id} if team_id else {}), + key_alias=key_alias, + metadata={"spend_logs_metadata": spend_logs_metadata or {"ticket": "LIT-8283"}}, + ) + model: Final = scenario.model(api_base=provider.url + "/v1") + end_user: Final = "lf-end-user-" + uuid.uuid4().hex + return key, { + "key_alias": key_alias, + "team_id": team_id, + "team_alias": team_alias, + "end_user": end_user, + "model": model, + } + + +def _assert_identity(attrs: Mapping[str, object], expected: Mapping[str, object]) -> None: + assert "langfuse.observation.metadata" in attrs, sorted(attrs) + observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) + assert {key: observation.get(key) for key in expected} == dict(expected), observation + assert set(observation) <= _EXPORTED_FIELDS, observation + assert all(isinstance(value, str) and value for value in observation.values()), observation + flattened: Final = { + attribute: attrs.get(attribute) + for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES) + if attribute in attrs + } + assert flattened == {f"langfuse.trace.metadata.{field}": value for field, value in expected.items()}, attrs + + +def _chat_identity_fields(identity: Mapping[str, str | None]) -> dict[str, object]: + return { + "user_api_key_alias": identity["key_alias"], + **({"user_api_key_team_id": identity["team_id"]} if identity["team_id"] else {}), + **({"user_api_key_team_alias": identity["team_alias"]} if identity["team_alias"] else {}), + **({"user_api_key_end_user_id": identity["end_user"]} if identity["end_user"] else {}), + } + + +def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-sdkstream-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + with OpenAI( + base_url=str(rig.candidate.client.base_url).rstrip("/") + "/v1", + api_key=key, + http_client=httpx.Client(trust_env=False, timeout=30), + ) as client: + chunks: Final = tuple( + client.chat.completions.create( + model=identity["model"], + messages=[{"role": "user", "content": marker}], + stream=True, + user=identity["end_user"], + extra_body={"cache": {"no-cache": True}}, + ) + ) + assert chunks, "stream produced no chunks" + response_id: Final = chunks[-1].id + attrs: Final = _v2_llm_span(rig.collector, response_id) + _assert_identity(attrs, _chat_identity_fields(identity)) + + +async def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_async(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-sdkasync-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + async with AsyncOpenAI( + base_url=str(rig.candidate.client.base_url).rstrip("/") + "/v1", + api_key=key, + http_client=httpx.AsyncClient(trust_env=False, timeout=30), + ) as client: + response: Final = await client.chat.completions.create( + model=identity["model"], + messages=[{"role": "user", "content": marker}], + user=identity["end_user"], + extra_body={"cache": {"no-cache": True}}, + ) + attrs: Final = _v2_llm_span(rig.collector, response.id) + _assert_identity(attrs, _chat_identity_fields(identity)) + + +def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_messages(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-anthropic-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url) + with Anthropic( + base_url=str(rig.candidate.client.base_url).rstrip("/"), + api_key=key, + http_client=httpx.Client(trust_env=False, timeout=30), + ) as client: + message: Final = client.messages.create( + model=model, + max_tokens=64, + messages=[{"role": "user", "content": marker}], + extra_body={"litellm_metadata": {"tags": ["lf8283"]}, "metadata": {"user_id": identity["end_user"]}}, + ) + attrs: Final = _v2_llm_span(rig.collector, message.id) + _assert_identity(attrs, _chat_identity_fields(identity)) + + +async def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_stream( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-anthstream-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url) + async with ( + AsyncAnthropic( + base_url=str(rig.candidate.client.base_url).rstrip("/"), + api_key=key, + http_client=httpx.AsyncClient(trust_env=False, timeout=30), + ) as client, + client.messages.stream( + model=model, + max_tokens=64, + messages=[{"role": "user", "content": marker}], + extra_body={"metadata": {"user_id": identity["end_user"]}}, + ) as stream, + ): + final: Final = await stream.get_final_message() + attrs: Final = _v2_llm_span(rig.collector, final.id) + _assert_identity(attrs, _chat_identity_fields(identity)) + + +def test_langfuse_otel_metadata_reaches_langfuse_for_responses_api(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-responses-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + client: Final = OpenAI( + base_url=str(rig.candidate.client.base_url).rstrip("/") + "/v1", + api_key=key, + http_client=httpx.Client(trust_env=False, timeout=30), + ) + response: Final = client.responses.create( + model=identity["model"], + input=marker, + user=identity["end_user"], + extra_body={"cache": {"no-cache": True}}, + ) + attrs: Final = _v2_llm_span(rig.collector, response.id) + _assert_identity(attrs, _chat_identity_fields(identity)) + + +def test_langfuse_otel_metadata_reaches_langfuse_for_responses_sse_stream(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-respstream-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + with rig.candidate.client.stream( + "POST", + "/v1/responses", + json={ + "model": identity["model"], + "input": marker, + "user": identity["end_user"], + "stream": True, + "cache": {"no-cache": True}, + }, + headers={"Authorization": f"Bearer {key}"}, + ) as stream: + events: Final = tuple( + json.loads(line.removeprefix("data: ")) + for line in stream.iter_lines() + if line.startswith("data: ") and line != "data: [DONE]" + ) + response_id: Final = next( + event["response_id"] for event in events if event.get("type") == "response.output_text.delta" + ) + attrs: Final = _v2_llm_span(rig.collector, response_id) + _assert_identity(attrs, _chat_identity_fields(identity)) + + +def test_langfuse_otel_failure_span_carries_request_metadata(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-fail-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker} scripted -fail- upstream"}], + "user": identity["end_user"], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code >= 500, response.text + + def failed_calls(spans: Sequence[Mapping[str, object]]) -> tuple[Mapping[str, object], ...]: + return tuple( + span + for span in spans + if span.get("langfuse.trace.metadata.team_id") == identity["team_id"] and "error.type" in span + ) + + failed: Final = failed_calls( + eventually(_watched_spans(rig.collector), lambda spans: bool(failed_calls(spans)), seconds=25) + ) + for attrs in failed: + _assert_identity(attrs, _chat_identity_fields(identity)) + + +def test_langfuse_otel_oversized_caller_metadata_is_not_copied_into_the_langfuse_keys( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-bigmeta-" + uuid.uuid4().hex + spend_logs_metadata: Final = { + "blob": "x" * 5000, + "count": 7, + "steps": ["a", "b"], + "nested": {"inner": {"leaf": True}}, + } + caller_note: Final = "y" * 50000 + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider, spend_logs_metadata=spend_logs_metadata) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "user": identity["end_user"], + "metadata": {"caller_note": caller_note}, + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + attrs: Final = _v2_llm_span(rig.collector, response.json()["id"]) + _assert_identity(attrs, _chat_identity_fields(identity)) + assert len(str(attrs["langfuse.observation.metadata"])) < 1000, len(str(attrs["langfuse.observation.metadata"])) + langfuse_metadata: Final = json.dumps( + { + name: value + for name, value in attrs.items() + if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.") + } + ) + assert caller_note not in langfuse_metadata, len(langfuse_metadata) + assert spend_logs_metadata["blob"] not in langfuse_metadata, len(langfuse_metadata) + + +def test_langfuse_otel_unauthenticated_request_leaks_no_identity(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-unauth-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + denied: Final = rig.candidate.client.post( + "/v1/chat/completions", + json={ + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker}-denied"}], + }, + headers={}, + ) + assert denied.status_code == 401, denied.text + watch: Final = _watched_spans(rig.collector) + first: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert first.status_code == 200, first.text + first_id: Final = first.json()["id"] + eventually(watch, lambda spans: sum(_span_response_id(s) == first_id for s in spans) == 1, seconds=25) + second: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker + "-settle"}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert second.status_code == 200, second.text + second_id: Final = second.json()["id"] + settled: Final = eventually( + watch, lambda spans: sum(_span_response_id(s) == second_id for s in spans) == 1, seconds=25 + ) + carriers: Final = tuple( + _span_response_id(span) + for span in settled + if "langfuse.observation.metadata" in span + or any(name.startswith("langfuse.trace.metadata.user_api_key_") for name in span) + ) + assert sorted(carriers, key=str) == sorted((first_id, second_id)), carriers + + +def test_langfuse_otel_repeated_identical_requests_emit_one_span_each(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-repeat-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + watch: Final = _watched_spans(rig.collector) + responses: Final = tuple( + rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + }, + key=key, + ) + for _ in range(3) + ) + assert all(response.status_code == 200 for response in responses), [r.text for r in responses] + response_ids: Final = tuple(response.json()["id"] for response in responses) + assert len(set(response_ids)) == 3, response_ids + settled: Final = eventually( + watch, + lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in response_ids), + seconds=40, + ) + for rid in response_ids: + assert sum(_span_response_id(s) == rid for s in settled) == 1 + + +def test_langfuse_otel_sink_outage_mid_burst_lands_every_response_id_once(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-chaos-outage-" + uuid.uuid4().hex + outage: Final = threading.Event() + rejected: Final[list[bytes]] = [] # mutable-ok: the wire server thread records each batch it answers 503 + + def sink_reply(request: Request) -> Reply: + if outage.is_set(): + rejected.append(request.body) + return Reply(status=503, body=b'{"error": "sink outage"}') + return Reply() + + with ( + _langfuse_rig(gateway, tmp_path, {}, marker, sink_reply=sink_reply) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + anthropic_model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url) + received: Final[list[bytes]] = [] # mutable-ok: drain() consumes the wire queue, so batches accumulate + + def watch() -> tuple[dict[str, object], ...]: + # The wire also records the batches the sink rejected; only the accepted ones landed. + received.extend(batch.body for batch in rig.collector.drain()) + accepted: Final = Counter(received) - Counter(rejected) + return tuple(attributes for body in accepted.elements() for attributes in _span_attributes(body)) + + def call(index: int) -> str: + path: Final = ("/v1/chat/completions", "/v1/messages", "/v1/responses")[index % 3] + streaming: Final = index % 2 == 0 + body: Final = { + "/v1/chat/completions": { + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker}-{index}"}], + "stream": streaming, + "cache": {"no-cache": True}, + }, + "/v1/messages": { + "model": anthropic_model, + "messages": [{"role": "user", "content": f"{marker}-{index}"}], + "max_tokens": 64, + "stream": streaming, + }, + "/v1/responses": {"model": identity["model"], "input": f"{marker}-{index}", "stream": streaming}, + }[path] + if streaming: + with rig.candidate.client.stream( + "POST", path, json=body, headers={"Authorization": f"Bearer {key}"} + ) as stream: + assert stream.status_code == 200, f"burst {index}: {stream.status_code}" + events: Final = tuple( + line.removeprefix("data: ") + for line in stream.iter_lines() + if line.startswith("data: ") and not line.endswith("[DONE]") + ) + payloads: Final = tuple(json.loads(line) for line in events) + found: Final = next( + payload.get("response_id") or (payload.get("message") or {}).get("id") or payload.get("id") + for payload in payloads + if payload.get("response_id") or payload.get("message") or payload.get("id") + ) + return str(found) + response: Final = rig.candidate.client.post(path, json=body, headers={"Authorization": f"Bearer {key}"}) + assert response.status_code == 200, f"burst {index}: {response.text}" + return response.json()["id"] + + with ThreadPoolExecutor(max_workers=30) as pool: + futures: Final = tuple(pool.submit(call, index) for index in range(30)) + while sum(f.done() for f in futures) < 10: + wait(futures, timeout=0.05) + outage.set() + eventually( + lambda: (sum(f.done() for f in futures), len(rejected)), + lambda progress: progress[0] >= 25 and progress[1] >= 1, + seconds=20, + ) + outage.clear() + response_ids: Final = tuple(f.result() for f in futures) + assert len(set(response_ids)) == 30, response_ids + settled: Final = eventually( + watch, + lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in response_ids), + seconds=80, + return_last_on_timeout=True, + ) + landed: Final = sorted( + str(_span_response_id(span)) for span in settled if _span_response_id(span) in response_ids + ) + assert landed == sorted(response_ids), sorted(set(response_ids) - set(landed)) + assert len(rejected) >= 1, "sink outage never served a 503" + _assert_key_alias([span for span in settled if _span_response_id(span) in response_ids], identity["key_alias"]) + + +def test_langfuse_otel_worker_kill_mid_burst_keeps_serving_and_exports(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-chaos-kill-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + watch: Final = _watched_spans(rig.collector) + + def call(index: int) -> tuple[int, str]: + try: + response: Final = rig.candidate.client.post( + "/v1/chat/completions", + json={ + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker}-{index}"}], + "cache": {"no-cache": True}, + }, + headers={"Authorization": f"Bearer {key}"}, + timeout=30, + ) + except httpx.HTTPError as error: + return 0, repr(error) + return response.status_code, response.json()["id"] if response.status_code == 200 else response.text + + workers: Final = rig.proxy.worker_pids() + assert len(workers) >= 2, f"expected multiple uvicorn workers, got {sorted(workers)!r}" + victim: Final = min(workers) + with ThreadPoolExecutor(max_workers=20) as pool: + futures: Final = tuple(pool.submit(call, index) for index in range(20)) + while sum(f.done() for f in futures) < 5: + wait(futures, timeout=0.05) + replacement: Final = rig.proxy.kill_worker(victim) + outcomes: Final = tuple(f.result() for f in futures) + assert replacement != victim + succeeded: Final = tuple(identifier for status, identifier in outcomes if status == 200) + assert succeeded, outcomes + post_kill: Final = tuple(call(20 + index) for index in range(5)) + post_kill_ids: Final = tuple(identifier for _, identifier in post_kill) + assert all(status == 200 for status, _ in post_kill), post_kill + settled: Final = eventually( + watch, + lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in post_kill_ids), + seconds=60, + ) + for rid in succeeded: + assert sum(_span_response_id(s) == rid for s in settled) <= 1, rid + _assert_key_alias([span for span in settled if _span_response_id(span) in post_kill_ids], identity["key_alias"]) + + +def test_langfuse_otel_slow_sink_does_not_deadlock_exports(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-chaos-slow-" + uuid.uuid4().hex + + def sink_reply(request: Request) -> Reply: + gate: Final = threading.Event() + threading.Timer(2, gate.set).start() + return Reply(chunks=(b"{}",), gate_after_first=gate) + + with ( + _langfuse_rig(gateway, tmp_path, {}, marker, sink_reply=sink_reply) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + watch: Final = _watched_spans(rig.collector) + responses: Final = tuple( + rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker}-{index}"}], + "cache": {"no-cache": True}, + }, + key=key, + ) + for index in range(10) + ) + assert all(response.status_code == 200 for response in responses), [r.text for r in responses] + response_ids: Final = tuple(response.json()["id"] for response in responses) + settled: Final = eventually( + watch, + lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in response_ids), + seconds=60, + ) + assert sorted(_span_response_id(span) for span in settled if _span_response_id(span) in response_ids) == sorted( + response_ids + ) + _assert_key_alias([span for span in settled if _span_response_id(span) in response_ids], identity["key_alias"]) + + +def test_langfuse_otel_v2_mapper_emits_request_metadata_and_identity(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-v2-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {}, + marker, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "user": identity["end_user"], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + watch: Final = _watched_spans(rig.collector) + observed: Final = eventually( + watch, + lambda spans: sum(s.get("gen_ai.response.id") == response.json()["id"] for s in spans) == 1, + seconds=25, + ) + attrs: Final = next(s for s in observed if s.get("gen_ai.response.id") == response.json()["id"]) + assert attrs.get("langfuse.trace.metadata.team_id") == identity["team_id"], attrs + assert attrs.get("langfuse.trace.metadata.team_alias") == identity["team_alias"], attrs + assert "langfuse.observation.metadata" in attrs, sorted(attrs) + observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) + expected: Final = _chat_identity_fields(identity) + assert {key: observation.get(key) for key in expected} == dict(expected), observation + assert set(observation) <= _EXPORTED_FIELDS, observation + flattened: Final = { + attribute: attrs.get(attribute) + for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES) + if attribute in attrs + } + assert flattened == {f"langfuse.trace.metadata.{field}": value for field, value in expected.items()}, attrs + + +def test_langfuse_otel_v2_mapper_redacts_user_api_key_fields(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-v2-redact-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {"redact_user_api_key_info": True}, + marker, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key: Final = scenario.key( + key_alias="lf-v2-redact-" + uuid.uuid4().hex, + metadata={"spend_logs_metadata": {"ticket": "LIT-8283"}}, + ) + model: Final = scenario.model(api_base=rig.provider.url + "/v1") + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + watch: Final = _watched_spans(rig.collector) + observed: Final = eventually( + watch, + lambda spans: sum(s.get("gen_ai.response.id") == response.json()["id"] for s in spans) == 1, + seconds=25, + ) + attrs: Final = next(s for s in observed if s.get("gen_ai.response.id") == response.json()["id"]) + assert "langfuse.observation.metadata" not in attrs, attrs["langfuse.observation.metadata"] + assert not [name for name in attrs if name.startswith("langfuse.trace.metadata.user_api_key_")], sorted(attrs) + + +def test_langfuse_otel_v2_mapper_teamless_key_emits_no_team_identity(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-v2-noteam-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {}, + marker, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider, team=False) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + watch: Final = _watched_spans(rig.collector) + observed: Final = eventually( + watch, + lambda spans: sum(s.get("gen_ai.response.id") == response.json()["id"] for s in spans) == 1, + seconds=25, + ) + attrs: Final = next(s for s in observed if s.get("gen_ai.response.id") == response.json()["id"]) + assert "langfuse.observation.metadata" in attrs, sorted(attrs) + observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) + for field in ("user_api_key_team_id", "user_api_key_team_alias", "user_api_key_end_user_id"): + assert field not in observation, observation + assert set(observation) <= _EXPORTED_FIELDS, observation + assert { + attribute: attrs.get(attribute) + for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES) + if attribute in attrs + } == {"langfuse.trace.metadata.user_api_key_alias": identity["key_alias"]}, attrs + + +def _scalar_requester_metadata_request(route: str, model: str, marker: str) -> dict[str, object]: + caller: Final = {"litellm_metadata": {"requester_metadata": 1}} + if route == "/v1/messages": + return {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": marker}], **caller} + return {"model": model, "input": marker, **caller} + + +@pytest.mark.parametrize( + ("callback", "route"), + [ + ("otel", "/v1/messages"), + ("otel", "/v1/responses"), + ("langfuse_otel", "/v1/messages"), + ("langfuse_otel", "/v1/responses"), + ], +) +def test_otel_v2_scalar_requester_metadata_keeps_the_llm_span( + callback: str, route: str, gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-v2-scalar-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {"callbacks": [callback]}, + marker, + overrides={ + **({"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": "__collector__"} if callback == "otel" else {}), + }, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + model: Final = ( + scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url) + if route == "/v1/messages" + else identity["model"] + ) + response: Final = rig.candidate.request( + "POST", route, _scalar_requester_metadata_request(route, str(model), marker), key=key + ) + assert response.status_code == 200, response.text + + attrs: Final = _v2_llm_span(rig.collector, response.json()["id"]) + assert attrs.get("gen_ai.usage.input_tokens") == 3, attrs + if callback == "langfuse_otel": + observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) + assert "requester_metadata" not in observation, observation + assert observation["user_api_key_alias"] == identity["key_alias"], observation + + +@pytest.mark.parametrize( + "callbacks", + [["langfuse_otel", "otel"], ["langfuse_otel"]], + ids=["langfuse-listed-before-otel", "langfuse-with-otel-env"], +) +def test_otel_v2_collector_beside_langfuse_gets_no_langfuse_request_metadata( + callbacks: list[str], gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-v2-generic-" + uuid.uuid4().hex + probe: Final = "lf-probe-" + uuid.uuid4().hex + with ( + wire_server(lambda _request: Reply()) as generic, + _langfuse_rig( + gateway, + tmp_path, + {"callbacks": callbacks}, + marker, + overrides={"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url}, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "metadata": {"caller_note": probe}, + "cache": {"no-cache": True}, + }, + key=key, + headers={"x-lit8283-probe": probe}, + ) + assert response.status_code == 200, response.text + + langfuse: Final = _v2_llm_span(rig.collector, response.json()["id"]) + observation: Final = json.loads(str(langfuse["langfuse.observation.metadata"])) + assert observation["user_api_key_alias"] == identity["key_alias"], observation + assert langfuse["langfuse.trace.metadata.user_api_key_alias"] == identity["key_alias"], langfuse + assert probe not in json.dumps(langfuse, default=str), langfuse + + collector: Final = _v2_llm_span(generic, response.json()["id"]) + assert "langfuse.observation.metadata" not in collector, sorted(collector) + assert not [name for name in collector if name.startswith("langfuse.trace.metadata.user_api_key_")], sorted( + collector + ) + assert probe not in json.dumps(collector, default=str), collector + assert collector["langfuse.trace.metadata.team_id"] == identity["team_id"], collector + assert collector.get("gen_ai.usage.input_tokens") == 3, collector + + +@pytest.mark.parametrize("route", ["/v1/chat/completions", "/v1/messages", "/v1/responses"]) +def test_otel_v2_langfuse_gets_no_request_headers_or_caller_metadata( + route: str, gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-v2-probe-" + uuid.uuid4().hex + probe: Final = "lf-probe-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + model: Final = ( + scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url) + if route == "/v1/messages" + else identity["model"] + ) + body: Final = { + "/v1/chat/completions": { + "model": model, + "messages": [{"role": "user", "content": marker}], + "metadata": {"caller_note": probe}, + "cache": {"no-cache": True}, + }, + "/v1/messages": { + "model": model, + "max_tokens": 16, + "messages": [{"role": "user", "content": marker}], + "litellm_metadata": {"caller_note": probe}, + }, + "/v1/responses": {"model": model, "input": marker, "litellm_metadata": {"caller_note": probe}}, + }[route] + watch: Final = _watched_spans(rig.collector) + response: Final = rig.candidate.request("POST", route, body, key=key, headers={"x-lit8283-probe": probe}) + assert response.status_code == 200, response.text + settle: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert settle.status_code == 200, settle.text + + def llm_span(spans: Sequence[Mapping[str, object]], response_id: str) -> Mapping[str, object] | None: + return next( + (span for span in spans if _response_id(span.get("gen_ai.response.id")) == _response_id(response_id)), + None, + ) + + exported: Final = eventually( + watch, + lambda spans: ( + llm_span(spans, response.json()["id"]) is not None and llm_span(spans, settle.json()["id"]) is not None + ), + seconds=25, + ) + attrs: Final = llm_span(exported, response.json()["id"]) + assert attrs is not None + observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) + assert observation["user_api_key_alias"] == identity["key_alias"], observation + assert set(observation) <= _EXPORTED_FIELDS, observation + assert not [span for span in exported if probe in json.dumps(span, default=str)], exported + + +def _keyword_router(router: str, keyword: str) -> Callable[[str], Sequence[Mapping[str, object]]]: + def models(provider_url: str) -> Sequence[Mapping[str, object]]: + tier: Final = f"{router}-tier" + return ( + { + "model_name": tier, + "litellm_params": { + "model": "openai/gpt-4o-mini", + "api_key": "integration-provider-key", + "api_base": provider_url + "/v1", + }, + }, + { + "model_name": router, + "litellm_params": { + "model": "auto_router/complexity_router", + "complexity_router_config": { + "classifier_type": "heuristic", + "keyword_tier_rules": [{"keywords": [keyword], "tier": "REASONING"}], + "tiers": {"SIMPLE": tier, "MEDIUM": tier, "COMPLEX": tier, "REASONING": tier}, + }, + }, + }, + ) + + return models + + +def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-route-" + uuid.uuid4().hex + router: Final = "lf-router-" + uuid.uuid4().hex[:8] + keyword: Final = "acquire-globex-" + uuid.uuid4().hex[:8] + with ( + _langfuse_rig( + gateway, + tmp_path, + {}, + marker, + overrides={"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "no_content"}, + model_list=_keyword_router(router, keyword), + ) as rig, + rig.candidate.scenario() as scenario, + ): + key: Final = scenario.key(key_alias="lf-route-" + uuid.uuid4().hex) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": router, + "messages": [{"role": "user", "content": f"{marker} please {keyword} now"}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + + attrs: Final = _v2_llm_span(rig.collector, response.json()["id"]) + observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) + assert set(observation) <= _EXPORTED_FIELDS, observation + langfuse_metadata: Final = { + name: value + for name, value in attrs.items() + if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.") + } + assert keyword not in json.dumps(langfuse_metadata, default=str), langfuse_metadata + assert keyword not in json.dumps(attrs, default=str), attrs + + +def _span_records(body: bytes) -> tuple[tuple[dict[str, object], int], ...]: + request: Final = ExportTraceServiceRequest.FromString(body) + return tuple( + ( + { + attribute.key: getattr(attribute.value, attribute.value.WhichOneof("value")) + for attribute in span.attributes + }, + span.dropped_attributes_count, + ) + for resource in request.resource_spans + for scope in resource.scope_spans + for span in scope.spans + ) + + +def _identity_attributes(attrs: Mapping[str, object]) -> frozenset[str]: + names: Final = {"langfuse.observation.metadata", *(f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)} + return frozenset(key for key in attrs if key in names) + + +def _limit_sweep( + rig: _Rig, + key: str, + marker: str, + bodies: Mapping[int, Mapping[str, object]], +) -> dict[int, tuple[dict[str, object], int]]: + """The exported span and dropped-attribute count per size; each body's prompt carries ``::``.""" + drained: list[ + tuple[dict[str, object], int] + ] = [] # mutable-ok: drain() consumes the wire queue, so observed spans must accumulate across eventually polls + for body in bodies.values(): + response = rig.candidate.request("POST", "/v1/chat/completions", dict(body), key=key) + assert response.status_code == 200, response.text + + def by_size() -> dict[int, tuple[dict[str, object], int] | None]: + drained.extend(record for batch in rig.collector.drain() for record in _span_records(batch.body)) + return { + size: next( + ( + (attrs, dropped) + for attrs, dropped in drained + if f"{marker}:{size}:" in str(attrs.get("langfuse.observation.input", "")) + ), + None, + ) + for size in bodies + } + + observed: Final = eventually(by_size, lambda spans: None not in spans.values(), seconds=30) + return {size: record for size, record in observed.items() if record is not None} + + +def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-v2-limit-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {}, + marker, + overrides={ + "MAPPER_NAMES": "genai,openinference", + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only", + }, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + + def body(message_count: int) -> dict[str, object]: + return { + "model": identity["model"], + "messages": [ + {"role": ("user", "assistant")[n % 2], "content": f"{marker} turn {n}"} + for n in range(message_count - 1) + ] + + [{"role": "user", "content": f"{marker}:{message_count}: turn {message_count - 1}"}], + "user": identity["end_user"], + "cache": {"no-cache": True}, + } + + spans: Final = _limit_sweep( + rig, key, marker, {message_count: body(message_count) for message_count in (1, *range(20, 61))} + ) + + reference, _ = spans[1] + identity_count: Final = len(_identity_attributes(reference)) + assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference) + shed_by_size: Final[dict[int, tuple[int, int]]] = {} # mutable-ok: filled once per swept span below + for message_count, (attrs, _) in spans.items(): + kept = {n for n in range(message_count) if f"llm.input_messages.{n}.message.content" in attrs} + shed_messages = message_count - len(kept) + shed_identity = identity_count - len(_identity_attributes(attrs)) + shed_by_size[message_count] = (shed_identity, shed_messages) + if shed_messages: + assert shed_identity == identity_count, (message_count, shed_identity, shed_messages) + assert [ + size for size, (shed_identity, shed_messages) in shed_by_size.items() if shed_identity and not shed_messages + ], shed_by_size + assert [size for size, (_, shed_messages) in shed_by_size.items() if shed_messages], shed_by_size + + +_APP_SPAN_HOOK: Final = """ +from opentelemetry import trace + +from litellm.integrations.custom_logger import CustomLogger + + +class AppSpanHook(CustomLogger): + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + with trace.get_tracer("lit8283-app").start_as_current_span("app.audit") as span: + span.set_attributes( + { + "app.response_id": response_obj.id, + "langfuse.observation.metadata": '{"source": "app"}', + "langfuse.trace.metadata.user_api_key_alias": "app-set-alias", + } + ) + + +instance = AppSpanHook() +""" + + +def test_otel_v2_span_another_tracer_wrote_keeps_its_langfuse_metadata_keys_at_the_collector( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-v2-app-" + uuid.uuid4().hex + (tmp_path / "lit8283_app_span.py").write_text(_APP_SPAN_HOOK) + with ( + wire_server(lambda _request: Reply()) as generic, + _langfuse_rig( + gateway, + tmp_path, + {"callbacks": ["langfuse_otel", "otel", "lit8283_app_span.instance"]}, + marker, + overrides={"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url}, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + + response_id: Final = response.json()["id"] + + def app(span: Mapping[str, object]) -> bool: + return span.get("app.response_id") == response_id + + def llm(span: Mapping[str, object]) -> bool: + return _response_id(span.get("gen_ai.response.id")) == _response_id(response_id) + + spans: Final = eventually( + _watched_spans(generic), lambda seen: any(map(app, seen)) and any(map(llm, seen)), seconds=25 + ) + app_span: Final = next(span for span in spans if app(span)) + assert app_span["langfuse.observation.metadata"] == '{"source": "app"}', app_span + assert app_span["langfuse.trace.metadata.user_api_key_alias"] == "app-set-alias", app_span + collector_llm: Final = next(span for span in spans if llm(span)) + assert not _identity_attributes(collector_llm), sorted(collector_llm) + + +def _chat_span_drops(body: bytes) -> tuple[tuple[int, int, int], ...]: + request: Final = ExportTraceServiceRequest.FromString(body) + return tuple( + (span.dropped_attributes_count, span.dropped_events_count, span.dropped_links_count) + for resource in request.resource_spans + for scope in resource.scope_spans + for span in scope.spans + if span.name.startswith("chat ") + ) + + +def test_otel_v2_collector_keeps_the_dropped_counts_of_a_span_without_langfuse_request_metadata( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-v2-drops-" + uuid.uuid4().hex + with ( + wire_server(lambda _request: Reply()) as generic, + _langfuse_rig( + gateway, + tmp_path, + {"callbacks": ["langfuse_otel", "otel"]}, + marker, + overrides={ + "OTEL_EXPORTER": "otlp_http", + "OTEL_ENDPOINT": generic.url, + "OTEL_SPAN_EVENT_COUNT_LIMIT": "0", + }, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker} scripted -fail- upstream"}], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code >= 500, response.text + + drops: Final[dict[str, list[tuple[int, int, int]]]] = {"langfuse": [], "generic": []} # mutable-ok: drained + + def observed() -> dict[str, list[tuple[int, int, int]]]: + for name, wire in (("langfuse", rig.collector), ("generic", generic)): + drops[name].extend(drop for batch in wire.drain() for drop in _chat_span_drops(batch.body)) + return drops + + settled: Final = eventually( + observed, lambda seen: bool(seen["langfuse"]) and len(seen["generic"]) == len(seen["langfuse"]), seconds=25 + ) + assert any(events for _, events, _ in settled["langfuse"]), settled + assert sorted(settled["generic"]) == sorted(settled["langfuse"]), settled + + +def test_otel_v2_langfuse_mapper_sends_no_langfuse_request_metadata_to_a_plain_collector( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-v2-mapper-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {"callbacks": ["otel"]}, + marker, + overrides={ + "MAPPER_NAMES": "genai,langfuse", + "OTEL_EXPORTER": "otlp_http", + "OTEL_ENDPOINT": "__collector__", + }, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + response: Final = rig.candidate.request( + "POST", + "/v1/chat/completions", + { + "model": identity["model"], + "messages": [{"role": "user", "content": marker}], + "user": identity["end_user"], + "cache": {"no-cache": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + + collector: Final = _v2_llm_span(rig.collector, response.json()["id"]) + assert collector["langfuse.trace.metadata.team_id"] == identity["team_id"], collector + assert not _identity_attributes(collector), sorted(collector) diff --git a/tests/integration/observability/test_otel_team_capture_message_content.py b/tests/integration/observability/test_otel_team_capture_message_content.py index 84da9a40486..c964dd3a772 100644 --- a/tests/integration/observability/test_otel_team_capture_message_content.py +++ b/tests/integration/observability/test_otel_team_capture_message_content.py @@ -465,6 +465,27 @@ def test_arize_destination_drops_openinference_content(rig: Rig) -> None: assert model.get("llm.model_name"), sorted(model) +def test_arize_destination_gets_no_langfuse_request_metadata(rig: Rig) -> None: + team, key = rig.team_key( + None, "arize-meta", callback="arize", variables={"arize_space_key": "space-meta", "arize_api_key": "arize-meta"} + ) + cursors: Final = rig.cursors() + sent: Final = _served(rig.send(key)) + operator: Final = _arrived(rig.sinks.operator, cursors.operator, sent, frozenset()) + assert any("langfuse.observation.metadata" in span["attributes"] for span in _model_spans(operator)), _names( + operator + ) + tenant: Final = _arrived(rig.sinks.arize, cursors.arize, sent, frozenset(), ("arize-space-id", "space-meta")) + models: Final = [span["attributes"] for span in _model_spans(tenant)] + assert any(attributes.get("langfuse.trace.metadata.team_id") == team for attributes in models), models + assert not [ + name + for attributes in models + for name in attributes + if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.user_api_key_") + ], models + + def test_weave_destination_drops_its_content(rig: Rig) -> None: _, key = rig.team_key( "no_content", diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index c7a057cca5c..fdd1e0fc451 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -18,7 +18,7 @@ from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from opentelemetry.trace import Status, StatusCode +from opentelemetry.trace import Link, SpanContext, Status, StatusCode import litellm from litellm.integrations.custom_logger import CustomLogger @@ -5067,3 +5067,129 @@ class TestArizeProjectRouting: monkeypatch.setenv("ARIZE_PROJECT_NAME", "gateway-prod") assert self._forwarded_projects() == {"gateway-prod"} + + +_LANGFUSE_REQUEST_METADATA_SPAN: Final = MappingProxyType( + { + "langfuse.observation.metadata": '{"requester_metadata": {"headers": {"x-tenant": "acme"}}}', + "langfuse.trace.metadata.user_api_key_alias": "prod-key", + "langfuse.trace.metadata.user_api_key_user_id": "user-1", + "langfuse.trace.metadata.team_id": "team-9", + "gen_ai.operation.name": "chat", + "gen_ai.request.model": "gpt-4o", + } +) +_WITHOUT_LANGFUSE_REQUEST_METADATA: Final = MappingProxyType( + {"langfuse.trace.metadata.team_id": "team-9", "gen_ai.operation.name": "chat", "gen_ai.request.model": "gpt-4o"} +) + + +class TestLangfuseRequestMetadataStaysWithLangfuse: + """The request metadata the Langfuse mapper writes reaches a Langfuse destination and no other.""" + + @staticmethod + def _emit( + provider: TracerProvider, + destinations: tuple[OtelDestination, ...] = (), + tracer_name: str = "litellm", + overflow: bool = False, + ) -> None: + def run() -> None: + set_request_destinations(destinations) + links: Final = (Link(SpanContext(trace_id=1, span_id=1, is_remote=True)),) if overflow else () + with get_tracer(provider, tracer_name).start_as_current_span("chat gpt-4o", links=links) as span: + if overflow: + span.set_attribute("app.evicted", "first") + span.add_event("app.event") + span.set_attributes(_LANGFUSE_REQUEST_METADATA_SPAN) + + in_fresh_context(run) + + @staticmethod + def _provider( + monkeypatch: pytest.MonkeyPatch, *owners: ExporterOwner | None + ) -> tuple[TracerProvider, dict[ExporterOwner | None, InMemorySpanExporter]]: + exporters: dict[ExporterOwner | None, InMemorySpanExporter] = {} # mutable-ok: filled as specs build + kind: Final = "lit8283_request_metadata" + monkeypatch.setitem( + otel_providers._EXPORTER_FACTORIES, + kind, + lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()), + ) + config: Final = OpenTelemetryV2Config( + mapper_names=["genai", "langfuse"], + exporters=[ExporterSpec(kind=kind, owner=owner) for owner in owners], + ) + return build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True), exporters + + @staticmethod + def _fan_out() -> tuple[TracerProvider, InMemorySpanExporter, InMemorySpanExporter]: + langfuse_exporter, arize_exporter = InMemorySpanExporter(), InMemorySpanExporter() + provider: Final = TracerProvider() + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda destination: SimpleSpanProcessor( + langfuse_exporter if destination.callback_name == "langfuse_otel" else arize_exporter + ) + ) + ) + return provider, langfuse_exporter, arize_exporter + + def test_a_collector_beside_the_langfuse_exporter_gets_no_langfuse_request_metadata( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL) + + self._emit(provider) + + assert dict(by_name(exporters[ExporterOwner.LANGFUSE_OTEL])["chat gpt-4o"].attributes) == dict( + _LANGFUSE_REQUEST_METADATA_SPAN + ) + assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA) + + def test_a_collector_with_the_langfuse_mapper_and_no_langfuse_exporter_gets_no_langfuse_request_metadata( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + provider, exporters = self._provider(monkeypatch, None) + + self._emit(provider) + + assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA) + + def test_the_fan_out_forwards_langfuse_request_metadata_to_langfuse_destinations_only(self) -> None: + provider, langfuse_exporter, arize_exporter = self._fan_out() + + self._emit(provider, (LANGFUSE_DEST, INHERITING_DEST)) + + assert dict(by_name(langfuse_exporter)["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN) + assert dict(by_name(arize_exporter)["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA) + + def test_a_span_another_tracer_wrote_keeps_the_langfuse_metadata_keys_it_set( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL) + fanned, _langfuse_exporter, arize_exporter = self._fan_out() + + self._emit(provider, tracer_name="app") + self._emit(fanned, (LANGFUSE_DEST, INHERITING_DEST), tracer_name="app") + + assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN) + assert dict(by_name(arize_exporter)["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN) + + def test_a_span_without_the_langfuse_request_metadata_keeps_its_dropped_counts( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(_LANGFUSE_REQUEST_METADATA_SPAN))) + monkeypatch.setenv("OTEL_SPAN_EVENT_COUNT_LIMIT", "0") + monkeypatch.setenv("OTEL_SPAN_LINK_COUNT_LIMIT", "0") + provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL) + fanned, _langfuse_exporter, arize_exporter = self._fan_out() + + self._emit(provider, overflow=True) + self._emit(fanned, (LANGFUSE_DEST, INHERITING_DEST), overflow=True) + + stripped: Final = (by_name(exporters[None])["chat gpt-4o"], by_name(arize_exporter)["chat gpt-4o"]) + assert [dict(span.attributes or {}) for span in stripped] == [dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)] * 2 + assert [(span.dropped_attributes, span.dropped_events, span.dropped_links) for span in stripped] == [ + (1, 1, 1) + ] * 2 diff --git a/tests/unit/integrations/otel/test_otel_v2_emitter.py b/tests/unit/integrations/otel/test_otel_v2_emitter.py index 6e29b2e9c68..775b1ad38de 100644 --- a/tests/unit/integrations/otel/test_otel_v2_emitter.py +++ b/tests/unit/integrations/otel/test_otel_v2_emitter.py @@ -899,6 +899,37 @@ def test_indexed_messages_follow_the_span_limits_of_a_per_request_tracer_overrid assert span.attributes["llm.output_messages.0.message.content"] == "reply 0" +def _team_conversation_span(monkeypatch, redact, max_span_attributes): + monkeypatch.setattr("litellm.redact_user_api_key_info", redact) + return _conversation_span( + ["genai", "openinference", "langfuse"], + _conversation_payload(60), + span_limits=SpanLimits(max_span_attributes=max_span_attributes), + ) + + +def test_langfuse_identity_sheds_before_any_attribute_the_span_carries_without_it(monkeypatch): + """At its attribute limit a span drops the Langfuse identity keys first, then exactly what it drops without them.""" + plain = _team_conversation_span(monkeypatch, True, SpanLimits.UNSET).attributes + full = _team_conversation_span(monkeypatch, False, SpanLimits.UNSET).attributes + identity_keys = set(full) - set(plain) + assert identity_keys == { + "langfuse.observation.metadata", + "langfuse.trace.metadata.user_api_key_team_id", + "langfuse.trace.metadata.user_api_key_team_alias", + } + + squeezed_limit = len(plain) - 10 + squeezed_plain = _team_conversation_span(monkeypatch, True, squeezed_limit) + squeezed = _team_conversation_span(monkeypatch, False, squeezed_limit) + assert dict(squeezed.attributes) == dict(squeezed_plain.attributes) + assert squeezed.dropped_attributes == squeezed_plain.dropped_attributes == 0 + + roomy = _team_conversation_span(monkeypatch, False, len(plain) + 2).attributes + assert {key: roomy.get(key) for key in plain} == dict(plain) + assert len(set(roomy) & identity_keys) == 2 + + def test_span_attribute_limit_falls_back_to_the_environment_for_spans_outside_the_sdk(monkeypatch): monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "48") assert span_attribute_limit(INVALID_SPAN) == 48 diff --git a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py index 7e112fcf6a4..d87a74d0ab0 100644 --- a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py +++ b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py @@ -22,6 +22,7 @@ from litellm.integrations.otel.mappers import ( resolve_mappers, ) from litellm.integrations.otel.mappers.openinference import fit_indexed_messages +from litellm.integrations.otel.model.baggage import DEFAULT_BAGGAGE_METADATA_KEYS from litellm.integrations.otel.model.payloads import ( EmbeddingOutput, LLMCallSpanData, @@ -802,3 +803,157 @@ def test_resolve_mappers_composition_layers_vocabularies(): def test_resolve_mappers_rejects_unknown_name(): with pytest.raises(ValueError, match="unknown mapper name 'nope'"): resolve_mappers(["genai", "nope"]) + + +def _langfuse_attrs( + metadata: Mapping[str, object], metadata_keys: tuple[str, ...] = DEFAULT_BAGGAGE_METADATA_KEYS +) -> Mapping[str, object]: + payload: Final[dict[str, object]] = { + "call_type": "acompletion", + "custom_llm_provider": "openai", + "model": "gpt-4o", + "metadata": metadata, + "response": {}, + } + return LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload, metadata_keys=metadata_keys)) + + +def test_langfuse_mapper_emits_request_metadata_and_identity(): + attrs: Final = _langfuse_attrs( + { + "user_api_key_alias": "prod-key", + "user_api_key_user_id": "user-1", + "user_api_key_end_user_id": "end-1", + "user_api_key_team_id": "team-9", + "user_api_key_team_alias": "team nine", + "user_api_key_max_budget": None, + "requester_ip_address": "10.0.0.1", + "spend_logs_metadata": {"ticket": "LIT-8283"}, + "requester_metadata": {"headers": {"x-tenant": "acme"}}, + } + ) + + assert json.loads(attrs["langfuse.observation.metadata"]) == { + "user_api_key_team_id": "team-9", + "user_api_key_team_alias": "team nine", + "user_api_key_user_id": "user-1", + "user_api_key_alias": "prod-key", + "user_api_key_end_user_id": "end-1", + } + assert attrs["langfuse.trace.metadata.user_api_key_alias"] == "prod-key" + assert attrs["langfuse.trace.metadata.user_api_key_user_id"] == "user-1" + assert attrs["langfuse.trace.metadata.user_api_key_end_user_id"] == "end-1" + assert attrs["langfuse.trace.metadata.user_api_key_team_id"] == "team-9" + assert attrs["langfuse.trace.metadata.user_api_key_team_alias"] == "team nine" + assert attrs["langfuse.trace.metadata.team_id"] == "team-9" + assert attrs["langfuse.trace.metadata.team_alias"] == "team nine" + assert attrs["langfuse.observation.metadata.provider"] == "openai" + + +def test_langfuse_mapper_exports_only_the_promoted_metadata_and_the_team(): + attrs: Final = _langfuse_attrs( + { + "user_api_key_alias": "prod-key", + "user_api_key_team_id": "team-9", + "requester_ip_address": "10.0.0.1", + "user_api_key_user_email": "caller-email-secret", + "user_api_key_auth_metadata": {"note": "auth-metadata-secret"}, + "user_agent": "user-agent-secret", + "headers": {"x-tenant-secret": "header-secret"}, + "requester_custom_headers": {"x-tenant-secret": "custom-header-secret"}, + "requester_metadata": {"note": "requester-metadata-secret"}, + "spend_logs_metadata": {"note": "spend-logs-secret"}, + "prompt_management_metadata": {"prompt_variables": {"input": "prompt-variables-secret"}}, + "mcp_tool_call_metadata": {"arguments": {"location": "mcp-arguments-secret"}}, + "vector_store_request_metadata": [{"query": "vector-query-secret"}], + "routing_decision": {"matched_keyword": "routing-prompt-secret"}, + } + ) + + assert "secret" not in json.dumps(attrs) + assert json.loads(attrs["langfuse.observation.metadata"]) == { + "user_api_key_team_id": "team-9", + "user_api_key_alias": "prod-key", + } + + +@pytest.mark.parametrize( + ("metadata_keys", "expected"), + [ + ((), {"user_api_key_team_id": "team-9"}), + (("user_api_key_user_id",), {"user_api_key_team_id": "team-9", "user_api_key_user_id": "user-1"}), + (("requester_metadata.trace_id", "requester_ip_address"), {"user_api_key_team_id": "team-9"}), + ], +) +def test_langfuse_mapper_follows_the_operator_metadata_allowlist( + metadata_keys: tuple[str, ...], expected: Mapping[str, str] +) -> None: + attrs: Final = _langfuse_attrs( + { + "user_api_key_alias": "prod-key", + "user_api_key_user_id": "user-1", + "user_api_key_team_id": "team-9", + "requester_ip_address": "10.0.0.1", + "requester_metadata": {"trace_id": "trace-7", "note": "requester-metadata-secret"}, + }, + metadata_keys=metadata_keys, + ) + + assert json.loads(attrs["langfuse.observation.metadata"]) == expected + assert {key for key in attrs if key.startswith("langfuse.trace.metadata.user_api_key_")} == { + f"langfuse.trace.metadata.{key}" for key in expected if key.startswith("user_api_key_") + } + + +def test_langfuse_mapper_skips_empty_identity_values(): + attrs: Final = _langfuse_attrs( + { + "user_api_key_alias": None, + "user_api_key_user_id": "", + "user_api_key_end_user_id": "end-1", + } + ) + + assert "langfuse.trace.metadata.user_api_key_alias" not in attrs + assert "langfuse.trace.metadata.user_api_key_user_id" not in attrs + assert attrs["langfuse.trace.metadata.user_api_key_end_user_id"] == "end-1" + + +def test_langfuse_mapper_redacts_user_api_key_fields(monkeypatch: pytest.MonkeyPatch) -> None: + import litellm + + monkeypatch.setattr(litellm, "redact_user_api_key_info", True) + attrs: Final = _langfuse_attrs( + { + "user_api_key_alias": "prod-key", + "user_api_key_team_id": "team-9", + "requester_ip_address": "10.0.0.1", + } + ) + + assert "langfuse.observation.metadata" not in attrs + assert not [key for key in attrs if key.startswith("langfuse.trace.metadata.user_api_key_")] + assert attrs["langfuse.trace.metadata.team_id"] == "team-9" + + +def test_langfuse_mapper_omits_observation_metadata_without_payload_metadata(): + attrs: Final = _langfuse_attrs({}) + + assert "langfuse.observation.metadata" not in attrs + + +@pytest.mark.parametrize( + "unexpected_shape", + [ + {"prompt_management_metadata": {1: "one", "prompt_id": "prompt-1"}}, + {"mcp_tool_call_metadata": {1: "one", "name": "lookup_weather"}}, + {"vector_store_request_metadata": [{1: "one", "vector_store_id": "vs-1"}]}, + {"requester_metadata": 1}, + {"requester_metadata": True}, + {"requester_metadata": 1.5}, + ], +) +def test_span_data_builds_from_request_metadata_of_unexpected_shape(unexpected_shape: dict[str, object]) -> None: + attrs: Final = _langfuse_attrs({**unexpected_shape, "user_api_key_alias": "prod-key"}) + + assert json.loads(attrs["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}