From ac5b208546fe6d7ed829e1e72dbcaf3f9d2fcde9 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Thu, 15 Oct 2026 05:30:53 -0700 Subject: [PATCH] fix(langfuse_otel): read the caller trace user once and stamp the end user on the v2 generation span The trace-user fallback re-ran the header merge, which mutates the request metadata in place, so every request with a langfuse_* header logged a false 'Overwriting Langfuse trace_user_id' warning. Read the caller's trace user through caller_trace_controls instead. On OTel v2 the end user reached only the root span; the generation span, which carries usage and cost, kept an empty Langfuse user unless the caller sent trace_user_id. LLMCallEvent now takes the same end-user fallback. --- .../integrations/langfuse/langfuse_otel.py | 24 +++++-------------- litellm/integrations/otel/model/metadata.py | 4 ++-- .../integrations/otel/model/trace_controls.py | 6 ++--- .../test_langfuse_otel_identity.py | 11 +++++---- .../integrations/otel/test_langfuse_logger.py | 2 ++ .../otel/test_otel_v2_sources_of_truth.py | 1 + tests/unit/integrations/test_langfuse_otel.py | 12 ++++++++++ 7 files changed, 32 insertions(+), 28 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 69d03fdcdd5..995dc6c0ace 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -12,8 +12,8 @@ from litellm.integrations.langfuse.langfuse_otel_attributes import ( LangfuseLLMObsOTELAttributes, ) from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig -from litellm.integrations.otel.model.trace_controls import metadata_bodies -from litellm.integrations.otel.model.utils import as_str, as_str_mapping +from litellm.integrations.otel.model.trace_controls import caller_trace_controls, langfuse_trace_controls +from litellm.integrations.otel.model.utils import as_str_mapping from litellm.litellm_core_utils.safe_json_loads import safe_json_loads from litellm.types.integrations.langfuse_otel import ( LangfuseSpanAttributes, @@ -56,13 +56,7 @@ class LangfuseOtelLogger(OpenTelemetry): # Set Langfuse specific attributes ######################################################### LangfuseOtelLogger._set_langfuse_specific_attributes(span=span, kwargs=kwargs, response_obj=response_obj) - LangfuseOtelLogger._set_trace_user_attribute( - span=span, - kwargs=kwargs, - metadata=as_str_mapping( - LangfuseOtelLogger._extract_langfuse_metadata(kwargs) # pyright: ignore[reportUnknownArgumentType,reportUnknownMemberType] # helper returns a loosely typed dict - ), - ) + LangfuseOtelLogger._set_trace_user_attribute(span=span, kwargs=kwargs) @staticmethod def _extract_langfuse_metadata(kwargs: dict) -> dict: @@ -267,25 +261,19 @@ class LangfuseOtelLogger(OpenTelemetry): LangfuseOtelLogger._set_observation_output(span=span, response_obj=response_obj) @staticmethod - def _set_trace_user_attribute( - span: Span, kwargs: Mapping[str, object], metadata: Mapping[str, object] | None - ) -> None: + def _set_trace_user_attribute(span: Span, kwargs: Mapping[str, object]) -> None: slp: Final = as_str_mapping(kwargs.get("standard_logging_object")) slp_metadata: Final = as_str_mapping(slp.get("metadata")) if slp is not None else None if slp is None or slp_metadata is None: return - litellm_params: Final = as_str_mapping(kwargs.get("litellm_params")) or {} - bodies: Final = metadata_bodies(litellm_params) - caller: Final = (as_str(metadata.get("trace_user_id")) if metadata is not None else None) or next( - (value for body in bodies if (value := as_str(body.get("trace_user_id")))), None - ) + caller: Final = caller_trace_controls(kwargs).user_id if caller is not None: safe_set_attribute(span, LangfuseSpanAttributes.TRACE_USER_ID.value, caller) return end_user: Final = ( slp_metadata.get("user_api_key_end_user_id") or slp.get("end_user") - or next((found for body in bodies if (found := as_str(body.get("user_api_key_end_user_id")))), None) + or langfuse_trace_controls(kwargs).user_id ) if end_user: safe_set_attribute(span, LangfuseSpanAttributes.TRACE_USER_ID.value, str(end_user)) diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index 8524df71b44..9d2d896a8e1 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -48,7 +48,7 @@ from litellm.constants import ( SESSION_ID_GENERATED_METADATA_KEY, ) from litellm.integrations.otel.model.semconv import resolve_operation -from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls +from litellm.integrations.otel.model.trace_controls import TraceControls, langfuse_trace_controls from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds from litellm.types.utils import InternalCallOrigin @@ -260,7 +260,7 @@ class LLMCallEvent: payload: Final = cast("StandardLoggingPayload", raw_payload) if raw_payload else None operation: Final = resolve_operation(as_str(kwargs.get("call_type"))) model: Final = as_str(kwargs.get("model")) or "" - trace: Final = caller_trace_controls(kwargs) + trace: Final = langfuse_trace_controls(kwargs) return cls( call_id=_call_id(payload, kwargs), payload=payload, diff --git a/litellm/integrations/otel/model/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py index dc64d5d5743..2f683c7fd1b 100644 --- a/litellm/integrations/otel/model/trace_controls.py +++ b/litellm/integrations/otel/model/trace_controls.py @@ -33,7 +33,7 @@ def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: return TraceControls() proxy_request: Final = as_str_mapping(request.get("proxy_server_request")) headers: Final = as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None - bodies: Final = metadata_bodies(request) + bodies: Final = _metadata_bodies(request) def scalar(control: str) -> str | None: from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) if headers is not None else None @@ -57,13 +57,13 @@ def langfuse_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: if request is None: return controls end_user: Final = next( - (value for body in metadata_bodies(request) if (value := as_str(body.get("user_api_key_end_user_id")))), + (value for body in _metadata_bodies(request) if (value := as_str(body.get("user_api_key_end_user_id")))), None, ) return replace(controls, user_id=end_user) -def metadata_bodies(request: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: +def _metadata_bodies(request: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: return tuple( metadata for key in ("metadata", "litellm_metadata") diff --git a/tests/integration/observability/test_langfuse_otel_identity.py b/tests/integration/observability/test_langfuse_otel_identity.py index 8ad5e5509e6..2348610c4fc 100644 --- a/tests/integration/observability/test_langfuse_otel_identity.py +++ b/tests/integration/observability/test_langfuse_otel_identity.py @@ -9,6 +9,7 @@ from _langfuse_otel import ( _span_attributes_containing_marker, _trace_user_span_attributes, _upstream_reply, + _user_id_session_id, ) from integration._support.client import Gateway, eventually from integration._support.wire import Reply, Request, wire_server @@ -243,13 +244,13 @@ def test_langfuse_otel_v2_header_end_user_lands_in_user_id(gateway: Gateway, tmp batches: Final[list[bytes]] = [] user_spans: Final = eventually( lambda: _trace_user_span_attributes(collector, batches, marker), - lambda spans: len(spans) == 1, + lambda spans: len(spans) == 2, seconds=30, ) - assert {key: user_spans[0].get(key) for key in ("user.id", "session.id")} == { - "user.id": f"end-user-{marker}", - "session.id": None, - }, user_spans[0] + assert [_user_id_session_id(attributes) for attributes in user_spans] == [ + {"user.id": f"end-user-{marker}", "session.id": None} + ] * 2, user_spans + assert any(attributes.get("gen_ai.response.id") == marker for attributes in user_spans), user_spans def test_langfuse_otel_messages_caller_trace_user_id_under_litellm_metadata_wins_over_the_end_user( diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py index 784f4878533..f99f4d3ee42 100644 --- a/tests/unit/integrations/otel/test_langfuse_logger.py +++ b/tests/unit/integrations/otel/test_langfuse_logger.py @@ -447,6 +447,7 @@ def test_the_proxy_end_user_fills_user_id_when_the_caller_names_no_trace_user(): ) assert root_attrs["user.id"] == "end-1" + assert generation_attrs["user.id"] == "end-1" def test_a_callers_trace_user_id_still_wins_over_the_proxy_end_user(): @@ -462,6 +463,7 @@ def test_a_callers_trace_user_id_still_wins_over_the_proxy_end_user(): ) assert root_attrs["user.id"] == "caller-1" + assert generation_attrs["user.id"] == "caller-1" def test_caller_metadata_cannot_override_the_proxy_team_identity(): diff --git a/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py b/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py index ec51263d3db..a2690f39494 100644 --- a/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py +++ b/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py @@ -1387,6 +1387,7 @@ def test_caller_trace_controls_carry_user_session_and_tags(request_data, expecte ) def test_langfuse_trace_controls_fall_back_to_the_proxy_end_user(request_data, expected): assert langfuse_trace_controls({"litellm_params": request_data}) == expected + assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace == expected def test_llm_span_data_carries_the_caller_trace_controls(): diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 569fa5cb119..874d619fe12 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -1,5 +1,6 @@ import base64 import json +import logging import os from datetime import datetime, timezone from typing import Final @@ -1099,6 +1100,17 @@ class TestLangfuseOtelTraceIdentity: ) assert identity == {"user.id": "end-1", "session.id": "sess-1"} + def test_langfuse_trace_user_id_header_wins_without_an_overwrite_warning(self, caplog): + kwargs = self._kwargs( + slp_metadata={"user_api_key_end_user_id": "end-1"}, + litellm_metadata={"user_api_key_end_user_id": "end-1"}, + ) + kwargs["litellm_params"]["proxy_server_request"] = {"headers": {"langfuse_trace_user_id": "header-1"}} + with caplog.at_level(logging.WARNING, logger="LiteLLM"): + identity, _ = self._identity(kwargs) + assert identity == {"user.id": "header-1", "session.id": None} + assert [record.message for record in caplog.records if "Overwriting" in record.message] == [] + def test_internal_user_without_an_end_user_never_lands_in_user_id(self): identity, _ = self._identity(self._kwargs(slp_metadata={"user_api_key_user_id": "internal-1"})) assert identity == {"user.id": None, "session.id": None}