From d7f270ff039e225ee12aa3816e1f248dd5592308 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:22:34 -0700 Subject: [PATCH 1/3] fix(otel v2): send the request user as user.id from every OTel v2 callback The shared GenAI mapper now stamps user.id on the model-call span from the request's trace controls: the caller's trace_user_id, else the proxy end user. Every v2 preset starts with that mapper, so otel, langfuse_otel, arize, arize_phoenix, weave_otel, langtrace, signoz, levo, newrelic and agentops all send it. The key owner is never sent as user.id. --- litellm/integrations/otel/mappers/genai.py | 3 + litellm/integrations/otel/mappers/langfuse.py | 3 +- litellm/integrations/otel/model/semconv.py | 6 + .../integrations/otel/model/trace_controls.py | 7 +- .../test_otel_v2_user_id_every_callback.py | 110 ++++++++++++++++++ .../integrations/otel/test_otel_v2_logger.py | 17 +++ .../otel/test_otel_v2_sources_of_truth.py | 30 +++++ 7 files changed, 172 insertions(+), 4 deletions(-) create mode 100644 tests/integration/observability/test_otel_v2_user_id_every_callback.py diff --git a/litellm/integrations/otel/mappers/genai.py b/litellm/integrations/otel/mappers/genai.py index 0ac8b47e239..74fd26f1d6c 100644 --- a/litellm/integrations/otel/mappers/genai.py +++ b/litellm/integrations/otel/mappers/genai.py @@ -36,6 +36,7 @@ from litellm.integrations.otel.model.semconv import ( LiteLLM, RpcSystem, Server, + User, ) from litellm.integrations.otel.model.spans import postgres_operation @@ -46,6 +47,8 @@ class GenAIMapper: GenAI.PROVIDER_NAME: lambda d: d.provider or None, GenAI.OUTPUT_TYPE: lambda d: d.output_type.value if d.output_type else None, GenAI.CONVERSATION_ID: lambda d: d.session_id, + # The caller's ``trace_user_id``, else the proxy end user, for every backend that reads ``user.id``. + User.ID: lambda d: d.trace.user_id, GenAI.REQUEST_MODEL: lambda d: d.request_model or None, GenAI.REQUEST_TEMPERATURE: lambda d: d.request_params.temperature, GenAI.REQUEST_TOP_P: lambda d: d.request_params.top_p, diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index fab852894ef..b192c5e5d9d 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -27,12 +27,13 @@ from litellm.integrations.otel.model.payloads import ( LLMRequestParams, LLMUsage, ) +from litellm.integrations.otel.model.semconv import User from litellm.integrations.otel.model.trace_controls import TraceControls 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_USER_ID: Final = User.ID LANGFUSE_TRACE_SESSION_ID: Final = "session.id" LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags" diff --git a/litellm/integrations/otel/model/semconv.py b/litellm/integrations/otel/model/semconv.py index 38cf7266593..179462e5701 100644 --- a/litellm/integrations/otel/model/semconv.py +++ b/litellm/integrations/otel/model/semconv.py @@ -254,6 +254,12 @@ class Server: PORT: Final = "server.port" +class User: + """OTel ``user.*`` registry keys. OpenInference and Langfuse read the same ``user.id``.""" + + ID: Final = "user.id" + + class DB: """Database / cache client-span keys (OTel ``db.*`` semconv). diff --git a/litellm/integrations/otel/model/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py index d349467261d..99ce6a49c3f 100644 --- a/litellm/integrations/otel/model/trace_controls.py +++ b/litellm/integrations/otel/model/trace_controls.py @@ -1,4 +1,4 @@ -"""Langfuse trace controls, parsed from the live callback kwargs.""" +"""The request's trace controls, parsed from the live callback kwargs.""" from __future__ import annotations @@ -18,8 +18,9 @@ _ITEMS: Final = TypeAdapter(tuple[object, ...]) 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_`` headers winning over the - body for the scalar ones. ``request_trace_controls`` fills a missing ``user_id`` with the proxy end user. Mutation - controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are deliberately not carried.""" + body for the scalar ones. ``request_trace_controls`` fills a missing ``user_id`` with the proxy end user, which + every v2 callback then sends as ``user.id``. Mutation controls (``trace_id``, ``existing_trace_id``, + ``update_trace_keys``) are deliberately not carried.""" name: str | None = None user_id: str | None = None diff --git a/tests/integration/observability/test_otel_v2_user_id_every_callback.py b/tests/integration/observability/test_otel_v2_user_id_every_callback.py new file mode 100644 index 00000000000..9ea3e470dba --- /dev/null +++ b/tests/integration/observability/test_otel_v2_user_id_every_callback.py @@ -0,0 +1,110 @@ +import itertools +import json +import uuid +from collections.abc import Mapping +from pathlib import Path +from typing import Final + +import pytest +from _openinference_support import _chat_plain_response, _rig, _spans +from integration._support.client import Gateway, eventually +from integration._support.wire import Reply, Request, Wire +from pydantic import JsonValue + +_OPENINFERENCE_CALLBACKS: Final = ("arize", "arize_phoenix", "weave_otel") +_GENAI_CALLBACKS: Final = ("otel", "langfuse_otel", "langtrace", "signoz", "levo", "newrelic", "agentops") +_UNUSED_ENDPOINT: Final[dict[str, JsonValue]] = {"otel": {"exporter": "http/protobuf", "endpoint": "unused"}} + + +def _settings(callback: str) -> Mapping[str, JsonValue] | None: + if callback == "arize": + return None + if callback in _OPENINFERENCE_CALLBACKS: + return _UNUSED_ENDPOINT + return {"otel": {"exporter": "http/protobuf", "endpoint": "unused", "mapper_names": ["genai"]}} + + +def _environment(callback: str) -> dict[str, str]: + return { + "PHOENIX_PROJECT_NAME": "integration", + "WANDB_BASE_URL": "http://127.0.0.1", + "LANGFUSE_HOST": "http://127.0.0.1", + **( + {"HTTPS_PROXY": "http://127.0.0.1:0", "https_proxy": "http://127.0.0.1:0", "NO_PROXY": "", "no_proxy": ""} + if callback == "agentops" + else {} + ), + } + + +def _removed_environment(callback: str) -> tuple[str, ...]: + if callback == "newrelic": + return ("NEW_RELIC_LICENSE_KEY",) + if callback == "agentops": + return ("AGENTOPS_API_KEY",) + return () + + +def _upstream(request: Request) -> Reply: + marker: Final = json.loads(request.body)["messages"][0]["content"] + return _chat_plain_response(marker, "ok") + + +def _model_call_spans(destination: Wire, markers: tuple[str, ...]) -> Mapping[str, tuple[dict[str, str], ...]]: + seen: Final[list[Request]] = [] + + def by_marker() -> Mapping[str, tuple[dict[str, str], ...]]: + seen.extend(destination.drain()) + spans: Final = tuple(attributes for attributes in _spans(tuple(seen)) if "gen_ai.response.id" in attributes) + return { + marker: tuple(attributes for attributes in spans if attributes["gen_ai.response.id"] == marker) + for marker in markers + } + + return eventually(by_marker, lambda found: all(found.values()), seconds=30) + + +@pytest.mark.parametrize("callback", (*_OPENINFERENCE_CALLBACKS, *_GENAI_CALLBACKS)) +def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gateway: Gateway, tmp_path: Path) -> None: + run: Final = uuid.uuid4().hex + end_user: Final = f"end-user-{run}" + caller: Final = f"caller-{run}" + markers: Final = (f"end-{run}", f"caller-{run}-marker", f"owner-{run}") + with ( + _rig( + gateway, + tmp_path, + _upstream, + callbacks=(callback,), + callback_settings=_settings(callback), + environment=_environment(callback), + remove_environment=_removed_environment(callback), + ) as rig, + rig.proxy.scenario() as scenario, + ): + service_key: Final = scenario.key() + owned_key: Final = scenario.key(user_id=scenario.user()) + sent: Final = ( + (markers[0], service_key, {"x-litellm-end-user-id": end_user}, {}), + (markers[1], service_key, {"x-litellm-end-user-id": end_user}, {"metadata": {"trace_user_id": caller}}), + (markers[2], owned_key, {}, {}), + ) + for marker, key, headers, extra in sent: + response = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": [{"role": "user", "content": marker}], + "cache": {"no-cache": True}, + **extra, + }, + key=key, + headers=headers, + ) + assert response.status_code == 200, response.text + spans: Final = _model_call_spans(rig.destination, markers) + users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers} + assert users == {markers[0]: {end_user}, markers[1]: {caller}, markers[2]: {None}}, spans + sessions: Final = {attributes.get("session.id") for attributes in itertools.chain.from_iterable(spans.values())} + assert end_user not in sessions, spans diff --git a/tests/unit/integrations/otel/test_otel_v2_logger.py b/tests/unit/integrations/otel/test_otel_v2_logger.py index 6a9a963948e..8a9eae51fb3 100644 --- a/tests/unit/integrations/otel/test_otel_v2_logger.py +++ b/tests/unit/integrations/otel/test_otel_v2_logger.py @@ -47,6 +47,7 @@ from litellm.integrations.otel import ( # noqa: E402 from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402 from litellm.integrations.otel.model.config import CaptureMessageContent, ExporterSpec # noqa: E402 from litellm.integrations.otel.model.destination import OtelDestination # noqa: E402 +from litellm.integrations.otel.model.semconv import User # noqa: E402 from litellm.integrations.otel.model.spans import ( # noqa: E402 LITELLM_PROXY_REQUEST_SPAN_NAME, SpanRole, @@ -301,6 +302,22 @@ def test_replayed_llm_call_span_does_not_take_the_payloads_session_id(): assert GenAI.CONVERSATION_ID not in span.attributes +@pytest.mark.parametrize( + ("metadata", "expected"), + [ + ({"user_api_key_end_user_id": "end-1"}, "end-1"), + ({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"), + ({"user_api_key_user_id": "internal-1"}, None), + ], + ids=["proxy-end-user", "caller-trace-user-wins", "internal-user-never-sent"], +) +def test_llm_call_span_carries_the_request_user_as_user_id(metadata, expected): + logger, exporter = _logger() + _emit_llm(logger, {**_kwargs(), "litellm_params": {"metadata": metadata}}) + (span,) = exporter.get_finished_spans() + assert span.attributes.get(User.ID) == expected + + def test_streaming_span_carries_time_to_first_chunk(): logger, exporter = _logger() kwargs = { 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 2435c122a70..e6bef788d4c 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 @@ -29,6 +29,7 @@ from litellm.integrations.otel import ( resolve_output_type, resolve_provider, ) +from litellm.integrations.otel.mappers import resolve_mappers 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 @@ -38,6 +39,7 @@ from litellm.integrations.otel.model.payloads import ( RequestIdentity, _upstream_address_port, ) +from litellm.integrations.otel.model.semconv import User from litellm.integrations.otel.model.spans import ( SPAN_REGISTRY, LiteLLMSpanKind, @@ -51,6 +53,7 @@ from litellm.integrations.otel.model.trace_controls import ( caller_trace_controls, request_trace_controls, ) +from litellm.integrations.otel.presets import PRESET_BY_CALLBACK @pytest.fixture(autouse=True) @@ -1540,6 +1543,33 @@ def test_llm_span_stamps_gen_ai_conversation_id_only_when_the_caller_sent_one(): assert GenAI.CONVERSATION_ID not in GenAIMapper().map(without) +def test_llm_span_stamps_user_id_only_when_the_request_names_a_user(): + with_user: Final = LLMCallSpanData.from_standard_logging_payload( + _sample_payload(), trace=TraceControls(user_id="end-1") + ) + assert GenAIMapper().map(with_user)[User.ID] == "end-1" + + without: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload()) + assert User.ID not in GenAIMapper().map(without) + + +@pytest.mark.parametrize("callback_name", ["otel", *sorted(PRESET_BY_CALLBACK)]) +def test_every_v2_callback_sends_the_request_user_as_user_id(callback_name, monkeypatch): + for name in ("LEVOAI_API_KEY", "LEVOAI_ORG_ID", "LEVOAI_WORKSPACE_ID"): + monkeypatch.setenv(name, "levo-test") + monkeypatch.setenv("LEVOAI_COLLECTOR_URL", "http://127.0.0.1:4318") + config: Final = ( + OpenTelemetryV2Config() + if callback_name == "otel" + else PRESET_BY_CALLBACK[callback_name](allow_missing_credentials=True) + ) + data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace=TraceControls(user_id="end-1")) + attrs: Final = { + key: value for mapper in resolve_mappers(config.mapper_names) for key, value in mapper.map(data).items() + } + assert attrs[User.ID] == "end-1" + + def test_llm_span_carries_proxy_request_route(): """The LLM span records the proxy route the request arrived on, so it can be filtered by endpoint (``/v1/responses`` vs ``/v1/chat/completions``) without From abc7fe5a67e9b6488b9b4da59cf15e78daf70f70 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:45:49 -0700 Subject: [PATCH 2/3] fix(otel v2): keep the end user off user.id when the metadata allowlist drops it The end-user fallback only reaches user.id, on the model-call span and the Langfuse root, when baggage_metadata_keys includes user_api_key_end_user_id. A caller's own trace_user_id is always sent. Drops a comment the repo policy does not allow. --- litellm/integrations/otel/langfuse_logger.py | 6 +- litellm/integrations/otel/mappers/genai.py | 1 - litellm/integrations/otel/model/payloads.py | 2 +- .../integrations/otel/model/trace_controls.py | 15 ++++- .../test_otel_v2_user_id_every_callback.py | 63 ++++++++++++++----- .../integrations/otel/test_langfuse_logger.py | 25 ++++++++ .../integrations/otel/test_otel_v2_logger.py | 18 ++++++ .../otel/test_otel_v2_sources_of_truth.py | 25 +++++++- 8 files changed, 131 insertions(+), 24 deletions(-) diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index ead0f0b7981..5efdc9df91d 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -31,7 +31,11 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2): def log_pre_api_call(self, model: str, messages: object, kwargs: Mapping[str, object]) -> None: root: Final = request_root_span() if root is not None and root.is_recording(): - controls: Final = request_trace_controls(kwargs) if self._started(root) else caller_trace_controls(kwargs) + controls: Final = ( + request_trace_controls(kwargs).released(self.config.baggage_metadata_keys) + if self._started(root) + else caller_trace_controls(kwargs) + ) root.set_attributes(LangfuseMapper.trace_attributes(controls)) super().log_pre_api_call(model, messages, kwargs) diff --git a/litellm/integrations/otel/mappers/genai.py b/litellm/integrations/otel/mappers/genai.py index 74fd26f1d6c..74b51b975b3 100644 --- a/litellm/integrations/otel/mappers/genai.py +++ b/litellm/integrations/otel/mappers/genai.py @@ -47,7 +47,6 @@ class GenAIMapper: GenAI.PROVIDER_NAME: lambda d: d.provider or None, GenAI.OUTPUT_TYPE: lambda d: d.output_type.value if d.output_type else None, GenAI.CONVERSATION_ID: lambda d: d.session_id, - # The caller's ``trace_user_id``, else the proxy end user, for every backend that reads ``user.id``. User.ID: lambda d: d.trace.user_id, GenAI.REQUEST_MODEL: lambda d: d.request_model or None, GenAI.REQUEST_TEMPERATURE: lambda d: d.request_params.temperature, diff --git a/litellm/integrations/otel/model/payloads.py b/litellm/integrations/otel/model/payloads.py index 101e6285946..d20460a00c4 100644 --- a/litellm/integrations/otel/model/payloads.py +++ b/litellm/integrations/otel/model/payloads.py @@ -503,7 +503,7 @@ class LLMCallSpanData: call_type=call_type or None, request_route=request_route or context.identity.request_route, request_purpose=request_purpose, - trace=trace or TraceControls(), + trace=(trace or TraceControls()).released(metadata_keys), session_id=session_id or None, embedding_output=embedding_output if capture_content else None, routing_attributes=routing_decision_attributes(decision), diff --git a/litellm/integrations/otel/model/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py index 99ce6a49c3f..f1fff0c853d 100644 --- a/litellm/integrations/otel/model/trace_controls.py +++ b/litellm/integrations/otel/model/trace_controls.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Collection, Mapping from dataclasses import dataclass, replace from typing import Final @@ -11,6 +11,7 @@ from pydantic import TypeAdapter, ValidationError from litellm.integrations.otel.model.utils import as_str, as_str_mapping LANGFUSE_HEADER_PREFIX: Final = "langfuse_" +END_USER_METADATA_KEY: Final = "user_api_key_end_user_id" _ITEMS: Final = TypeAdapter(tuple[object, ...]) @@ -26,6 +27,14 @@ class TraceControls: user_id: str | None = None session_id: str | None = None tags: tuple[str, ...] = () + user_id_is_end_user: bool = False + + def released(self, metadata_keys: Collection[str]) -> TraceControls: + """These controls minus an end-user ``user_id`` when the operator's ``baggage_metadata_keys`` keeps the + end-user id in the proxy. A caller's own ``trace_user_id`` is always kept.""" + if self.user_id_is_end_user and END_USER_METADATA_KEY not in metadata_keys: + return replace(self, user_id=None, user_id_is_end_user=False) + return self def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: @@ -58,10 +67,10 @@ def request_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(END_USER_METADATA_KEY)))), None, ) - return replace(controls, user_id=end_user) + return replace(controls, user_id=end_user, user_id_is_end_user=end_user is not None) def _metadata_bodies(request: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: diff --git a/tests/integration/observability/test_otel_v2_user_id_every_callback.py b/tests/integration/observability/test_otel_v2_user_id_every_callback.py index 9ea3e470dba..adc760fea07 100644 --- a/tests/integration/observability/test_otel_v2_user_id_every_callback.py +++ b/tests/integration/observability/test_otel_v2_user_id_every_callback.py @@ -6,7 +6,7 @@ from pathlib import Path from typing import Final import pytest -from _openinference_support import _chat_plain_response, _rig, _spans +from _openinference_support import Rig, _chat_plain_response, _rig, _spans from integration._support.client import Gateway, eventually from integration._support.wire import Reply, Request, Wire from pydantic import JsonValue @@ -14,6 +14,7 @@ from pydantic import JsonValue _OPENINFERENCE_CALLBACKS: Final = ("arize", "arize_phoenix", "weave_otel") _GENAI_CALLBACKS: Final = ("otel", "langfuse_otel", "langtrace", "signoz", "levo", "newrelic", "agentops") _UNUSED_ENDPOINT: Final[dict[str, JsonValue]] = {"otel": {"exporter": "http/protobuf", "endpoint": "unused"}} +_METADATA_ALLOWLIST: Final = "LITELLM_OTEL_BAGGAGE_METADATA_KEYS" def _settings(callback: str) -> Mapping[str, JsonValue] | None: @@ -64,6 +65,17 @@ def _model_call_spans(destination: Wire, markers: tuple[str, ...]) -> Mapping[st return eventually(by_marker, lambda found: all(found.values()), seconds=30) +def _chat(rig: Rig, marker: str, key: str, headers: Mapping[str, str], extra: Mapping[str, JsonValue]) -> None: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + {"model": rig.model, "messages": [{"role": "user", "content": marker}], "cache": {"no-cache": True}, **extra}, + key=key, + headers=headers, + ) + assert response.status_code == 200, response.text + + @pytest.mark.parametrize("callback", (*_OPENINFERENCE_CALLBACKS, *_GENAI_CALLBACKS)) def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gateway: Gateway, tmp_path: Path) -> None: run: Final = uuid.uuid4().hex @@ -78,7 +90,8 @@ def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gatew callbacks=(callback,), callback_settings=_settings(callback), environment=_environment(callback), - remove_environment=_removed_environment(callback), + remove_environment=(*_removed_environment(callback), _METADATA_ALLOWLIST), + disabled_environment=(_METADATA_ALLOWLIST,), ) as rig, rig.proxy.scenario() as scenario, ): @@ -90,21 +103,41 @@ def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gatew (markers[2], owned_key, {}, {}), ) for marker, key, headers, extra in sent: - response = rig.proxy.request( - "POST", - "/v1/chat/completions", - { - "model": rig.model, - "messages": [{"role": "user", "content": marker}], - "cache": {"no-cache": True}, - **extra, - }, - key=key, - headers=headers, - ) - assert response.status_code == 200, response.text + _chat(rig, marker, key, headers, extra) spans: Final = _model_call_spans(rig.destination, markers) users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers} assert users == {markers[0]: {end_user}, markers[1]: {caller}, markers[2]: {None}}, spans sessions: Final = {attributes.get("session.id") for attributes in itertools.chain.from_iterable(spans.values())} assert end_user not in sessions, spans + + +@pytest.mark.parametrize("callback", ("arize", "otel", "langfuse_otel")) +def test_otel_v2_callback_withholds_an_end_user_the_metadata_allowlist_drops( + callback: str, gateway: Gateway, tmp_path: Path +) -> None: + run: Final = uuid.uuid4().hex + end_user: Final = f"end-user-{run}" + caller: Final = f"caller-{run}" + markers: Final = (f"end-{run}", f"caller-{run}-marker") + with ( + _rig( + gateway, + tmp_path, + _upstream, + callbacks=(callback,), + callback_settings=_settings(callback), + environment={**_environment(callback), _METADATA_ALLOWLIST: "user_api_key_org_id"}, + ) as rig, + rig.proxy.scenario() as scenario, + ): + service_key: Final = scenario.key() + _chat(rig, markers[0], service_key, {"x-litellm-end-user-id": end_user}, {}) + _chat( + rig, markers[1], service_key, {"x-litellm-end-user-id": end_user}, {"metadata": {"trace_user_id": caller}} + ) + spans: Final = _model_call_spans(rig.destination, markers) + users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers} + assert users == {markers[0]: {None}, markers[1]: {caller}}, spans + model_call_spans: Final = tuple(itertools.chain.from_iterable(spans.values())) + values: Final = tuple(itertools.chain.from_iterable(span.values() for span in model_call_spans)) + assert not [value for value in values if end_user in str(value)], spans diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py index c8c9c7c0ccb..6b53b048d9a 100644 --- a/tests/unit/integrations/otel/test_langfuse_logger.py +++ b/tests/unit/integrations/otel/test_langfuse_logger.py @@ -466,6 +466,31 @@ def test_a_callers_trace_user_id_still_wins_over_the_proxy_end_user(): assert generation_attrs["user.id"] == "caller-1" +@pytest.mark.parametrize( + ("metadata", "expected_user"), + [ + ({"user_api_key_end_user_id": "end-1"}, None), + ({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"), + ], + ids=["end-user-withheld", "caller-trace-user-still-sent"], +) +def test_an_end_user_the_metadata_allowlist_drops_stays_off_the_root_and_the_generation(metadata, expected_user): + cfg: Final = OpenTelemetryV2Config( + exporter="in_memory", mapper_names=["genai", "langfuse"], baggage_metadata_keys=["user_api_key_org_id"] + ) + exporter: Final = InMemorySpanExporter() + logger: Final = build_otel_v2_logger( + config=cfg, tracer_provider=providers.build_tracer_provider(cfg, exporter=exporter) + ) + + root_attrs, generation_attrs = _run_named_request( + logger, exporter, {"metadata": metadata, "proxy_server_request": {"headers": {}}} + ) + + assert root_attrs.get("user.id") == expected_user + assert generation_attrs.get("user.id") == expected_user + + @pytest.mark.parametrize( ("metadata", "expected_root_user"), [ diff --git a/tests/unit/integrations/otel/test_otel_v2_logger.py b/tests/unit/integrations/otel/test_otel_v2_logger.py index 8a9eae51fb3..2cc9cff5899 100644 --- a/tests/unit/integrations/otel/test_otel_v2_logger.py +++ b/tests/unit/integrations/otel/test_otel_v2_logger.py @@ -318,6 +318,24 @@ def test_llm_call_span_carries_the_request_user_as_user_id(metadata, expected): assert span.attributes.get(User.ID) == expected +@pytest.mark.parametrize( + ("metadata", "expected"), + [ + ({"user_api_key_end_user_id": "end-1"}, None), + ({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"), + ], + ids=["end-user-withheld", "caller-trace-user-still-sent"], +) +def test_llm_call_span_withholds_an_end_user_the_metadata_allowlist_drops(metadata, expected): + cfg: Final = OpenTelemetryV2Config(exporter="in_memory", baggage_metadata_keys=["user_api_key_org_id"]) + exporter: Final = InMemorySpanExporter() + logger: Final = OpenTelemetryV2(config=cfg, tracer_provider=providers.build_tracer_provider(cfg, exporter=exporter)) + _emit_llm(logger, {**_kwargs(), "litellm_params": {"metadata": metadata}}) + (span,) = exporter.get_finished_spans() + assert span.attributes.get(User.ID) == expected + assert "end-1" not in {str(value) for value in span.attributes.values()} + + def test_streaming_span_carries_time_to_first_chunk(): logger, exporter = _logger() kwargs = { 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 e6bef788d4c..21b7733a5dd 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 @@ -1358,8 +1358,11 @@ def test_caller_trace_controls_carry_user_session_and_tags(request_data, expecte @pytest.mark.parametrize( ("request_data", "expected"), [ - ({"metadata": {"user_api_key_end_user_id": "end-1"}}, TraceControls(user_id="end-1")), - ({"litellm_metadata": {"user_api_key_end_user_id": "end-2"}}, TraceControls(user_id="end-2")), + ({"metadata": {"user_api_key_end_user_id": "end-1"}}, TraceControls(user_id="end-1", user_id_is_end_user=True)), + ( + {"litellm_metadata": {"user_api_key_end_user_id": "end-2"}}, + TraceControls(user_id="end-2", user_id_is_end_user=True), + ), ( {"metadata": {"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}}, TraceControls(user_id="caller-1"), @@ -1373,7 +1376,7 @@ def test_caller_trace_controls_carry_user_session_and_tags(request_data, expecte ), ( {"metadata": {"user_api_key_end_user_id": "end-1", "session_id": "s-1"}}, - TraceControls(user_id="end-1", session_id="s-1"), + TraceControls(user_id="end-1", session_id="s-1", user_id_is_end_user=True), ), ({"metadata": {"user_api_key_user_id": "internal-1"}}, TraceControls()), ({}, TraceControls()), @@ -1553,6 +1556,22 @@ def test_llm_span_stamps_user_id_only_when_the_request_names_a_user(): assert User.ID not in GenAIMapper().map(without) +@pytest.mark.parametrize( + ("trace", "metadata_keys", "expected"), + [ + (TraceControls(user_id="end-1", user_id_is_end_user=True), ("user_api_key_end_user_id",), "end-1"), + (TraceControls(user_id="end-1", user_id_is_end_user=True), ("user_api_key_org_id",), None), + (TraceControls(user_id="caller-1"), ("user_api_key_org_id",), "caller-1"), + ], + ids=["end-user-allowlisted", "end-user-withheld", "caller-trace-user-always-sent"], +) +def test_llm_span_sends_the_end_user_only_when_the_metadata_allowlist_releases_it(trace, metadata_keys, expected): + data: Final = LLMCallSpanData.from_standard_logging_payload( + _sample_payload(), trace=trace, metadata_keys=metadata_keys + ) + assert GenAIMapper().map(data).get(User.ID) == expected + + @pytest.mark.parametrize("callback_name", ["otel", *sorted(PRESET_BY_CALLBACK)]) def test_every_v2_callback_sends_the_request_user_as_user_id(callback_name, monkeypatch): for name in ("LEVOAI_API_KEY", "LEVOAI_ORG_ID", "LEVOAI_WORKSPACE_ID"): From c45d05ca23275bcec2aaf8d1ec4e7cb944613e97 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Mon, 19 Oct 2026 13:29:17 -0700 Subject: [PATCH 3/3] fix(otel v2): send only the already-exported end user as user.id outside Langfuse A caller's trace_user_id stays a Langfuse-only control, so non-Langfuse collectors get no value they did not already receive as litellm.metadata.user_api_key_end_user_id. user.id also sheds first at the span attribute limit, so it never costs an indexed message attribute. --- litellm/integrations/otel/emitter.py | 11 ++- litellm/integrations/otel/mappers/genai.py | 3 +- .../otel/mappers/openinference.py | 10 ++- .../integrations/otel/model/trace_controls.py | 5 +- .../test_otel_v2_user_id_every_callback.py | 21 +++-- .../integrations/otel/test_otel_v2_logger.py | 77 +++++++++++++++---- .../otel/test_otel_v2_sources_of_truth.py | 45 ++++++----- 7 files changed, 116 insertions(+), 56 deletions(-) diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index a4cb0a84541..31e60342606 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 LangfuseMapper 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 ( @@ -23,7 +24,7 @@ from litellm.integrations.otel.model.payloads import ( ServiceSpanData, SpanError, ) -from litellm.integrations.otel.model.semconv import Error, ExceptionEvent, LiteLLMError +from litellm.integrations.otel.model.semconv import Error, ExceptionEvent, LiteLLMError, User from litellm.integrations.otel.model.spans import ( SPAN_REGISTRY, SpanRole, @@ -171,6 +172,11 @@ class SpanEmitter: self._mappers: list[AttributeMapper] = ( list(mappers) if mappers is not None else resolve_mappers(config.mapper_names) ) + self._shed_first: Final = ( + frozenset[str]() + if any(isinstance(mapper, LangfuseMapper) for mapper in self._mappers) + else frozenset({User.ID}) + ) # Bounded LRU (ordered by insertion / most-recent touch). Storing keys # only — the value is unused — so it behaves like a capped set. self._emitted: OrderedDict[tuple[str, SpanRole], None] = OrderedDict() @@ -303,7 +309,8 @@ class SpanEmitter: ) stamped_later: Final = error_attributes(error) if error else _NO_ATTRIBUTES reserved: Final = len(stamped_later.keys() - mapped.keys()) - for key, value in fit_indexed_messages(mapped, attribute_budget(span, reserved, frozenset(mapped))).items(): + budget: Final = attribute_budget(span, reserved, frozenset(mapped)) + for key, value in fit_indexed_messages(mapped, budget, self._shed_first).items(): span.set_attribute(key, value) if error: stamped: Final = stamp_error(span, error) diff --git a/litellm/integrations/otel/mappers/genai.py b/litellm/integrations/otel/mappers/genai.py index 74b51b975b3..622e94e2d34 100644 --- a/litellm/integrations/otel/mappers/genai.py +++ b/litellm/integrations/otel/mappers/genai.py @@ -39,6 +39,7 @@ from litellm.integrations.otel.model.semconv import ( User, ) from litellm.integrations.otel.model.spans import postgres_operation +from litellm.integrations.otel.model.trace_controls import END_USER_METADATA_KEY class GenAIMapper: @@ -47,7 +48,7 @@ class GenAIMapper: GenAI.PROVIDER_NAME: lambda d: d.provider or None, GenAI.OUTPUT_TYPE: lambda d: d.output_type.value if d.output_type else None, GenAI.CONVERSATION_ID: lambda d: d.session_id, - User.ID: lambda d: d.trace.user_id, + User.ID: lambda d: d.promoted_metadata.get(END_USER_METADATA_KEY), GenAI.REQUEST_MODEL: lambda d: d.request_model or None, GenAI.REQUEST_TEMPERATURE: lambda d: d.request_params.temperature, GenAI.REQUEST_TOP_P: lambda d: d.request_params.top_p, 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/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py index f1fff0c853d..ad9fec8492d 100644 --- a/litellm/integrations/otel/model/trace_controls.py +++ b/litellm/integrations/otel/model/trace_controls.py @@ -19,9 +19,8 @@ _ITEMS: Final = TypeAdapter(tuple[object, ...]) 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_`` headers winning over the - body for the scalar ones. ``request_trace_controls`` fills a missing ``user_id`` with the proxy end user, which - every v2 callback then sends as ``user.id``. Mutation controls (``trace_id``, ``existing_trace_id``, - ``update_trace_keys``) are deliberately not carried.""" + body for the scalar ones. ``request_trace_controls`` fills a missing ``user_id`` with the proxy end user. Mutation + controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are deliberately not carried.""" name: str | None = None user_id: str | None = None diff --git a/tests/integration/observability/test_otel_v2_user_id_every_callback.py b/tests/integration/observability/test_otel_v2_user_id_every_callback.py index adc760fea07..96b4e794a1a 100644 --- a/tests/integration/observability/test_otel_v2_user_id_every_callback.py +++ b/tests/integration/observability/test_otel_v2_user_id_every_callback.py @@ -15,6 +15,7 @@ _OPENINFERENCE_CALLBACKS: Final = ("arize", "arize_phoenix", "weave_otel") _GENAI_CALLBACKS: Final = ("otel", "langfuse_otel", "langtrace", "signoz", "levo", "newrelic", "agentops") _UNUSED_ENDPOINT: Final[dict[str, JsonValue]] = {"otel": {"exporter": "http/protobuf", "endpoint": "unused"}} _METADATA_ALLOWLIST: Final = "LITELLM_OTEL_BAGGAGE_METADATA_KEYS" +_LANGFUSE_CALLBACK: Final = "langfuse_otel" def _settings(callback: str) -> Mapping[str, JsonValue] | None: @@ -65,6 +66,11 @@ def _model_call_spans(destination: Wire, markers: tuple[str, ...]) -> Mapping[st return eventually(by_marker, lambda found: all(found.values()), seconds=30) +def _carrying(spans: Mapping[str, tuple[dict[str, str], ...]], value: str) -> tuple[str, ...]: + model_call_spans: Final = itertools.chain.from_iterable(spans.values()) + return tuple(key for span in model_call_spans for key, attribute in span.items() if value in str(attribute)) + + def _chat(rig: Rig, marker: str, key: str, headers: Mapping[str, str], extra: Mapping[str, JsonValue]) -> None: response: Final = rig.proxy.request( "POST", @@ -81,7 +87,7 @@ def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gatew run: Final = uuid.uuid4().hex end_user: Final = f"end-user-{run}" caller: Final = f"caller-{run}" - markers: Final = (f"end-{run}", f"caller-{run}-marker", f"owner-{run}") + markers: Final = (f"end-{run}", f"trace-{run}", f"owner-{run}") with ( _rig( gateway, @@ -106,9 +112,11 @@ def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gatew _chat(rig, marker, key, headers, extra) spans: Final = _model_call_spans(rig.destination, markers) users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers} - assert users == {markers[0]: {end_user}, markers[1]: {caller}, markers[2]: {None}}, spans + caller_user: Final = caller if callback == _LANGFUSE_CALLBACK else end_user + assert users == {markers[0]: {end_user}, markers[1]: {caller_user}, markers[2]: {None}}, spans sessions: Final = {attributes.get("session.id") for attributes in itertools.chain.from_iterable(spans.values())} assert end_user not in sessions, spans + assert callback == _LANGFUSE_CALLBACK or not _carrying(spans, caller), spans @pytest.mark.parametrize("callback", ("arize", "otel", "langfuse_otel")) @@ -118,7 +126,7 @@ def test_otel_v2_callback_withholds_an_end_user_the_metadata_allowlist_drops( run: Final = uuid.uuid4().hex end_user: Final = f"end-user-{run}" caller: Final = f"caller-{run}" - markers: Final = (f"end-{run}", f"caller-{run}-marker") + markers: Final = (f"end-{run}", f"trace-{run}") with ( _rig( gateway, @@ -137,7 +145,6 @@ def test_otel_v2_callback_withholds_an_end_user_the_metadata_allowlist_drops( ) spans: Final = _model_call_spans(rig.destination, markers) users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers} - assert users == {markers[0]: {None}, markers[1]: {caller}}, spans - model_call_spans: Final = tuple(itertools.chain.from_iterable(spans.values())) - values: Final = tuple(itertools.chain.from_iterable(span.values() for span in model_call_spans)) - assert not [value for value in values if end_user in str(value)], spans + assert users == {markers[0]: {None}, markers[1]: {caller if callback == _LANGFUSE_CALLBACK else None}}, spans + assert not _carrying(spans, end_user), spans + assert callback == _LANGFUSE_CALLBACK or not _carrying(spans, caller), spans diff --git a/tests/unit/integrations/otel/test_otel_v2_logger.py b/tests/unit/integrations/otel/test_otel_v2_logger.py index 2cc9cff5899..98537616488 100644 --- a/tests/unit/integrations/otel/test_otel_v2_logger.py +++ b/tests/unit/integrations/otel/test_otel_v2_logger.py @@ -45,6 +45,7 @@ from litellm.integrations.otel import ( # noqa: E402 OpenTelemetryV2Config, ) from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402 +from litellm.integrations.otel.mappers.genai import GenAIMapper # noqa: E402 from litellm.integrations.otel.model.config import CaptureMessageContent, ExporterSpec # noqa: E402 from litellm.integrations.otel.model.destination import OtelDestination # noqa: E402 from litellm.integrations.otel.model.semconv import User # noqa: E402 @@ -302,38 +303,80 @@ def test_replayed_llm_call_span_does_not_take_the_payloads_session_id(): assert GenAI.CONVERSATION_ID not in span.attributes +def _user_kwargs(metadata, messages=None, call_id="call_1"): + payload: Final = _payload(metadata={**_payload()["metadata"], **metadata}, litellm_call_id=call_id) + if messages is not None: + payload["messages"] = messages + return {**_kwargs(payload), "litellm_params": {"metadata": metadata}, "messages": messages or []} + + +def _logger_with(**config): + cfg: Final = OpenTelemetryV2Config(exporter="in_memory", **config) + exporter: Final = InMemorySpanExporter() + return OpenTelemetryV2( + config=cfg, tracer_provider=providers.build_tracer_provider(cfg, exporter=exporter) + ), exporter + + @pytest.mark.parametrize( ("metadata", "expected"), [ ({"user_api_key_end_user_id": "end-1"}, "end-1"), - ({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"), + ({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "end-1"), + ({"trace_user_id": "caller-1"}, None), ({"user_api_key_user_id": "internal-1"}, None), ], - ids=["proxy-end-user", "caller-trace-user-wins", "internal-user-never-sent"], + ids=["proxy-end-user", "caller-trace-user-not-sent", "caller-trace-user-alone", "internal-user-never-sent"], ) -def test_llm_call_span_carries_the_request_user_as_user_id(metadata, expected): +def test_llm_call_span_carries_the_end_user_as_user_id(metadata, expected): logger, exporter = _logger() - _emit_llm(logger, {**_kwargs(), "litellm_params": {"metadata": metadata}}) + _emit_llm(logger, _user_kwargs(metadata)) (span,) = exporter.get_finished_spans() assert span.attributes.get(User.ID) == expected + assert "caller-1" not in {str(value) for value in span.attributes.values()} @pytest.mark.parametrize( - ("metadata", "expected"), - [ - ({"user_api_key_end_user_id": "end-1"}, None), - ({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"), - ], - ids=["end-user-withheld", "caller-trace-user-still-sent"], + "metadata", + [{"user_api_key_end_user_id": "end-1"}, {"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}], + ids=["end-user", "end-user-and-caller-trace-user"], ) -def test_llm_call_span_withholds_an_end_user_the_metadata_allowlist_drops(metadata, expected): - cfg: Final = OpenTelemetryV2Config(exporter="in_memory", baggage_metadata_keys=["user_api_key_org_id"]) - exporter: Final = InMemorySpanExporter() - logger: Final = OpenTelemetryV2(config=cfg, tracer_provider=providers.build_tracer_provider(cfg, exporter=exporter)) - _emit_llm(logger, {**_kwargs(), "litellm_params": {"metadata": metadata}}) +def test_llm_call_span_withholds_an_end_user_the_metadata_allowlist_drops(metadata): + logger, exporter = _logger_with(baggage_metadata_keys=["user_api_key_org_id"]) + _emit_llm(logger, _user_kwargs(metadata)) (span,) = exporter.get_finished_spans() - assert span.attributes.get(User.ID) == expected - assert "end-1" not in {str(value) for value in span.attributes.values()} + assert User.ID not in span.attributes + assert not {"end-1", "caller-1"} & {str(value) for value in span.attributes.values()} + + +def test_user_id_costs_no_other_attribute_at_the_span_limit(monkeypatch): + def spans() -> list: + logger, exporter = _logger_with( + mapper_names=["genai", "openinference"], capture_message_content=CaptureMessageContent.SPAN_ONLY + ) + for count in range(40, 70): + messages = [{"role": "user", "content": f"m{index}"} for index in range(count)] + _emit_llm(logger, _user_kwargs({"user_api_key_end_user_id": "end-1"}, messages, f"call_{count}")) + return list(exporter.get_finished_spans()) + + with_user_id: Final = spans() + monkeypatch.delitem(GenAIMapper._LLM_CALL_ATTRS, User.ID) + without_user_id: Final = spans() + assert {span.attributes.get(User.ID) for span in with_user_id} == {"end-1", None} + for with_user, without in zip(with_user_id, without_user_id, strict=True): + assert set(with_user.attributes) - {User.ID} == set(without.attributes) + assert with_user.dropped_attributes == without.dropped_attributes + + +def test_langfuse_vocabulary_keeps_the_caller_trace_user_at_the_span_limit(): + logger, exporter = _logger_with( + mapper_names=["genai", "openinference", "langfuse"], capture_message_content=CaptureMessageContent.SPAN_ONLY + ) + messages: Final = [{"role": "user", "content": f"m{index}"} for index in range(80)] + _emit_llm(logger, _user_kwargs({"trace_user_id": "caller-1"}, messages)) + (span,) = exporter.get_finished_spans() + assert span.attributes[User.ID] == "caller-1" + assert span.dropped_attributes == 0 def test_streaming_span_carries_time_to_first_chunk(): 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 21b7733a5dd..1f4da25236c 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 @@ -1546,34 +1546,28 @@ def test_llm_span_stamps_gen_ai_conversation_id_only_when_the_caller_sent_one(): assert GenAI.CONVERSATION_ID not in GenAIMapper().map(without) -def test_llm_span_stamps_user_id_only_when_the_request_names_a_user(): - with_user: Final = LLMCallSpanData.from_standard_logging_payload( - _sample_payload(), trace=TraceControls(user_id="end-1") - ) - assert GenAIMapper().map(with_user)[User.ID] == "end-1" - - without: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload()) - assert User.ID not in GenAIMapper().map(without) +def _end_user_payload(): + payload: Final = _sample_payload() + return {**payload, "metadata": {**payload["metadata"], "user_api_key_end_user_id": "end-1"}} @pytest.mark.parametrize( - ("trace", "metadata_keys", "expected"), + ("payload", "trace", "metadata_keys", "expected"), [ - (TraceControls(user_id="end-1", user_id_is_end_user=True), ("user_api_key_end_user_id",), "end-1"), - (TraceControls(user_id="end-1", user_id_is_end_user=True), ("user_api_key_org_id",), None), - (TraceControls(user_id="caller-1"), ("user_api_key_org_id",), "caller-1"), + (_end_user_payload(), TraceControls(), ("user_api_key_end_user_id",), "end-1"), + (_end_user_payload(), TraceControls(), ("user_api_key_org_id",), None), + (_end_user_payload(), TraceControls(user_id="caller-1"), ("user_api_key_end_user_id",), "end-1"), + (_sample_payload(), TraceControls(user_id="caller-1"), ("user_api_key_end_user_id",), None), ], - ids=["end-user-allowlisted", "end-user-withheld", "caller-trace-user-always-sent"], + ids=["end-user-allowlisted", "end-user-withheld", "caller-trace-user-not-sent", "no-end-user"], ) -def test_llm_span_sends_the_end_user_only_when_the_metadata_allowlist_releases_it(trace, metadata_keys, expected): - data: Final = LLMCallSpanData.from_standard_logging_payload( - _sample_payload(), trace=trace, metadata_keys=metadata_keys - ) +def test_llm_span_sends_only_the_allowlisted_end_user_as_user_id(payload, trace, metadata_keys, expected): + data: Final = LLMCallSpanData.from_standard_logging_payload(payload, trace=trace, metadata_keys=metadata_keys) assert GenAIMapper().map(data).get(User.ID) == expected @pytest.mark.parametrize("callback_name", ["otel", *sorted(PRESET_BY_CALLBACK)]) -def test_every_v2_callback_sends_the_request_user_as_user_id(callback_name, monkeypatch): +def test_every_v2_callback_sends_the_end_user_and_only_langfuse_the_caller_trace_user(callback_name, monkeypatch): for name in ("LEVOAI_API_KEY", "LEVOAI_ORG_ID", "LEVOAI_WORKSPACE_ID"): monkeypatch.setenv(name, "levo-test") monkeypatch.setenv("LEVOAI_COLLECTOR_URL", "http://127.0.0.1:4318") @@ -1582,11 +1576,16 @@ def test_every_v2_callback_sends_the_request_user_as_user_id(callback_name, monk if callback_name == "otel" else PRESET_BY_CALLBACK[callback_name](allow_missing_credentials=True) ) - data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace=TraceControls(user_id="end-1")) - attrs: Final = { - key: value for mapper in resolve_mappers(config.mapper_names) for key, value in mapper.map(data).items() - } - assert attrs[User.ID] == "end-1" + mappers: Final = resolve_mappers(config.mapper_names) + + def user_id(trace: TraceControls) -> object: + data: Final = LLMCallSpanData.from_standard_logging_payload( + _end_user_payload(), trace=trace, metadata_keys=tuple(config.baggage_metadata_keys) + ) + return {key: value for mapper in mappers for key, value in mapper.map(data).items()}.get(User.ID) + + assert user_id(TraceControls(user_id="end-1", user_id_is_end_user=True)) == "end-1" + assert user_id(TraceControls(user_id="caller-1")) == ("caller-1" if callback_name == "langfuse_otel" else "end-1") def test_llm_span_carries_proxy_request_route():