fix(otel v2): map the caller's Langfuse user, session and tags onto the root and generation spans

`langfuse_otel` (OTel v2) only carried `trace_name` from the caller's metadata, so `metadata.trace_user_id` / `session_id` / `tags` (and the `langfuse_trace_user_id` / `langfuse_session_id` proxy headers) never reached Langfuse's user, session and tags fields. Widen the typed caller boundary to `TraceControls`, map it through one `LangfuseMapper.trace_attributes` table on both the root observation and the generation span, and keep `team_id` / `team_alias` proxy-authoritative

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-14 22:31:21 +00:00
parent 7fd541efb9
commit 1e583e8e79
8 changed files with 255 additions and 42 deletions

View file

@ -6,9 +6,9 @@ from litellm.integrations.otel.logger import OpenTelemetryV2
from litellm.integrations.otel.mappers.langfuse import (
LANGFUSE_OBSERVATION_INPUT,
LANGFUSE_OBSERVATION_OUTPUT,
LANGFUSE_TRACE_NAME,
LangfuseMapper,
)
from litellm.integrations.otel.model.metadata import caller_trace_name
from litellm.integrations.otel.model.metadata import caller_trace_controls
from litellm.integrations.otel.model.request_io import request_input, response_output, stream_output
from litellm.integrations.otel.plumbing.context import request_root_span
@ -18,14 +18,13 @@ if TYPE_CHECKING:
class LangfuseOpenTelemetryV2(OpenTelemetryV2):
"""Names the trace from the request. Langfuse reads ``langfuse.trace.name`` off the root observation,
and the proxy's root span is still recording when the LLM call starts."""
"""Stamps the caller's trace controls (name, user, session, tags) on the request. Langfuse reads them off
the root observation, and the proxy's root span is still recording when the LLM call starts."""
def log_pre_api_call(self, model: str, messages: object, kwargs: Mapping[str, object]) -> None:
root: Final = request_root_span()
name: Final = caller_trace_name(kwargs)
if root is not None and root.is_recording() and name is not None:
root.set_attribute(LANGFUSE_TRACE_NAME, name)
if root is not None and root.is_recording():
root.set_attributes(LangfuseMapper.trace_attributes(caller_trace_controls(kwargs)))
super().log_pre_api_call(model, messages, kwargs)

View file

@ -554,7 +554,7 @@ class OpenTelemetryV2(CustomLogger):
capture_content=self.config.capture_span_content,
time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
request_route=request_root_http_route(),
trace_name=call.trace_name,
trace=call.trace,
)
end_time_ns: Final = to_ns(end_time)
if carrier is not None and carrier.span is not None:

View file

@ -5,12 +5,14 @@ Langfuse ingests OTLP spans and reads from its own vendor namespace
``GenAIMapper`` to send canonical + Langfuse-flavored spans simultaneously.
Every attribute is declared as a ``key -> extractor`` table entry (one callable
per mapping operation): ``_LLM_CALL_ATTRS`` for scalars and ``_BLOB_ATTRS`` for
the JSON-serialized payloads. ``_llm_call`` just applies both tables.
per mapping operation): ``_LLM_CALL_ATTRS`` for scalars, ``_TRACE_ATTRS`` for the
caller's trace controls (shared with the root observation), and ``_BLOB_ATTRS`` for
the JSON-serialized payloads. ``_llm_call`` just applies the three tables.
"""
import json
from collections.abc import Callable
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import Final
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
@ -20,6 +22,7 @@ from litellm.integrations.otel.mappers.utils import (
output_messages,
serialize_messages,
)
from litellm.integrations.otel.model.metadata import TraceControls
from litellm.integrations.otel.model.payloads import (
LLMCallSpanData,
LLMRequestParams,
@ -29,6 +32,9 @@ from litellm.integrations.otel.model.payloads import (
LANGFUSE_OBSERVATION_INPUT: Final = "langfuse.observation.input"
LANGFUSE_OBSERVATION_OUTPUT: Final = "langfuse.observation.output"
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"
class LangfuseMapper:
@ -37,11 +43,19 @@ class LangfuseMapper:
"langfuse.observation.model.name": lambda d: d.request_model or None,
"langfuse.observation.metadata.provider": lambda d: d.provider or None,
"langfuse.observation.id": lambda d: d.identity.call_id or None,
LANGFUSE_TRACE_NAME: lambda d: d.trace_name or None,
"langfuse.trace.metadata.team_id": lambda d: d.identity.team_id or None,
"langfuse.trace.metadata.team_alias": lambda d: d.identity.team_alias or None,
}
_TRACE_ATTRS: Mapping[str, Callable[[TraceControls], AttrValue | None]] = MappingProxyType(
{
LANGFUSE_TRACE_NAME: lambda t: t.name or None,
LANGFUSE_TRACE_USER_ID: lambda t: t.user_id or None,
LANGFUSE_TRACE_SESSION_ID: lambda t: t.session_id or None,
LANGFUSE_TRACE_TAGS: lambda t: t.tags or None,
}
)
# Sub-tables folded into their respective JSON blobs.
_MODEL_PARAMS: dict[str, Callable[[LLMRequestParams], AttrValue | None]] = {
"temperature": lambda rp: rp.temperature,
@ -77,9 +91,14 @@ class LangfuseMapper:
case _:
return {}
@classmethod
def trace_attributes(cls, trace: TraceControls) -> AttributeMap:
return collect(cls._TRACE_ATTRS, trace)
@classmethod
def _llm_call(cls, data: LLMCallSpanData) -> AttributeMap:
return {
**collect(cls._LLM_CALL_ATTRS, data),
**cls.trace_attributes(data.trace),
**collect(cls._BLOB_ATTRS, data),
}

View file

@ -36,7 +36,7 @@ model. They coincide on the SDK path, which is correct.
from __future__ import annotations
from collections.abc import Iterator, Mapping
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, cast
@ -48,7 +48,20 @@ from litellm.integrations.otel.model.utils import as_str, to_seconds
if TYPE_CHECKING:
from litellm.types.utils import StandardLoggingPayload
LANGFUSE_TRACE_NAME_HEADER: Final = "langfuse_trace_name"
LANGFUSE_HEADER_PREFIX: Final = "langfuse_"
@dataclass(frozen=True, slots=True)
class TraceControls:
"""The caller's trace-level Langfuse controls: ``metadata.trace_name`` / ``trace_user_id`` / ``session_id`` /
``tags`` on the request (SDK or proxy body), with the proxy's ``langfuse_<control>`` headers winning over the
body for the scalar ones. Mutation controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are
deliberately not carried."""
name: str | None = None
user_id: str | None = None
session_id: str | None = None
tags: tuple[str, ...] = ()
@dataclass(frozen=True)
@ -217,7 +230,7 @@ class LLMCallEvent:
# needs to be reasonable for a span that never gets closed (a leak).
provisional_span_name: str
time_to_first_chunk_seconds: float | None
trace_name: str | None
trace: TraceControls
@classmethod
def from_dict(cls, kwargs: Mapping[str, Any]) -> LLMCallEvent:
@ -234,29 +247,41 @@ class LLMCallEvent:
upstream_started=kwargs.get("api_call_start_time") is not None,
provisional_span_name=f"{operation.value} {model}".strip(),
time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs),
trace_name=caller_trace_name(kwargs),
trace=caller_trace_controls(kwargs),
)
def caller_trace_name(kwargs: Mapping[str, object]) -> str | None:
def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls:
request: Final = _as_str_mapping(kwargs.get("litellm_params"))
if request is None:
return None
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
from_header: Final = as_str(headers.get(LANGFUSE_TRACE_NAME_HEADER)) if headers is not None else None
if from_header:
return from_header
return next(
(
name
for key in ("metadata", "litellm_metadata")
if (metadata := _as_str_mapping(request.get(key))) is not None
and (name := as_str(metadata.get("trace_name")))
),
None,
headers: Final = (_as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None) or {}
bodies: Final = tuple(
metadata
for key in ("metadata", "litellm_metadata")
if (metadata := _as_str_mapping(request.get(key))) is not None
)
def scalar(control: str) -> str | None:
from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}"))
if from_header:
return from_header
return next((value for body in bodies if (value := as_str(body.get(control)))), None)
return TraceControls(
name=scalar("trace_name"),
user_id=scalar("trace_user_id"),
session_id=scalar("session_id"),
tags=next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()),
)
def _str_items(value: object) -> tuple[str, ...]:
if not isinstance(value, (list, tuple)):
return ()
return tuple(item for item in cast("Sequence[object]", value) if isinstance(item, str) and item)
def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None:
"""Seconds from the upstream request being issued (``api_call_start_time``)

View file

@ -13,6 +13,7 @@ from urllib.parse import urlsplit
from litellm.integrations.otel.model.metadata import (
RequestContext,
RequestIdentity,
TraceControls,
)
from litellm.integrations.otel.model.semconv import (
GenAIOperation,
@ -387,7 +388,7 @@ class LLMCallSpanData:
output_type: GenAIOutputType | None = None
call_type: str | None = None
request_route: str | None = None
trace_name: str | None = None
trace: TraceControls = field(default_factory=TraceControls)
@classmethod
def from_standard_logging_payload(
@ -396,7 +397,7 @@ class LLMCallSpanData:
capture_content: bool = False,
time_to_first_chunk_seconds: float | None = None,
request_route: str | None = None,
trace_name: str | None = None,
trace: TraceControls | None = None,
) -> LLMCallSpanData:
params: Final = cast(Mapping[str, object], payload.get("model_parameters") or {})
# The single parse of the request's metadata — the request-vs-provider
@ -438,7 +439,7 @@ class LLMCallSpanData:
output_type=resolve_output_type(call_type),
call_type=call_type or None,
request_route=request_route or context.identity.request_route,
trace_name=trace_name,
trace=trace or TraceControls(),
)

View file

@ -42,6 +42,7 @@ from litellm.types.utils import ( # noqa: E402
INPUT_ATTR: Final = "langfuse.observation.input"
OUTPUT_ATTR: Final = "langfuse.observation.output"
TRACE_NAME_ATTR: Final = "langfuse.trace.name"
TRACE_CONTROL_ATTRS: Final = (TRACE_NAME_ATTR, "user.id", "session.id", "langfuse.trace.tags")
CHAT_DATA: Final = {"model": "gpt-5.4-mini", "messages": [{"role": "user", "content": "ping"}]}
@ -374,6 +375,99 @@ def test_unnamed_request_leaves_the_trace_name_off_both_spans():
assert TRACE_NAME_ATTR not in root_attrs and TRACE_NAME_ATTR not in generation_attrs
@pytest.mark.parametrize("capture", ["span_only", "no_content"])
def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation(capture):
logger, exporter = _logger(capture=capture)
root_attrs, generation_attrs = _run_named_request(
logger,
exporter,
{
"metadata": {
"trace_user_id": "user-42",
"session_id": "session-7",
"tags": ["prod", "eval", "nightly"],
"user_api_key_team_id": "team-from-proxy",
},
"proxy_server_request": {"headers": {}},
},
)
for attrs in (root_attrs, generation_attrs):
assert attrs["user.id"] == "user-42"
assert attrs["session.id"] == "session-7"
assert tuple(attrs["langfuse.trace.tags"]) == ("prod", "eval", "nightly")
assert TRACE_NAME_ATTR not in attrs
def test_langfuse_user_and_session_headers_beat_body_metadata_on_both_spans():
logger, exporter = _logger()
root_attrs, generation_attrs = _run_named_request(
logger,
exporter,
{
"metadata": {"trace_user_id": "from-body", "session_id": "from-body"},
"proxy_server_request": {
"headers": {"langfuse_trace_user_id": "from-header", "langfuse_session_id": "from-header-s"}
},
},
)
for attrs in (root_attrs, generation_attrs):
assert attrs["user.id"] == "from-header"
assert attrs["session.id"] == "from-header-s"
def test_caller_metadata_cannot_override_the_proxy_team_identity():
logger, exporter = _logger()
response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))])
litellm_params: Final = {
"metadata": {"trace_user_id": "u", "trace_metadata": {"team_id": "spoofed"}, "team_id": "spoofed"}
}
logger.log_pre_api_call(
model="gpt-5.4-mini", messages=[], kwargs={"litellm_call_id": "call_1", "litellm_params": litellm_params}
)
payload: Final = {
"call_type": "acompletion",
"custom_llm_provider": "openai",
"model": "gpt-5.4-mini",
"messages": CHAT_DATA["messages"],
"response": response.model_dump(),
"status": "success",
"litellm_call_id": "call_1",
"metadata": {
"user_api_key_team_id": "real-team",
"user_api_key_team_alias": "real-alias",
"team_id": "spoofed",
"team_alias": "spoofed",
},
"hidden_params": {},
}
asyncio.run(
logger.async_log_success_event(
{"standard_logging_object": payload, "litellm_params": litellm_params}, response, None, None
)
)
attrs: Final = dict(exporter.get_finished_spans()[0].attributes or {})
assert attrs["user.id"] == "u"
assert attrs["langfuse.trace.metadata.team_id"] == "real-team"
assert attrs["langfuse.trace.metadata.team_alias"] == "real-alias"
assert "langfuse.trace.metadata" not in attrs and "langfuse.trace.id" not in attrs
def test_a_request_without_trace_controls_stamps_none_of_them():
logger, exporter = _logger()
root_attrs, generation_attrs = _run_named_request(
logger, exporter, {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}}
)
assert set(TRACE_CONTROL_ATTRS).isdisjoint(root_attrs)
assert set(TRACE_CONTROL_ATTRS).isdisjoint(generation_attrs)
@pytest.mark.parametrize(
("capture", "mappers"),
[("no_content", ("genai", "langfuse")), ("span_only", ("genai",))],

View file

@ -28,7 +28,7 @@ from litellm.integrations.otel import (
)
from litellm.integrations.otel.mappers.genai import GenAIMapper
from litellm.integrations.otel.model import spans as spans_mod
from litellm.integrations.otel.model.metadata import LLMCallEvent, caller_trace_name
from litellm.integrations.otel.model.metadata import LLMCallEvent, TraceControls, caller_trace_controls
from litellm.integrations.otel.model.payloads import (
LLMCallSpanData,
RequestIdentity,
@ -743,15 +743,62 @@ def test_request_identity_falls_back_to_legacy_team_keys():
ids=["header", "body", "anthropic-body", "header-beats-body", "blank-header-falls-through", "neither", "empty"],
)
def test_caller_trace_name_prefers_the_langfuse_header_over_body_metadata(request_data, expected):
assert caller_trace_name({"litellm_params": request_data}) == expected
assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace_name == expected
assert caller_trace_controls({"litellm_params": request_data}).name == expected
assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace.name == expected
def test_llm_span_data_carries_the_caller_trace_name():
data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace_name="nightly-eval")
@pytest.mark.parametrize(
("request_data", "expected"),
[
(
{"metadata": {"trace_user_id": "u-body", "session_id": "s-body", "tags": ["a", "b", "c"]}},
TraceControls(user_id="u-body", session_id="s-body", tags=("a", "b", "c")),
),
(
{
"proxy_server_request": {
"headers": {"langfuse_trace_user_id": "u-header", "langfuse_session_id": "s-header"}
},
"metadata": {"trace_user_id": "u-body", "session_id": "s-body"},
},
TraceControls(user_id="u-header", session_id="s-header"),
),
(
{"litellm_metadata": {"trace_user_id": "u-anthropic", "session_id": "s-anthropic", "tags": ["x"]}},
TraceControls(user_id="u-anthropic", session_id="s-anthropic", tags=("x",)),
),
(
{"metadata": {"tags": ["kept", 7, "", None, "also-kept"]}},
TraceControls(tags=("kept", "also-kept")),
),
({"metadata": {"tags": "not-a-list", "trace_user_id": "", "session_id": 12}}, TraceControls(session_id="12")),
(
{
"metadata": {
"trace_id": "forced",
"existing_trace_id": "forced",
"update_trace_keys": ["name"],
"trace_metadata": {"team_id": "spoofed"},
"user_api_key_team_id": "t1",
}
},
TraceControls(),
),
({}, TraceControls()),
],
ids=["body", "headers-beat-body", "anthropic-body", "non-string-tags-dropped", "scalar-coercion", "mutation-controls-ignored", "empty"],
)
def test_caller_trace_controls_carry_user_session_and_tags(request_data, expected):
assert caller_trace_controls({"litellm_params": request_data}) == expected
assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace == expected
assert data.trace_name == "nightly-eval"
assert LLMCallSpanData.from_standard_logging_payload(_sample_payload()).trace_name is None
def test_llm_span_data_carries_the_caller_trace_controls():
controls: Final = TraceControls(name="nightly-eval", user_id="u1", session_id="s1", tags=("a", "b"))
data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace=controls)
assert data.trace == controls
assert LLMCallSpanData.from_standard_logging_payload(_sample_payload()).trace == TraceControls()
def test_llm_span_carries_proxy_request_route():

View file

@ -18,6 +18,7 @@ from litellm.integrations.otel.mappers import (
WeaveMapper,
resolve_mappers,
)
from litellm.integrations.otel.model.metadata import TraceControls
from litellm.integrations.otel.model.payloads import (
LLMCallSpanData,
LLMRequestParams,
@ -135,8 +136,35 @@ def test_langfuse_mapper_observation_attrs():
def test_langfuse_mapper_names_the_trace_from_the_caller():
assert LangfuseMapper().map(_llm_call(trace_name="nightly-eval"))["langfuse.trace.name"] == "nightly-eval"
assert "langfuse.trace.name" not in LangfuseMapper().map(_llm_call(trace_name=None))
named = LangfuseMapper().map(_llm_call(trace=TraceControls(name="nightly-eval")))
assert named["langfuse.trace.name"] == "nightly-eval"
assert "langfuse.trace.name" not in LangfuseMapper().map(_llm_call(trace=TraceControls()))
def test_langfuse_mapper_carries_the_caller_user_session_and_tags():
controls = TraceControls(user_id="u-42", session_id="s-7", tags=("prod", "eval", "nightly"))
attrs = LangfuseMapper().map(_llm_call(trace=controls))
assert attrs["user.id"] == "u-42"
assert attrs["session.id"] == "s-7"
assert attrs["langfuse.trace.tags"] == ("prod", "eval", "nightly")
assert attrs["langfuse.trace.metadata.team_id"] == "t1"
assert attrs["langfuse.trace.metadata.team_alias"] == "team one"
def test_langfuse_mapper_omits_unset_trace_controls():
attrs = LangfuseMapper().map(_llm_call(trace=TraceControls(user_id="", session_id=None, tags=())))
assert {"user.id", "session.id", "langfuse.trace.tags", "langfuse.trace.name"}.isdisjoint(attrs)
def test_langfuse_trace_attributes_match_between_root_and_generation():
controls = TraceControls(name="n", user_id="u", session_id="s", tags=("t",))
generation = LangfuseMapper().map(_llm_call(trace=controls))
root = LangfuseMapper.trace_attributes(controls)
assert root == {"langfuse.trace.name": "n", "user.id": "u", "session.id": "s", "langfuse.trace.tags": ("t",)}
assert all(generation[key] == value for key, value in root.items())
def test_langfuse_mapper_skips_when_no_messages():