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.
This commit is contained in:
yucheng-berri 2026-10-15 05:30:53 -07:00
parent 5ac2bd29a9
commit ac5b208546
7 changed files with 32 additions and 28 deletions

View file

@ -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))

View file

@ -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,

View file

@ -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")

View file

@ -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(

View file

@ -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():

View file

@ -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():

View file

@ -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}