scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 86

Merge c3810ea755 into 461a58c40a

Browse source
This commit is contained in:
devin-ai-integration[bot] 2026-10-04 12:47:44 -07:00 • committed by GitHub
commit 9c82815aec
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 1409 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
litellm/integrations/otel/model/destination.py
View file

@ -10,7 +10,7 @@ from urllib.parse import quote
from pydantic import BaseModel, ConfigDict, Field
from litellm.types.utils import OtelSpanScope
from litellm.types.utils import OtelSpanScope, TeamCaptureMessageContent
class OtelDestination(BaseModel):
@ -31,6 +31,17 @@ class OtelDestination(BaseModel):
default="full",
description="``llm_only`` keeps just the model-call spans; the rest of the request tree is not forwarded.",
)
capture_message_content: TeamCaptureMessageContent | None = Field(
default=None,
description=(
"``no_content`` strips prompt and response content from this destination's copy. ``span_only`` "
"and ``None`` (omitted) forward whatever the global capture policy collected."
),
)
@property
def redacts_message_content(self) -> bool:
return self.capture_message_content == "no_content"
def header_string(self) -> str:
"""Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects.
@ -45,9 +56,9 @@ class OtelDestination(BaseModel):
def cache_key(self) -> tuple[str, tuple[tuple[str, str], ...], tuple[tuple[str, str], ...], str | None]:
"""Identity for processor reuse, so one destination means one exporter.
``span_scope`` is left out on purpose: the scope decides which spans reach the
processor, not how the processor exports them, so a full and an ``llm_only``
view of the same account share one exporter.
``span_scope`` and ``capture_message_content`` are left out on purpose: they
decide what reaches the processor, not how the processor exports it, so two
views of the same account share one exporter.
"""
return (
self.endpoint,

107
litellm/integrations/otel/plumbing/providers.py
View file

@ -39,7 +39,11 @@ from opentelemetry.util.types import Attributes, AttributeValue
from litellm._logging import verbose_logger
from litellm._version import version as litellm_version
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_TRACE_NAME
from litellm.integrations.otel.mappers.langfuse import (
LANGFUSE_OBSERVATION_INPUT,
LANGFUSE_OBSERVATION_OUTPUT,
LANGFUSE_TRACE_NAME,
)
from litellm.integrations.otel.model.config import ExporterOwner, ExporterSpec, OpenTelemetryV2Config
from litellm.integrations.otel.model.semconv import (
DB,
@ -384,6 +388,26 @@ _CAPTURED_HEADER_PREFIXES: Final = ("http.request.header.", "http.response.heade
# ``?key=`` query parameter.
_URL_KEYS: Final = frozenset({"http.url", "http.target", "url.full"})
_URL_QUERY_KEY: Final = "url.query"
# Prompt and response content, in every vocabulary the mappers write it in once the
# global capture policy collects it: GenAI semconv (MCP tool-call arguments and results
# included), Langfuse, OpenInference's blobs and per-index messages, Langtrace, Weave.
_MESSAGE_CONTENT_KEYS: Final = frozenset(
{
GenAI.INPUT_MESSAGES,
GenAI.OUTPUT_MESSAGES,
GenAI.SYSTEM_INSTRUCTIONS,
GenAI.TOOL_CALL_ARGUMENTS,
GenAI.TOOL_CALL_RESULT,
LANGFUSE_OBSERVATION_INPUT,
LANGFUSE_OBSERVATION_OUTPUT,
"input.value",
"output.value",
"llm.prompts",
"llm.completions",
"weave.output",
}
)
_INDEXED_MESSAGE_PREFIXES: Final = ("llm.input_messages.", "llm.output_messages.")
class _SpanView(ReadableSpan):
@ -414,6 +438,19 @@ class _SpanView(ReadableSpan):
)
def _is_message_content(key: str) -> bool:
return key in _MESSAGE_CONTENT_KEYS or key.startswith(_INDEXED_MESSAGE_PREFIXES)
def without_message_content(span: ReadableSpan) -> ReadableSpan:
"""``span`` without its prompt and response content, everything else left as it was."""
attributes: Final = span.attributes or _NO_ATTRIBUTES
if not any(_is_message_content(key) for key in attributes):
return span
kept: Final = MappingProxyType({key: value for key, value in attributes.items() if not _is_message_content(key)})
return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent)
def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool:
return any(key in attributes for key in _DB_SYSTEM_KEYS)
@ -510,19 +547,21 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read
exception it raised and names the operator's guardrail endpoint. Stack traces walk
the operator's install and come off every span, as do the headers the operator
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. The span
itself stays, so the tenant still gets the whole trace tree.
and the query string of the request URL, which can hold the same key. A
destination set to ``no_content`` loses the prompt and response content too. The
span itself stays, so the tenant still gets the whole trace tree.
"""
extra: Final = destination.resource_attributes
attributes: Final = span.attributes or _NO_ATTRIBUTES
database: Final = _is_database_span(attributes)
owned: Final = _is_tenant_owned_span(attributes)
unreachable: Final = _guardrail_unreachable(attributes)
redacted: Final = destination.redacts_message_content
kept: Final = MappingProxyType(
{
key: _without_query(key, value)
for key, value in attributes.items()
if _tenant_visible(key, database, owned, unreachable)
if _tenant_visible(key, database, owned, unreachable) and not (redacted and _is_message_content(key))
}
)
recorded: Final = span.events
@ -816,7 +855,8 @@ class _OverriddenBackendFilter(SpanProcessor):
``llm_only`` the model-call spans go through as trace roots and the rest of the
tree is held back, unless a destination of the request names ``sink``, the account
this exporter writes to, with a wider scope: the fan-out then delivers the rest of
the tree there and the model call keeps its place in it.
the tree there and the model call keeps its place in it. A destination naming
``sink`` with ``no_content`` strips the content from this exporter's copy.
"""
def __init__(
@ -837,7 +877,19 @@ class _OverriddenBackendFilter(SpanProcessor):
def on_end(self, span: ReadableSpan) -> None:
if self._owner in suppressed_backends() or not _in_scope(span, self._scope):
return
self._inner.on_end(_scoped(span, self._account_scope()))
scoped: Final = _scoped(span, self._account_scope())
self._inner.on_end(without_message_content(scoped) if self._account_redacts_content() else scoped)
def _account_redacts_content(self) -> bool:
"""Whether a destination of this request that names ``sink`` asked for no content.
The fan-out skips a destination this exporter already writes to, so this copy is
the one that account gets and has to honor the destination's restriction.
"""
return self._sink is not None and any(
destination.redacts_message_content and _sink_key(destination.endpoint, destination.headers) == self._sink
for destination in request_destinations()
)
def _account_scope(self) -> "OtelSpanScope":
if self._scope == "full" or self._sink is None:
@ -856,6 +908,25 @@ class _OverriddenBackendFilter(SpanProcessor):
return self._inner.force_flush(timeout_millis)
class _MessageContentFilter(SpanProcessor):
"""Strip prompt and response content from every span before ``inner`` exports it."""
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_message_content(span))
def shutdown(self) -> None:
self._inner.shutdown()
def force_flush(self, timeout_millis: int = 30000) -> bool:
return self._inner.force_flush(timeout_millis)
def build_span_exporter(config: OpenTelemetryV2Config) -> SpanExporter:
"""Build a single exporter from the top-level config fields.
@ -1107,12 +1178,26 @@ def build_resource(config: OpenTelemetryV2Config) -> Resource:
return Resource.create(attributes)
def _spec_processor(
spec: ExporterSpec, use_simple_processor: bool | None, content_redacted_owner: str | None
) -> 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),
)
redacts: Final = (
content_redacted_owner is not None and spec.owner is not None and spec.owner.value == content_redacted_owner
)
return _MessageContentFilter(exporting) if redacts else exporting
def build_tracer_provider(
config: OpenTelemetryV2Config,
exporter: SpanExporter | None = None,
baggage_processor: SpanProcessor | None = None,
use_simple_processor: bool | None = None,
tenant_overrides: bool = False,
content_redacted_owner: str | None = None,
) -> TracerProvider:
"""Build the shared :class:`TracerProvider`.
@ -1131,6 +1216,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.
``content_redacted_owner`` strips the prompt and response content from the exporter
that backend owns, again leaving every other exporter in the config alone.
"""
provider: Final = TracerProvider(resource=build_resource(config))
if baggage_processor is None:
@ -1146,17 +1233,13 @@ def build_tracer_provider(
for spec in config.exporters:
if spec.requires_headers and not spec.headers:
continue
exp = _exporter_from_spec(spec)
processor = _processor_for(
exp,
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
)
processor = _spec_processor(spec, use_simple_processor, content_redacted_owner)
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
provider.add_span_processor(
_OverriddenBackendFilter(processor, owner, scope, sink)
if owner is not None or scope != "full"
if owner is not None or scope != "full" or sink is not None
else processor
)
return provider

16
litellm/integrations/otel/plumbing/routing.py
View file

@ -69,7 +69,7 @@ _MAX_RETIRED_PROVIDERS: Final = 64
_HeaderItems: TypeAlias = tuple[tuple[str, str], ...]
_RouteKey: TypeAlias = tuple[_HeaderItems, _HeaderItems, str | None, str | None]
_RouteKey: TypeAlias = tuple[_HeaderItems, _HeaderItems, str | None, str | None, bool]
_NO_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})
@ -249,15 +249,23 @@ class TenantTracerCache:
# A fixed per-integration region endpoint (New Relic us/eu), never a
# caller-supplied host; ``None`` keeps the preset's own endpoint.
endpoint: Final = dynamic_otlp_endpoint(self._callback_name, dynamic_params)
# The tenant's own account gets the content restriction its callback vars name; the
# operator's account, reached with a project header alone, keeps the global policy.
redacted: Final = (
bool(credential_headers)
and dynamic_params is not None
and dynamic_params.get("capture_message_content") == "no_content"
)
cache_key: Final = (
tuple(sorted(credential_headers.items())),
tuple(sorted(project_headers.items())),
endpoint,
service_name,
redacted,
)
with self._lock:
provider: Final = self._cached_provider_locked(
cache_key, credential_headers, project_headers, endpoint, service_name
cache_key, credential_headers, project_headers, endpoint, service_name, redacted
)
self._open_span_counts[provider] = self._open_span_counts.get(provider, 0) + 1
evicted: Final = self._evicted_on_overflow_locked()
@ -276,13 +284,15 @@ class TenantTracerCache:
project_headers: Mapping[str, str],
endpoint: str | None,
service_name: str | None,
redacted: bool,
) -> TracerProvider:
cached: Final = self._providers.get(cache_key)
if cached is not None:
self._providers.move_to_end(cache_key)
return cached
built: Final = build_tracer_provider(
self._routed_config(credential_headers, project_headers, endpoint, service_name)
self._routed_config(credential_headers, project_headers, endpoint, service_name),
content_redacted_owner=self._callback_name if redacted else None,
)
self._providers[cache_key] = built
return built

1
litellm/integrations/otel/presets/destinations.py
View file

@ -156,4 +156,5 @@ def destination_for(
callback_name=callback_name,
protocol=protocol,
span_scope=_span_scope(callback_name, params),
capture_message_content=params.get("capture_message_content"),
)

19
litellm/litellm_core_utils/initialize_dynamic_callback_params.py
View file

@ -4,7 +4,12 @@ from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any, Final
from litellm.types.utils import OTEL_SPAN_SCOPES, TRUSTED_CALLBACK_VARS_FIELD, StandardCallbackDynamicParams
from litellm.types.utils import (
OTEL_SPAN_SCOPES,
TEAM_CAPTURE_MESSAGE_CONTENT_VALUES,
TRUSTED_CALLBACK_VARS_FIELD,
StandardCallbackDynamicParams,
)
_CLIENT_CALLBACK_METADATA_SLOTS: Final[tuple[str, ...]] = ("litellm_metadata", "metadata")
_inherited_message_logging_disabled: Final[ContextVar[bool]] = ContextVar(
@ -81,6 +86,13 @@ def validate_langfuse_span_scope_value(value: str) -> None:
raise ValueError(f"Invalid langfuse_span_scope {value!r}: must be one of {sorted(OTEL_SPAN_SCOPES)}")
def validate_capture_message_content_value(value: str) -> None:
if value not in TEAM_CAPTURE_MESSAGE_CONTENT_VALUES:
raise ValueError(
f"Invalid capture_message_content {value!r}: must be one of {sorted(TEAM_CAPTURE_MESSAGE_CONTENT_VALUES)}"
)
# Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict
_supported_callback_params: Final[tuple[str, ...]] = (
"langfuse_public_key",
@ -133,10 +145,10 @@ _request_blocked_callback_params: Final = frozenset(
}
)
# Request-blocked params that must still reach ``standard_callback_dynamic_params``
# Params that must reach ``standard_callback_dynamic_params`` only
# when the proxy itself stamped them from admin-configured team/key callback
# settings (the trusted-vars channel). The OTel per-tenant tracer routing reads
# ``standard_callback_dynamic_params``, so without this overlay a blocked param
# ``standard_callback_dynamic_params``, so without this overlay such a param
# could never drive routing at all.
_trusted_overlay_callback_params: Final = frozenset(
{
@ -144,6 +156,7 @@ _trusted_overlay_callback_params: Final = frozenset(
"newrelic_region",
"signoz_ingestion_endpoint",
"signoz_ingestion_key",
"capture_message_content",
}
)

6
litellm/proxy/_types.py
View file

@ -25,6 +25,7 @@ from typing_extensions import NotRequired, ReadOnly, Required, TypedDict
from litellm._uuid import uuid
from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_COMMANDS
from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
validate_capture_message_content_value,
validate_langfuse_environment_value,
validate_langfuse_span_scope_value,
validate_no_callback_env_reference,
@ -2375,6 +2376,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
validate_langfuse_environment_value(callback_vars[key])
if key == "langfuse_span_scope":
validate_langfuse_span_scope_value(callback_vars[key])
if key == "capture_message_content":
validate_capture_message_content_value(callback_vars[key])
return values
@ -2425,6 +2428,9 @@ class TeamCallbackMetadata(LiteLLMPydanticObjectBase):
for key in callback_vars:
if key not in valid_keys:
raise ValueError(f"Invalid callback variable: {key}. Must be one of {valid_keys}")
capture_message_content: Final = callback_vars.get("capture_message_content")
if capture_message_content is not None:
validate_capture_message_content_value(str(capture_message_content))
return values

55
litellm/proxy/common_utils/callback_config_validation.py
View file

@ -14,6 +14,8 @@ _NEWRELIC_CALLBACK: Final = "newrelic"
_NEWRELIC_VAR_PREFIX: Final = "newrelic_"
_LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel"
_LANGFUSE_SPAN_SCOPE_VAR: Final = "langfuse_span_scope"
_CAPTURE_MESSAGE_CONTENT_VAR: Final = "capture_message_content"
_SHARED_OPTION_VARS: Final = (_LANGFUSE_SPAN_SCOPE_VAR, _CAPTURE_MESSAGE_CONTENT_VAR)
_ARIZE_CALLBACK: Final = "arize"
_ARIZE_SAMPLING_RATE_VARS: Final[frozenset[str]] = frozenset(
{"arize_success_sampling_rate", "arize_error_sampling_rate"}
@ -31,6 +33,9 @@ def callback_config_error(callback_name: str | None, callback_vars: Mapping[str,
)
if langfuse_error is not None:
return langfuse_error
capture_error: Final = _capture_message_content_error(callback_name, callback_vars)
if capture_error is not None:
return capture_error
if callback_name != _NEWRELIC_CALLBACK:
return None
return _newrelic_config_error(callback_vars)
@ -75,6 +80,27 @@ def _langfuse_span_scope_error(callback_name: str | None, callback_vars: Mapping
return None
def _capture_message_content_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None:
value: Final = callback_vars.get(_CAPTURE_MESSAGE_CONTENT_VAR)
if value is None:
return None
from litellm.integrations.otel.presets.destinations import destination_capable_backends
from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
validate_capture_message_content_value,
)
supported: Final = sorted(destination_capable_backends())
if callback_name not in supported:
return (
f"{_CAPTURE_MESSAGE_CONTENT_VAR} applies to the OTel v2 callbacks {supported} only, not {callback_name!r}"
)
try:
validate_capture_message_content_value(value)
except ValueError as e:
return str(e)
return None
# Which credential family a dynamic variable belongs to. The families are the
# integrations that share one account: every langfuse_* variable configures the
# same Langfuse project whether it rides the classic callback or the OTel one,
@ -94,7 +120,7 @@ _VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType(
}
)
_FAMILY_OPTION_VARS: Final[frozenset[str]] = frozenset({_LANGFUSE_SPAN_SCOPE_VAR, *_ARIZE_SAMPLING_RATE_VARS})
_FAMILY_OPTION_VARS: Final[frozenset[str]] = frozenset({*_SHARED_OPTION_VARS, *_ARIZE_SAMPLING_RATE_VARS})
def _family_of(var: str) -> str | None:
@ -165,19 +191,30 @@ def cross_entry_family_error(
)
def conflicting_span_scope_error(
def _conflicting_option_error(var: str, incoming: str, stored_vars_by_entry: Sequence[Mapping[str, str]]) -> str | None:
return next(
(
f"{var} is already set to {stored!r} by another callback entry. "
f"Every entry shares one value: remove that entry or send the same value."
for entry in stored_vars_by_entry
if (stored := entry.get(var)) not in (None, incoming)
),
None,
)
def conflicting_shared_option_error(
callback_vars: Mapping[str, str] | None,
stored_vars_by_entry: Sequence[Mapping[str, str]],
) -> str | None:
incoming: Final = None if callback_vars is None else callback_vars.get(_LANGFUSE_SPAN_SCOPE_VAR)
if incoming is None:
if not callback_vars:
return None
incoming_by_var: Final = {var: callback_vars[var] for var in _SHARED_OPTION_VARS if var in callback_vars}
return next(
(
f"{_LANGFUSE_SPAN_SCOPE_VAR} is already set to {stored!r} by another callback entry. "
f"Every entry shares one scope: remove that entry or send the same value."
for entry in stored_vars_by_entry
if (stored := entry.get(_LANGFUSE_SPAN_SCOPE_VAR)) not in (None, incoming)
error
for var, incoming in incoming_by_var.items()
if (error := _conflicting_option_error(var, incoming, stored_vars_by_entry)) is not None
),
None,
)
@ -196,7 +233,7 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str
error
for error in (
*(_logging_entry_error(entry) for entry in entries),
*(conflicting_span_scope_error(entry_vars[i], entry_vars[:i]) for i in range(len(entry_vars))),
*(conflicting_shared_option_error(entry_vars[i], entry_vars[:i]) for i in range(len(entry_vars))),
)
if error is not None
),

5
litellm/proxy/management_endpoints/team_callback_endpoints.py
View file

@ -31,7 +31,7 @@ from litellm.proxy._types import (
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.callback_config_validation import (
callback_config_error,
conflicting_span_scope_error,
conflicting_shared_option_error,
cross_entry_family_error,
)
from litellm.proxy.common_utils.callback_utils import (
@ -283,6 +283,7 @@ async def add_team_callbacks(
- langfuse_host: The host for the Langfuse callback
- langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)
- langfuse_span_scope: For langfuse_otel, "full" (default) sends the whole request trace, "llm_only" sends only the model-call spans
- capture_message_content: For the OTel v2 callbacks (langfuse_otel, arize, weave_otel, newrelic), "no_content" strips prompt and response content from this team's exported spans, "span_only" keeps whatever the global capture policy collects. Omitted behaves like "span_only". It only narrows: a team gets content only when the proxy's global OTel v2 capture is enabled
- gcs_bucket_name: The name of the GCS bucket
- gcs_path_service_account: The path to the GCS service account
- langsmith_api_key: The API key for the Langsmith callback
@ -348,7 +349,7 @@ async def add_team_callbacks(
decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging")
stored_entries: Final = decrypted_logging if isinstance(decrypted_logging, list) else ()
stored_entry_vars: Final = [entry.get("callback_vars") or {} for entry in stored_entries]
scope_error: Final = conflicting_span_scope_error(data.callback_vars, stored_entry_vars)
scope_error: Final = conflicting_shared_option_error(data.callback_vars, stored_entry_vars)
if scope_error is not None:
raise _callback_config_error(scope_error)
# One entry has to own a credential family end to end. The entries are

4
litellm/types/utils.py
View file

@ -3646,6 +3646,8 @@ OPENAI_RESPONSE_HEADERS: Final = [
OtelSpanScope = Literal["full", "llm_only"]
OTEL_SPAN_SCOPES: Final[frozenset[str]] = frozenset(get_args(OtelSpanScope))
TeamCaptureMessageContent = Literal["no_content", "span_only"]
TEAM_CAPTURE_MESSAGE_CONTENT_VALUES: Final[frozenset[str]] = frozenset(get_args(TeamCaptureMessageContent))
class StandardCallbackDynamicParams(TypedDict, total=False):
@ -3657,6 +3659,8 @@ class StandardCallbackDynamicParams(TypedDict, total=False):
langfuse_environment: ReadOnly[str | None]
langfuse_span_scope: ReadOnly[OtelSpanScope | None]
capture_message_content: ReadOnly[TeamCaptureMessageContent | None]
# Langfuse prompt version
langfuse_prompt_version: int | None

5
tests/integration/_support/otlp_sink.py
View file

@ -186,6 +186,7 @@ class _Handler(BaseHTTPRequestHandler):
"count": len(recorded),
"host": self.headers.get("host", ""),
"headers": dict(self.headers),
"span_ids": [span["span_id"] for span in recorded],
}
)
self._send_json({"recorded": len(recorded)}, status=self.state.status)
@ -466,9 +467,7 @@ def _await_sink(url: str) -> None:
def owned_grpc_sink(directory: Path) -> Iterator[GrpcSink]:
http_port: Final = _free_port()
grpc_port: Final = _free_port()
with _spawn_sink(
directory, "otlp-grpc-sink.log", ["--port", str(http_port), "--grpc-port", str(grpc_port)]
):
with _spawn_sink(directory, "otlp-grpc-sink.log", ["--port", str(http_port), "--grpc-port", str(grpc_port)]):
control_url: Final = f"http://127.0.0.1:{http_port}"
_await_sink(control_url)
yield GrpcSink(url=f"http://127.0.0.1:{grpc_port}", control_url=control_url)

685
tests/integration/observability/test_otel_team_capture_message_content.py Normal file
View file

@ -0,0 +1,685 @@
import base64
import json
import re
import uuid
from collections.abc import Generator, Iterator, Mapping
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Final, Literal
from urllib.parse import urlparse
import httpx
import pytest
import yaml
from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment, object_value
from integration._support.otlp_sink import (
ConnectSink,
Span,
SpanSinks,
configure_sink,
owned_connect_sink,
recorded_requests,
recorded_spans,
)
from integration._support.process import OwnedProxy, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
MARKER: Final = re.compile(rb"cmc-(?:tool-|fail-)?[0-9a-f]{32}")
JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
SERVER: Final = 2
CALL_ID: Final = "litellm.call_id"
OPERATION: Final = "gen_ai.operation.name"
INPUT_TOKENS: Final = "gen_ai.usage.input_tokens"
Endpoint = Literal["chat", "responses", "messages"]
Capture = Literal["no_content", "span_only"]
def _marker(kind: str = "") -> str:
return f"cmc-{kind}{uuid.uuid4().hex}"
def _usage() -> dict[str, JsonValue]:
return {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}
def _chat_reply(marker: str, stream: bool) -> Reply:
identity: Final = f"chatcmpl-{uuid.uuid4().hex}"
if marker.startswith("cmc-tool-"):
call: Final[dict[str, JsonValue]] = {
"id": "call_1",
"type": "function",
"function": {"name": "lookup", "arguments": json.dumps({"query": marker})},
}
message: Final[dict[str, JsonValue]] = {"role": "assistant", "content": None, "tool_calls": [call]}
return Reply(
body=json.dumps(
{
"id": identity,
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [{"index": 0, "message": message, "finish_reason": "tool_calls"}],
"usage": _usage(),
}
).encode()
)
if not stream:
return Reply(
body=json.dumps(
{
"id": identity,
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{"index": 0, "message": {"role": "assistant", "content": marker}, "finish_reason": "stop"}
],
"usage": _usage(),
}
).encode()
)
chunk: Final = {"id": identity, "object": "chat.completion.chunk", "created": 1, "model": "gpt-4o-mini"}
deltas: Final[tuple[dict[str, JsonValue], ...]] = (
{**chunk, "choices": [{"index": 0, "delta": {"role": "assistant", "content": marker}}]},
{**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
{**chunk, "choices": [], "usage": _usage()},
)
return Reply(
content_type="text/event-stream",
chunks=(*(b"data: " + json.dumps(delta).encode() + b"\n\n" for delta in deltas), b"data: [DONE]\n\n"),
)
def _responses_reply(marker: str, stream: bool) -> Reply:
identity: Final = uuid.uuid4().hex
response: Final[dict[str, JsonValue]] = {
"id": f"resp_{identity}",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-4o-mini",
"output": [
{
"id": f"msg_{identity}",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": marker, "annotations": []}],
}
],
"usage": {"input_tokens": 7, "output_tokens": 2, "total_tokens": 9},
}
if not stream:
return Reply(body=json.dumps(response).encode())
events: Final[tuple[dict[str, JsonValue], ...]] = (
{
"type": "response.created",
"sequence_number": 0,
"response": {**response, "status": "in_progress", "output": []},
},
{
"type": "response.output_text.delta",
"sequence_number": 1,
"item_id": f"msg_{identity}",
"output_index": 0,
"content_index": 0,
"delta": marker,
},
{"type": "response.completed", "sequence_number": 2, "response": response},
)
return Reply(
content_type="text/event-stream",
chunks=tuple(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() for event in events),
)
def _upstream(request: Request) -> Reply:
found: Final = MARKER.search(request.body)
if found is None:
return Reply(status=404, body=b'{"error":"no marker"}')
marker: Final = found.group(0).decode()
if marker.startswith("cmc-fail-"):
return Reply(status=500, body=b'{"error":{"message":"scripted upstream failure","type":"server_error"}}')
stream: Final = object_value(JSON.validate_json(request.body)).get("stream") is True
if request.target.endswith("/responses"):
return _responses_reply(marker, stream)
return _chat_reply(marker, stream)
def _body(model: str, endpoint: Endpoint, marker: str, stream: bool) -> tuple[str, dict[str, JsonValue]]:
if endpoint == "responses":
return "/v1/responses", {"model": model, "input": marker, "stream": stream}
messages: Final[list[JsonValue]] = [{"role": "user", "content": marker}]
if endpoint == "messages":
return "/v1/messages", {"model": model, "max_tokens": 16, "messages": messages, "stream": stream}
return "/v1/chat/completions", {"model": model, "messages": messages, "stream": stream}
@dataclass(frozen=True, slots=True)
class Sent:
marker: str
trace_id: str
call_id: str
status: int
text: str
@dataclass(frozen=True, slots=True)
class Cursors:
operator: int
tenant: int
arize: int
newrelic: int
@dataclass(frozen=True, slots=True)
class Rig:
proxy: Gateway
owned: OwnedProxy
scenario: Scenario
model: str
upstream: Wire
sinks: SpanSinks
newrelic: ConnectSink
def cursors(self) -> Cursors:
self.upstream.drain()
return Cursors(
recorded_spans(self.sinks.operator)[0],
recorded_spans(self.sinks.tenant)[0],
recorded_spans(self.sinks.arize)[0],
recorded_spans(self.newrelic.control_url)[0],
)
def upstream_hits(self, marker: str) -> int:
return sum(1 for request in self.upstream.drain() if marker.encode() in request.body)
def langfuse(self, name: str, host: str | None = None) -> dict[str, str]:
return {
"langfuse_public_key": f"pk-lf-{name}",
"langfuse_secret_key": f"sk-lf-{name}",
"langfuse_host": host or self.sinks.tenant,
}
def attach(self, team: str, callback: str, variables: Mapping[str, str], kind: str | None = None) -> httpx.Response:
body: Final[dict[str, JsonValue]] = {"callback_name": callback, "callback_vars": dict(variables)}
return self.proxy.request(
"POST", f"/team/{team}/callback", body if kind is None else {**body, "callback_type": kind}
)
def team_key(
self,
capture: Capture | None,
name: str,
*,
callback: str = "langfuse_otel",
variables: Mapping[str, str] | None = None,
kind: str | None = None,
) -> tuple[str, str]:
team: Final = self.scenario.team()
base: Final = dict(variables) if variables is not None else self.langfuse(name)
attached: Final = self.attach(
team, callback, base if capture is None else {**base, "capture_message_content": capture}, kind
)
assert attached.status_code == 200, attached.text
return team, self.scenario.key(team_id=team)
def send(
self,
key: str,
endpoint: Endpoint = "chat",
stream: bool = False,
*,
marker: str | None = None,
extra: Mapping[str, JsonValue] | None = None,
) -> Sent:
tagged: Final = marker or _marker()
path, body = _body(self.model, endpoint, tagged, stream)
trace_id: Final = uuid.uuid4().hex
call_id: Final = str(uuid.uuid4())
headers: Final = {
"Authorization": f"Bearer {key}",
"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01",
"x-litellm-call-id": call_id,
}
with self.proxy.client.stream("POST", path, json={**body, **(extra or {})}, headers=headers) as response:
text: Final = response.read().decode()
return Sent(tagged, trace_id, call_id, response.status_code, text)
def _served(sent: Sent) -> Sent:
assert sent.status == 200, sent.text
assert sent.marker in sent.text, sent.text
return sent
Credential = tuple[str, str]
def _sent_with(url: str, credential: Credential) -> frozenset[str]:
"""Span ids ``url`` received in export requests whose ``credential`` header matched."""
header, value = credential
return frozenset(
span_id
for request in recorded_requests(url)
if {name.lower(): seen for name, seen in object_value(request.get("headers") or {}).items()}.get(header)
== value
for span_id in request.get("span_ids") or ()
if isinstance(span_id, str)
)
def _of_request(url: str, since: int, sent: Sent, credential: Credential | None = None) -> tuple[Span, ...]:
"""Spans of ``sent``'s trace at ``url``, narrowed to one account's export requests when ``credential`` is set."""
spans: Final = recorded_spans(url, since)[1]
traces: Final = {sent.trace_id} | {
span["trace_id"] for span in spans if span["attributes"].get(CALL_ID) == sent.call_id
}
owned: Final = None if credential is None else _sent_with(url, credential)
return tuple(span for span in spans if span["trace_id"] in traces and (owned is None or span["span_id"] in owned))
def _carries(span: Span, marker: str) -> bool:
return marker in json.dumps(span["attributes"])
def _model_spans(spans: tuple[Span, ...]) -> tuple[Span, ...]:
return tuple(span for span in spans if OPERATION in span["attributes"])
def _names(spans: tuple[Span, ...]) -> list[str]:
return sorted(span["name"] for span in spans)
def _with_content(url: str, since: int, sent: Sent) -> tuple[Span, ...]:
"""The copy of ``sent`` at ``url`` once its model span arrived and some span carries the marker."""
return eventually(
lambda: _of_request(url, since, sent),
lambda spans: bool(_model_spans(spans)) and any(_carries(span, sent.marker) for span in spans),
seconds=40,
)
def _arrived(
url: str, since: int, sent: Sent, expected: frozenset[str], credential: Credential | None = None
) -> tuple[Span, ...]:
"""The copy of ``sent`` at ``url`` once every span name in ``expected`` and a usage-bearing model span landed."""
spans: Final = eventually(
lambda: _of_request(url, since, sent, credential),
lambda found: (
expected <= {span["name"] for span in found}
and any(INPUT_TOKENS in span["attributes"] for span in _model_spans(found))
),
seconds=40,
)
assert all(
span["attributes"].get(INPUT_TOKENS) == 7 for span in _model_spans(spans) if INPUT_TOKENS in span["attributes"]
)
return spans
def _assert_redacted_twin(
rig: Rig, sent: Sent, cursors: Cursors, sink: str, since: int, credential: Credential | None = None
) -> tuple[Span, ...]:
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
carried: Final = frozenset(span["name"] for span in operator if _carries(span, sent.marker))
tenant: Final = _arrived(sink, since, sent, carried, credential)
leaked: Final = [span["name"] for span in tenant if _carries(span, sent.marker)]
assert leaked == [], f"{leaked} carried content; tenant saw {_names(tenant)}"
model: Final = _model_spans(tenant)[0]["attributes"]
assert "gen_ai.input.messages" not in model and "gen_ai.output.messages" not in model, sorted(model)
assert model.get("gen_ai.request.model"), sorted(model)
return tenant
def _assert_content_twin(rig: Rig, sent: Sent, cursors: Cursors) -> None:
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
carried: Final = frozenset(span["name"] for span in operator if _carries(span, sent.marker))
tenant: Final = _arrived(rig.sinks.tenant, cursors.tenant, sent, carried)
kept: Final = frozenset(span["name"] for span in tenant if _carries(span, sent.marker))
assert carried <= kept, f"content lost on {sorted(carried - kept)}"
@pytest.mark.parametrize(
("endpoint", "stream"),
[("chat", False), ("chat", True), ("responses", False), ("messages", True)],
ids=["chat-unary", "chat-stream", "responses-unary", "messages-stream"],
)
def test_no_content_team_keeps_its_trace_and_metadata_without_content(
rig: Rig, endpoint: Endpoint, stream: bool
) -> None:
_, key = rig.team_key("no_content", "a")
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key, endpoint, stream))
assert rig.upstream_hits(sent.marker) == 1
_assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
@pytest.mark.parametrize("capture", ["span_only", None], ids=["span_only", "omitted"])
def test_span_only_and_omitted_teams_keep_the_globally_captured_content(rig: Rig, capture: Capture | None) -> None:
_, key = rig.team_key(capture, "b")
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key))
_assert_content_twin(rig, sent, cursors)
def test_tool_call_arguments_never_reach_a_no_content_team(rig: Rig) -> None:
_, key = rig.team_key("no_content", "a")
cursors: Final = rig.cursors()
tool: Final[dict[str, JsonValue]] = {
"type": "function",
"function": {"name": "lookup", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}},
}
sent: Final = rig.send(key, marker=_marker("tool-"), extra={"tools": [tool]})
assert sent.status == 200 and sent.marker in sent.text, sent.text
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
assert any(sent.marker in str(span["attributes"].get("gen_ai.output.messages", "")) for span in operator), (
"operator copy lacks the tool-call arguments"
)
tenant: Final = _assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
assert "lookup" in json.dumps([span["attributes"] for span in tenant]), "tool definitions are metadata and stay"
def test_cache_hit_twin_of_a_no_content_team_is_redacted(rig: Rig) -> None:
_, key = rig.team_key("no_content", "a")
marker: Final = _marker()
_served(rig.send(key, marker=marker))
assert rig.upstream_hits(marker) == 1
cursors: Final = rig.cursors()
hit: Final = eventually(
lambda: _served(rig.send(key, marker=marker)), lambda _: rig.upstream_hits(marker) == 0, seconds=20
)
operator: Final = eventually(
lambda: _of_request(rig.sinks.operator, cursors.operator, hit),
lambda spans: any(span["kind"] == SERVER for span in spans),
seconds=40,
)
expected: Final = frozenset(span["name"] for span in operator if span["kind"] == SERVER)
tenant: Final = eventually(
lambda: _of_request(rig.sinks.tenant, cursors.tenant, hit),
lambda spans: expected <= {span["name"] for span in spans},
seconds=40,
)
assert [span["name"] for span in tenant if _carries(span, marker)] == [], _names(tenant)
def test_failure_only_entry_routed_with_team_credentials_is_redacted(rig: Rig) -> None:
"""A failure-only entry gets no fan-out destination, so the team's account is reached by the routed fallback."""
_, key = rig.team_key("no_content", "fail", kind="failure")
team_account: Final = ("authorization", "Basic " + base64.b64encode(b"pk-lf-fail:sk-lf-fail").decode())
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key))
routed: Final = _arrived(rig.sinks.operator, cursors.operator, sent, frozenset(), team_account)
assert [span["name"] for span in routed if _carries(span, sent.marker)] == [], _names(routed)
assert _model_spans(routed)[0]["attributes"].get("gen_ai.request.model"), _names(routed)
def test_key_level_logging_with_no_content_is_redacted(rig: Rig) -> None:
logging: Final[list[JsonValue]] = [
{
"callback_name": "langfuse_otel",
"callback_type": "success",
"callback_vars": {**rig.langfuse("key"), "capture_message_content": "no_content"},
}
]
key: Final = rig.scenario.key(metadata={"logging": logging})
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key))
_assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
def test_same_account_destination_redacts_the_shared_copy_only_for_that_team(rig: Rig) -> None:
_, restricted = rig.team_key("no_content", "operator", variables=rig.langfuse("operator", host=rig.sinks.operator))
_, other = rig.team_key("span_only", "b")
cursors: Final = rig.cursors()
shared: Final = _served(rig.send(restricted))
assert rig.upstream_hits(shared.marker) == 1
copy: Final = eventually(
lambda: _of_request(rig.sinks.operator, cursors.operator, shared),
lambda spans: any(INPUT_TOKENS in span["attributes"] for span in _model_spans(spans)),
seconds=40,
)
assert [span["name"] for span in copy if _carries(span, shared.marker)] == [], _names(copy)
assert len([span for span in copy if span["kind"] == SERVER and span["name"].startswith("POST ")]) == 1, _names(
copy
)
_assert_content_twin(rig, _served(rig.send(other)), cursors)
def test_arize_destination_drops_openinference_content(rig: Rig) -> None:
_, key = rig.team_key(
"no_content", "arize", callback="arize", variables={"arize_space_key": "space-a", "arize_api_key": "arize-a"}
)
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key))
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
assert any(sent.marker in str(span["attributes"].get("input.value", "")) for span in operator), (
"operator copy has no OpenInference content to strip"
)
tenant: Final = _assert_redacted_twin(
rig, sent, cursors, rig.sinks.arize, cursors.arize, ("arize-space-id", "space-a")
)
model: Final = _model_spans(tenant)[0]["attributes"]
assert not any(name.startswith(("llm.input_messages.", "llm.output_messages.")) for name in model), sorted(model)
assert "input.value" not in model and "output.value" not in model, sorted(model)
assert model.get("llm.model_name"), sorted(model)
def test_weave_destination_drops_its_content(rig: Rig) -> None:
_, key = rig.team_key(
"no_content",
"weave",
callback="weave_otel",
variables={"wandb_api_key": "wandb-a", "weave_project_id": "team/a"},
)
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key))
operator: Final = eventually(
lambda: _of_request(rig.sinks.arize, cursors.arize, sent, ("project_id", "operator/weave")),
lambda spans: any(sent.marker in str(span["attributes"].get("weave.output", "")) for span in spans),
seconds=40,
)
assert operator, "operator weave copy has no weave.output to strip"
tenant: Final = _assert_redacted_twin(rig, sent, cursors, rig.sinks.arize, cursors.arize, ("project_id", "team/a"))
assert not any("weave.output" in span["attributes"] for span in tenant), _names(tenant)
def test_newrelic_destination_drops_genai_content(rig: Rig) -> None:
_, key = rig.team_key("no_content", "newrelic", callback="newrelic", variables={"newrelic_api_key": "nr-team-a"})
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key))
tenant: Final = _assert_redacted_twin(
rig, sent, cursors, rig.newrelic.control_url, cursors.newrelic, ("api-key", "nr-team-a")
)
hosts: Final = {
object_value(request.get("headers") or {}).get("Host")
for request in recorded_requests(rig.newrelic.control_url)
if object_value(request.get("headers") or {}).get("api-key") == "nr-team-a"
}
assert hosts == {"otlp.nr-data.net"}, hosts
assert tenant
def test_global_no_content_is_not_lifted_by_a_span_only_team(dark_rig: Rig) -> None:
_, key = dark_rig.team_key("span_only", "b")
cursors: Final = dark_rig.cursors()
sent: Final = _served(dark_rig.send(key))
assert dark_rig.upstream_hits(sent.marker) == 1
operator: Final = _arrived(dark_rig.sinks.operator, cursors.operator, sent, frozenset())
tenant: Final = _arrived(
dark_rig.sinks.tenant, cursors.tenant, sent, frozenset(span["name"] for span in _model_spans(operator))
)
leaked: Final = [span["name"] for span in (*operator, *tenant) if _carries(span, sent.marker)]
assert leaked == [], leaked
def test_an_unsupported_value_fails_registration(rig: Rig) -> None:
team: Final = rig.scenario.team()
response: Final = rig.attach(team, "langfuse_otel", {**rig.langfuse("a"), "capture_message_content": "bogus"})
assert response.status_code == 422, response.text
assert "Invalid capture_message_content 'bogus'" in response.text, response.text
def test_classic_langfuse_rejects_the_setting(rig: Rig) -> None:
team: Final = rig.scenario.team()
response: Final = rig.attach(team, "langfuse", {**rig.langfuse("a"), "capture_message_content": "no_content"})
assert response.status_code == 400, response.text
assert "capture_message_content" in response.text, response.text
def test_a_conflicting_second_entry_is_rejected(rig: Rig) -> None:
team, _ = rig.team_key("no_content", "a")
response: Final = rig.attach(
team,
"arize",
{"arize_space_key": "space-a", "arize_api_key": "arize-a", "capture_message_content": "span_only"},
)
assert response.status_code == 400, response.text
assert "already set to 'no_content'" in response.text, response.text
def test_replacing_the_registration_switches_the_team_to_content(rig: Rig) -> None:
team, key = rig.team_key("no_content", "a")
cursors: Final = rig.cursors()
_assert_redacted_twin(rig, _served(rig.send(key)), cursors, rig.sinks.tenant, cursors.tenant)
removed: Final = rig.proxy.request("DELETE", f"/team/{team}/callback/langfuse_otel")
assert removed.status_code == 200, removed.text
replaced: Final = rig.attach(team, "langfuse_otel", {**rig.langfuse("a"), "capture_message_content": "span_only"})
assert replaced.status_code == 200, replaced.text
def attempt() -> bool:
since: Final = rig.cursors()
sent: Final = _served(rig.send(key))
operator: Final = _with_content(rig.sinks.operator, since.operator, sent)
carried: Final = frozenset(span["name"] for span in operator if _carries(span, sent.marker))
tenant: Final = _arrived(rig.sinks.tenant, since.tenant, sent, carried)
return any(_carries(span, sent.marker) for span in tenant)
assert eventually(attempt, bool, seconds=60)
def test_request_metadata_cannot_lift_the_team_restriction(rig: Rig) -> None:
_, key = rig.team_key("no_content", "a")
cursors: Final = rig.cursors()
sent: Final = _served(rig.send(key, extra={"metadata": {"capture_message_content": "span_only"}}))
assert rig.upstream_hits(sent.marker) == 1
_assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
def test_stalled_tenant_during_a_mixed_burst_keeps_serving_and_each_teams_policy(rig: Rig) -> None:
_, restricted = rig.team_key("no_content", "a")
_, open_key = rig.team_key("span_only", "b")
shapes: Final[tuple[tuple[Endpoint, bool], ...]] = (
("chat", False),
("chat", True),
("responses", False),
("messages", True),
)
plan: Final = [(restricted if index % 2 else open_key, *shapes[index % len(shapes)]) for index in range(30)]
cursors: Final = rig.cursors()
configure_sink(rig.sinks.tenant, paused=True)
try:
with ThreadPoolExecutor(max_workers=10) as pool:
sent: Final = list(pool.map(lambda job: rig.send(job[0], job[1], job[2]), plan))
finally:
configure_sink(rig.sinks.tenant, paused=False)
assert [result.status for result in sent] == [200] * 30, [result.text for result in sent if result.status != 200]
for (key, _, _), result in zip(plan, sent, strict=True):
roots = [
span
for span in _with_content(rig.sinks.operator, cursors.operator, result)
if span["kind"] == SERVER and span["name"].startswith("POST ")
]
assert len(roots) == 1, roots
if key == restricted:
_assert_redacted_twin(rig, result, cursors, rig.sinks.tenant, cursors.tenant)
else:
_assert_content_twin(rig, result, cursors)
def _write_config(directory: Path, sinks: SpanSinks, name: str) -> Path:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["litellm_settings"] = {
**config.get("litellm_settings", {}),
"callbacks": ["langfuse_otel"],
"otel_tenant_destination_mode": "additive",
"provider_url_destination_allowed_hosts": [urlparse(sinks.tenant).netloc, urlparse(sinks.operator).netloc],
}
config["general_settings"] = {**config.get("general_settings", {}), "user_api_key_cache_ttl": 2}
path: Final = directory / f"{name}.yaml"
path.write_text(yaml.safe_dump(config))
return path
@contextmanager
def _started(
provider: Wire, sinks: SpanSinks, newrelic: ConnectSink, directory: Path, capture: str | None
) -> Generator[Rig]:
environment: Final = {
"LITELLM_OTEL_V2": "1",
"OTEL_BSP_SCHEDULE_DELAY": "300",
"MAPPER_NAMES": "genai,openinference",
"LANGFUSE_HOST": sinks.operator,
"LANGFUSE_PUBLIC_KEY": "pk-lf-operator",
"LANGFUSE_SECRET_KEY": "sk-lf-operator",
"ARIZE_SPACE_KEY": "space-operator",
"ARIZE_API_KEY": "arize-operator",
"ARIZE_HTTP_ENDPOINT": sinks.arize + "/v1/traces",
"WANDB_API_KEY": "wandb-operator",
"WANDB_PROJECT_ID": "operator/weave",
"WANDB_HOST": sinks.arize,
"HTTPS_PROXY": newrelic.proxy_url,
"NO_PROXY": "127.0.0.1,localhost",
"REQUESTS_CA_BUNDLE": newrelic.ca_pem,
**({} if capture is None else {"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": capture}),
}
with (
gateway_from_environment() as gateway,
owned_proxy_process(
gateway,
directory,
environment,
config=_write_config(directory, sinks, "capture"),
remove_environment=(
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
"LITELLM_OTEL_TENANT_DESTINATION_MODE",
),
workers=2,
) as owned,
owned.gateway.scenario() as scenario,
):
yield Rig(
owned.gateway, owned, scenario, scenario.model(api_base=provider.url + "/v1"), provider, sinks, newrelic
)
@pytest.fixture(scope="module")
def provider() -> Iterator[Wire]:
with wire_server(_upstream) as wire:
yield wire
@pytest.fixture(scope="module")
def newrelic_sink(tmp_path_factory: pytest.TempPathFactory) -> Iterator[ConnectSink]:
with owned_connect_sink(tmp_path_factory.mktemp("newrelic-sink")) as sink:
yield sink
@pytest.fixture(scope="module")
def rig(
provider: Wire, audit_sinks: SpanSinks, newrelic_sink: ConnectSink, tmp_path_factory: pytest.TempPathFactory
) -> Iterator[Rig]:
directory: Final = tmp_path_factory.mktemp("capture-span-only")
with _started(provider, audit_sinks, newrelic_sink, directory, "span_only") as started:
yield started
@pytest.fixture(scope="module")
def dark_rig(
provider: Wire, audit_sinks: SpanSinks, newrelic_sink: ConnectSink, tmp_path_factory: pytest.TempPathFactory
) -> Iterator[Rig]:
with _started(provider, audit_sinks, newrelic_sink, tmp_path_factory.mktemp("capture-no-content"), None) as started:
yield started

400
tests/unit/integrations/otel/test_otel_v2_destinations.py
View file

@ -5,19 +5,21 @@ import contextvars
import time
from base64 import b64encode
from collections.abc import Mapping
from dataclasses import replace
from datetime import datetime, timezone
from functools import reduce
from types import MappingProxyType
import pytest
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
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
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.otel import GenAIOperation
from litellm.integrations.otel import logger as otel_logger
from litellm.integrations.otel.logger import (
OpenTelemetryV2,
@ -25,6 +27,7 @@ from litellm.integrations.otel.logger import (
fan_out_provider,
publish_global_otel_v2_provider,
)
from litellm.integrations.otel.mappers import resolve_mappers
from litellm.integrations.otel.model.config import (
ExporterOwner,
ExporterSpec,
@ -32,6 +35,15 @@ from litellm.integrations.otel.model.config import (
is_otel_v2_enabled,
)
from litellm.integrations.otel.model.destination import OtelDestination
from litellm.integrations.otel.model.payloads import (
LLMCallSpanData,
LLMRequestParams,
LLMUsage,
MCPToolCallSpanData,
RequestIdentity,
ServerInfo,
ToolDefinition,
)
from litellm.integrations.otel.plumbing import providers as otel_providers
from litellm.integrations.otel.plumbing.context import (
destination_backends,
@ -45,6 +57,7 @@ from litellm.integrations.otel.plumbing.providers import (
build_tracer_provider,
deliverable_destinations,
operator_sink_scopes,
register_exporter_factory,
)
from litellm.integrations.otel.plumbing.routing import TenantTracerCache, get_tracer
from litellm.integrations.otel.presets.arize import arize_preset
@ -53,7 +66,7 @@ from litellm.integrations.otel.presets.destinations import (
destination_for,
)
from litellm.integrations.otel.presets.langfuse import langfuse_preset
from litellm.proxy._types import AddTeamCallback, UserAPIKeyAuth
from litellm.proxy._types import AddTeamCallback, TeamCallbackMetadata, UserAPIKeyAuth
from litellm.proxy.litellm_pre_call_utils import (
convert_key_logging_metadata_to_callback,
resolve_tenant_otel_destinations,
@ -3460,3 +3473,386 @@ class TestTenantHostSsrfGuard:
destination_for("langfuse_otel", self._langfuse("http://10.0.0.5:3000"))
assert sum("provider_url_destination_allowed_hosts" in record.message for record in caplog.records) == 1
_ALL_MAPPERS = ("genai", "legacy", "openinference", "langfuse", "weave", "langtrace")
_SECRET = "SECRET-MARKER"
_MODEL_CALL_WITH_CONTENT = LLMCallSpanData(
operation=GenAIOperation.CHAT,
provider="openai",
request_model="gpt-4o",
response_model="gpt-4o-2024",
response_id="resp_1",
request_params=LLMRequestParams(temperature=0.5, max_tokens=128),
usage=LLMUsage(input_tokens=12, output_tokens=8, total_tokens=20),
finish_reasons=("tool_calls",),
error=None,
response_cost=0.001,
server=ServerInfo("api.openai.com", 443),
identity=RequestIdentity(call_id="c1", team_id="t1"),
tools=(ToolDefinition(name="lookup", description="Find a city", parameters_json='{"type":"object"}'),),
messages_in=(
{"role": "system", "content": f"system {_SECRET}"},
{"role": "user", "content": f"prompt {_SECRET}"},
),
choices_out=(
{
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"content": f"answer {_SECRET}",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "lookup", "arguments": f'{{"city": "{_SECRET}"}}'},
}
],
},
},
),
)
_MODEL_CALL_WITHOUT_CONTENT = replace(_MODEL_CALL_WITH_CONTENT, messages_in=(), choices_out=())
_TOOL_CALL_WITH_CONTENT = MCPToolCallSpanData(
operation=GenAIOperation.EXECUTE_TOOL,
method="tools/call",
tool_name="lookup",
server_name="maps",
server_address="maps.local",
server_port=443,
session_id="s1",
arguments_json=f'{{"city": "{_SECRET}"}}',
result_json=f'{{"weather": "{_SECRET}"}}',
error=None,
response_cost=None,
identity=RequestIdentity(call_id="c2", team_id="t1"),
)
_TOOL_CALL_WITHOUT_CONTENT = replace(_TOOL_CALL_WITH_CONTENT, arguments_json=None, result_json=None)
NO_CONTENT_DEST = OtelDestination(
endpoint="http://team-a.local/api/public/otel",
headers=MappingProxyType({"Authorization": "Basic YQ=="}),
callback_name="langfuse_otel",
capture_message_content="no_content",
)
INHERITING_DEST = OtelDestination(
endpoint="http://team-b.local/v1",
headers=MappingProxyType({"api_key": "k", "arize-space-id": "s"}),
callback_name="arize",
)
def mapped(data: LLMCallSpanData | MCPToolCallSpanData) -> Mapping[str, object]:
"""What the configured mappers write on the span, the way the logger stamps it."""
return MappingProxyType(reduce(lambda acc, mapper: {**acc, **mapper.map(data)}, resolve_mappers(_ALL_MAPPERS), {}))
def recorded(attributes: Mapping[str, object]) -> dict[str, object]:
"""``attributes`` the way the SDK records them, sequences frozen into tuples."""
return {key: tuple(value) if isinstance(value, list) else value for key, value in attributes.items()}
def model_call_tree(provider: TracerProvider, attributes: Mapping[str, object]) -> None:
tracer = get_tracer(provider, "litellm")
with tracer.start_as_current_span("POST /v1/chat/completions"):
with tracer.start_as_current_span("chat gpt-4o") as llm:
llm.set_attributes(attributes)
llm.add_event("gen_ai.content.first_chunk", {"gen_ai.response.model": "gpt-4o-2024"})
def by_name(exporter: InMemorySpanExporter) -> dict[str, ReadableSpan]:
return {span.name: span for span in exporter.get_finished_spans()}
def carries_content(span: ReadableSpan) -> bool:
return any(_SECRET in str(value) for value in span.attributes.values())
class TestCaptureMessageContent:
"""A team narrows the globally captured content for its own destination only."""
@staticmethod
def _fan_out(operator: InMemorySpanExporter, exporters: Mapping[str, InMemorySpanExporter]) -> TracerProvider:
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(operator))
provider.add_span_processor(
TenantFanOutSpanProcessor(processor_factory=lambda d: SimpleSpanProcessor(exporters[d.endpoint]))
)
return provider
@staticmethod
def _run(
provider: TracerProvider, destinations: tuple[OtelDestination, ...], attributes: Mapping[str, object]
) -> None:
def run() -> None:
set_request_destinations(destinations)
model_call_tree(provider, attributes)
in_fresh_context(run)
@pytest.mark.parametrize(
("globally_captured", "setting", "content_exported"),
[
(True, None, True),
(True, "span_only", True),
(True, "no_content", False),
(False, None, False),
(False, "span_only", False),
(False, "no_content", False),
],
)
def test_the_team_setting_only_ever_narrows_the_global_capture(self, globally_captured, setting, content_exported):
attributes = mapped(_MODEL_CALL_WITH_CONTENT if globally_captured else _MODEL_CALL_WITHOUT_CONTENT)
destination = OtelDestination(
endpoint="http://team.local/api/public/otel",
headers=MappingProxyType({"Authorization": "Basic dA=="}),
callback_name="langfuse_otel",
capture_message_content=setting,
)
operator, tenant = InMemorySpanExporter(), InMemorySpanExporter()
self._run(self._fan_out(operator, {destination.endpoint: tenant}), (destination,), attributes)
exported = by_name(tenant)["chat gpt-4o"]
assert carries_content(exported) is content_exported
assert carries_content(by_name(operator)["chat gpt-4o"]) is globally_captured, (
"the operator's copy is untouched"
)
assert exported.attributes["gen_ai.request.model"] == "gpt-4o"
assert exported.attributes["gen_ai.usage.input_tokens"] == 12
assert exported.attributes["gen_ai.usage.output_tokens"] == 8
@pytest.mark.parametrize(
("with_content", "without_content"),
[
(_MODEL_CALL_WITH_CONTENT, _MODEL_CALL_WITHOUT_CONTENT),
(_TOOL_CALL_WITH_CONTENT, _TOOL_CALL_WITHOUT_CONTENT),
],
ids=["model_call", "mcp_tool_call"],
)
def test_no_content_removes_exactly_what_global_capture_adds_in_every_mapper_vocabulary(
self, with_content, without_content
):
captured, uncaptured = mapped(with_content), mapped(without_content)
assert frozenset(captured) - frozenset(uncaptured), "the fixture must exercise captured content"
operator, tenant = InMemorySpanExporter(), InMemorySpanExporter()
self._run(self._fan_out(operator, {NO_CONTENT_DEST.endpoint: tenant}), (NO_CONTENT_DEST,), captured)
exported = by_name(tenant)["chat gpt-4o"]
assert dict(exported.attributes) == recorded(uncaptured), "only content goes, every other attribute stays"
assert not carries_content(exported)
def test_restricting_one_destination_leaves_the_original_span_and_the_other_destination_alone(self):
attributes = mapped(_MODEL_CALL_WITH_CONTENT)
operator, restricted, inheriting = InMemorySpanExporter(), InMemorySpanExporter(), InMemorySpanExporter()
exporters = {NO_CONTENT_DEST.endpoint: restricted, INHERITING_DEST.endpoint: inheriting}
self._run(self._fan_out(operator, exporters), (NO_CONTENT_DEST, INHERITING_DEST), attributes)
original = by_name(operator)
kept = by_name(inheriting)
stripped = by_name(restricted)
assert dict(original["chat gpt-4o"].attributes) == recorded(attributes)
assert dict(kept["chat gpt-4o"].attributes) == recorded(attributes)
assert not carries_content(stripped["chat gpt-4o"])
for copy in (kept, stripped):
for name, span in copy.items():
assert span.context == original[name].context, "same trace id and span id"
assert span.parent == original[name].parent, "same place in the tree"
assert [e.name for e in copy["chat gpt-4o"].events] == [e.name for e in original["chat gpt-4o"].events]
def test_an_omitted_setting_and_span_only_export_the_same_span(self):
attributes = mapped(_MODEL_CALL_WITH_CONTENT)
span_only = INHERITING_DEST.model_copy(update={"capture_message_content": "span_only"})
omitted_exporter, span_only_exporter = InMemorySpanExporter(), InMemorySpanExporter()
for destination, exporter in ((INHERITING_DEST, omitted_exporter), (span_only, span_only_exporter)):
self._run(
self._fan_out(InMemorySpanExporter(), {destination.endpoint: exporter}), (destination,), attributes
)
assert {n: dict(s.attributes) for n, s in by_name(omitted_exporter).items()} == {
n: dict(s.attributes) for n, s in by_name(span_only_exporter).items()
}
@pytest.mark.parametrize(("setting", "content_exported"), [("no_content", False), (None, True)])
def test_the_operators_exporter_on_the_same_account_honors_the_teams_setting_once(
self, monkeypatch, setting, content_exported
):
"""The fan-out skips a destination the operator already writes to, so the operator's
copy is the one that account receives and must not bypass the team's restriction."""
monkeypatch.setattr(litellm, "otel_tenant_destination_mode", "additive", raising=False)
shared = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(
_OverriddenBackendFilter(
SimpleSpanProcessor(shared), "langfuse_otel", "full", TestRoutingMode.OPERATOR_SINK
)
)
provider.add_span_processor(
TenantFanOutSpanProcessor(
processor_factory=lambda _d: SimpleSpanProcessor(shared),
operator_sinks=MappingProxyType({TestRoutingMode.OPERATOR_SINK: "full"}),
)
)
destination = OtelDestination(
endpoint=TestRoutingMode.SAME_ACCOUNT_ENDPOINT,
headers=MappingProxyType({"Authorization": "Basic op"}),
callback_name="langfuse_otel",
capture_message_content=setting,
)
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
model_calls = [s for s in shared.get_finished_spans() if s.name == "chat gpt-4o"]
assert len(model_calls) == 1, "the same account received the span twice"
assert carries_content(model_calls[0]) is content_exported
@pytest.mark.parametrize("mode", ["override", "additive"])
def test_an_operator_collector_on_the_teams_account_does_not_bypass_no_content(self, monkeypatch, mode):
monkeypatch.setattr(litellm, "otel_tenant_destination_mode", mode, raising=False)
shared = InMemorySpanExporter()
kind = f"lit8244_collector_{mode}"
register_exporter_factory(kind, lambda _spec: shared)
config = OpenTelemetryV2Config(
exporters=[
ExporterSpec(
kind=kind,
endpoint=TestRoutingMode.OPERATOR_SINK[0],
headers="authorization=Basic op",
)
]
)
provider = build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True)
provider.add_span_processor(
TenantFanOutSpanProcessor(
processor_factory=lambda _d: SimpleSpanProcessor(shared),
operator_sinks=operator_sink_scopes(config),
)
)
destination = OtelDestination(
endpoint=TestRoutingMode.SAME_ACCOUNT_ENDPOINT,
headers=MappingProxyType({"Authorization": "Basic op"}),
callback_name="langfuse_otel",
capture_message_content="no_content",
)
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
model_calls = [s for s in shared.get_finished_spans() if s.name == "chat gpt-4o"]
assert model_calls, "the account must receive the model call before its content can be judged absent"
assert not any(carries_content(span) for span in model_calls)
@pytest.mark.parametrize(
("setting", "content_exported"), [("no_content", False), ("span_only", True), (None, True)]
)
def test_a_request_routed_to_the_teams_credentials_honors_the_setting(self, setting, content_exported):
"""A failure-only entry, or a destination the fan-out could not deliver, leaves the
model call on the per-request tracer route with the team's credentials."""
exporters: dict[ExporterOwner | None, InMemorySpanExporter] = {}
kind = f"lit8244_routed_{setting}"
register_exporter_factory(kind, lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()))
config = OpenTelemetryV2Config(
exporters=[
ExporterSpec(
kind=kind,
endpoint="http://op.local",
owner=ExporterOwner.LANGFUSE_OTEL,
use_simple_processor=True,
),
ExporterSpec(kind=kind),
]
)
cache = TenantTracerCache(config, "langfuse_otel", "litellm")
params = {"langfuse_public_key": "pk-team", "langfuse_secret_key": "sk-team"}
if setting is not None:
params["capture_message_content"] = setting
route = cache.route_for(get_tracer(TracerProvider(), "litellm"), params)
with route.tracer.start_as_current_span("chat gpt-4o") as llm:
llm.set_attributes(mapped(_MODEL_CALL_WITH_CONTENT))
cache.release(route.provider)
assert route.detached is True
team_copy = by_name(exporters[ExporterOwner.LANGFUSE_OTEL])["chat gpt-4o"]
assert carries_content(team_copy) is content_exported
assert team_copy.attributes["gen_ai.usage.input_tokens"] == 12
assert carries_content(by_name(exporters[None])["chat gpt-4o"]), "the operator's collector keeps the content"
def test_a_restricted_and_an_unrestricted_team_never_share_a_routed_provider(self):
cache = TenantTracerCache(
OpenTelemetryV2Config(
exporters=[
ExporterSpec(kind="otlp_http", endpoint="http://op.local", owner=ExporterOwner.LANGFUSE_OTEL)
]
),
"langfuse_otel",
"litellm",
)
default = get_tracer(TracerProvider(), "litellm")
creds = {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}
restricted = cache.route_for(default, {**creds, "capture_message_content": "no_content"})
unrestricted = cache.route_for(default, creds)
assert restricted.provider is not unrestricted.provider
cache.release(restricted.provider)
cache.release(unrestricted.provider)
@pytest.mark.usefixtures("allow_test_hosts")
@pytest.mark.parametrize("setting", ["no_content", "span_only", None])
def test_the_setting_rides_the_teams_destination_and_omission_stays_omitted(self, monkeypatch, setting):
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
is_otel_v2_enabled.cache_clear()
callback_vars = {
"langfuse_public_key": "pk-team",
"langfuse_secret_key": "sk-team",
"langfuse_host": "http://team.local",
}
if setting is not None:
callback_vars["capture_message_content"] = setting
auth = UserAPIKeyAuth(
team_metadata={
"logging": [
{"callback_name": "langfuse_otel", "callback_type": "success", "callback_vars": callback_vars},
{
"callback_name": "arize",
"callback_type": "success",
"callback_vars": {"arize_api_key": "k", "arize_space_id": "s"},
},
]
}
)
destinations = {d.callback_name: d for d in resolve_tenant_otel_destinations(auth)}
assert destinations["langfuse_otel"].capture_message_content == setting
assert destinations["arize"].capture_message_content is None, "the setting stays on its own callback"
@pytest.mark.parametrize("value", ["full", "NO_CONTENT", "span_and_event", "", "true"])
def test_an_unsupported_value_fails_registration(self, value):
with pytest.raises(ValueError, match=r"Invalid capture_message_content .*\['no_content', 'span_only'\]"):
AddTeamCallback(
callback_name="langfuse_otel",
callback_type="success",
callback_vars={
"langfuse_public_key": "pk",
"langfuse_secret_key": "sk",
"capture_message_content": value,
},
)
@pytest.mark.parametrize("value", ["no_content", "span_only"])
def test_a_supported_value_is_stored_as_given(self, value):
saved = AddTeamCallback(
callback_name="langfuse_otel",
callback_type="success",
callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "capture_message_content": value},
)
assert saved.callback_vars["capture_message_content"] == value
def test_the_legacy_team_callback_metadata_rejects_an_unsupported_value(self):
with pytest.raises(ValueError, match="Invalid capture_message_content"):
TeamCallbackMetadata(success_callback=["langfuse_otel"], callback_vars={"capture_message_content": "all"})

14
tests/unit/litellm_core_utils/test_initialize_dynamic_callback_params.py
View file

@ -279,3 +279,17 @@ def test_arize_sampling_rates_are_picked_up_from_metadata():
assert params.get("arize_success_sampling_rate") == "0.5"
assert params.get("arize_error_sampling_rate") == "0.1"
def test_capture_message_content_comes_only_from_the_trusted_team_vars():
from litellm.types.utils import TRUSTED_CALLBACK_VARS_FIELD
caller_only = {
"capture_message_content": "span_only",
"metadata": {"capture_message_content": "span_only"},
"litellm_params": {"metadata": {"capture_message_content": "span_only"}},
}
assert initialize_standard_callback_dynamic_params(caller_only).get("capture_message_content") is None
stamped = {**caller_only, TRUSTED_CALLBACK_VARS_FIELD: {"capture_message_content": "no_content"}}
assert initialize_standard_callback_dynamic_params(stamped).get("capture_message_content") == "no_content"

12
tests/unit/proxy/auth/test_auth_utils.py
View file

@ -3533,6 +3533,18 @@ def test_model_level_allow_does_not_skip_subsequent_banned_params(monkeypatch):
assert "langfuse_host" in str(exc.value)
@pytest.mark.parametrize(
"request_body",
[
{"model": "gpt-4", "capture_message_content": "span_only"},
{"model": "gpt-4", "metadata": {"capture_message_content": "span_only"}},
],
ids=["top-level", "metadata"],
)
def test_client_capture_message_content_is_not_rejected_at_the_request_body_boundary(request_body):
assert is_request_body_safe(request_body=request_body, general_settings={}, llm_router=None, model="gpt-4")
def test_observability_ban_covers_canonical_supported_callback_params():
"""Guard test: every entry in the canonical
``_supported_callback_params`` allow-list must end up either banned by

54
tests/unit/proxy/common_utils/test_callback_config_validation.py
View file

@ -2,7 +2,7 @@ import pytest
from litellm.proxy.common_utils.callback_config_validation import (
callback_config_error,
conflicting_span_scope_error,
conflicting_shared_option_error,
cross_entry_family_error,
logging_metadata_config_error,
)
@ -59,7 +59,7 @@ def test_a_bad_span_scope_is_reported_even_when_the_environment_is_fine():
def test_one_span_scope_per_team(new_vars, stored, rejected):
"""The entries flatten last-wins, so a second scope would export whichever entry
was stored last. An entry that names no scope leaves the stored one in charge."""
error = conflicting_span_scope_error(new_vars, stored)
error = conflicting_shared_option_error(new_vars, stored)
assert (error is not None) is rejected
if rejected:
assert "langfuse_span_scope" in error and stored[-1]["langfuse_span_scope"] in error
@ -128,3 +128,53 @@ def test_arize_sampling_rates_are_not_family_credentials():
stored = [{"arize_api_key": "k1", "arize_success_sampling_rate": "0.5"}]
assert cross_entry_family_error({"arize_success_sampling_rate": "0.1"}, stored) is None
assert cross_entry_family_error({"arize_error_sampling_rate": "0.5"}, stored) is None
@pytest.mark.parametrize("callback_name", ["langfuse_otel", "arize", "weave_otel", "newrelic"])
@pytest.mark.parametrize("value", ["no_content", "span_only"])
def test_capture_message_content_is_accepted_on_every_otel_v2_destination(callback_name, value):
assert callback_config_error(callback_name, {"capture_message_content": value}) is None
@pytest.mark.parametrize("callback_name", ["langfuse", "datadog", "otel", "arize_phoenix", None])
def test_capture_message_content_is_rejected_where_it_would_never_take_effect(callback_name):
error = callback_config_error(callback_name, {"capture_message_content": "no_content"})
assert error is not None and "capture_message_content" in error and "langfuse_otel" in error
def test_an_unsupported_capture_message_content_is_rejected_on_key_logging_metadata():
error = logging_metadata_config_error(
{
"logging": [
{
"callback_name": "langfuse_otel",
"callback_type": "success",
"callback_vars": {"capture_message_content": "full"},
}
]
}
)
assert error is not None and "Invalid capture_message_content" in error
@pytest.mark.parametrize(
"new_vars, stored, rejected",
[
({"capture_message_content": "no_content"}, [{"capture_message_content": "span_only"}], True),
(
{"capture_message_content": "span_only"},
[{"langfuse_public_key": "pk"}, {"capture_message_content": "no_content"}],
True,
),
({"capture_message_content": "no_content"}, [{"capture_message_content": "no_content"}], False),
({"capture_message_content": "no_content"}, [{"langfuse_span_scope": "llm_only"}], False),
({"langfuse_public_key": "pk"}, [{"capture_message_content": "no_content"}], False),
],
)
def test_one_capture_message_content_per_team(new_vars, stored, rejected):
"""The entries flatten last-wins on the routed path, so a second value would apply
whichever entry was stored last to the team's traffic."""
error = conflicting_shared_option_error(new_vars, stored)
assert (error is not None) is rejected
if rejected:
assert "capture_message_content" in error

45
tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py
View file

@ -1712,3 +1712,48 @@ def test_add_team_callback_accepts_arize_sampling_rate_vars():
)
assert data.callback_vars["arize_success_sampling_rate"] == "0.5"
assert data.callback_vars["arize_error_sampling_rate"] == "1.0"
@pytest.mark.asyncio
async def test_add_team_callbacks_rejects_capture_message_content_on_a_non_otel_v2_callback(patched_prisma):
data = AddTeamCallback(
callback_name="langfuse",
callback_type="success",
callback_vars={
"langfuse_public_key": "pk",
"langfuse_secret_key": "sk",
"capture_message_content": "no_content",
},
)
with pytest.raises(HTTPException) as exc:
await add_team_callbacks(
data=data,
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=_admin_auth(),
)
assert exc.value.status_code == 400
assert "capture_message_content" in str(exc.value.detail)
patched_prisma.db.litellm_teamtable.update.assert_not_called()
@pytest.mark.asyncio
async def test_add_team_callbacks_stores_capture_message_content_in_the_existing_callback_metadata(patched_prisma):
data = AddTeamCallback(
callback_name="langfuse_otel",
callback_type="success",
callback_vars={
"langfuse_public_key": "pk",
"langfuse_secret_key": "sk",
"capture_message_content": "no_content",
},
)
await add_team_callbacks(
data=data,
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=_admin_auth(),
)
patched_prisma.db.litellm_teamtable.update.assert_awaited_once()
stored = json.loads(patched_prisma.db.litellm_teamtable.update.await_args.kwargs["data"]["metadata"])
assert [entry["callback_vars"].get("capture_message_content") for entry in stored["logging"]] == ["no_content"]

1
tests/unit/proxy/test_credential_slot_registry.py
View file

@ -75,6 +75,7 @@ CALLBACK_PARAM_CLASSIFICATION: Final[Mapping[str, Classification]] = MappingProx
"langfuse_host": NotSecret("sink endpoint URL"),
"langfuse_environment": NotSecret("environment label"),
"langfuse_span_scope": NotSecret("span scope setting"),
"capture_message_content": NotSecret("content capture setting"),
"langfuse_prompt_version": NotSecret("prompt version number"),
"gcs_bucket_name": NotSecret("bucket name"),
"gcs_path_service_account": Unplanted(),

1
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -17621,6 +17621,7 @@ export interface paths {
* - langfuse_host: The host for the Langfuse callback
* - langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)
* - langfuse_span_scope: For langfuse_otel, "full" (default) sends the whole request trace, "llm_only" sends only the model-call spans
* - capture_message_content: For the OTel v2 callbacks (langfuse_otel, arize, weave_otel, newrelic), "no_content" strips prompt and response content from this team's exported spans, "span_only" keeps whatever the global capture policy collects. Omitted behaves like "span_only". It only narrows: a team gets content only when the proxy's global OTel v2 capture is enabled
* - gcs_bucket_name: The name of the GCS bucket
* - gcs_path_service_account: The path to the GCS service account
* - langsmith_api_key: The API key for the Langsmith callback