diff --git a/litellm/integrations/otel/model/destination.py b/litellm/integrations/otel/model/destination.py index c9c035f24a1..5887059e4b3 100644 --- a/litellm/integrations/otel/model/destination.py +++ b/litellm/integrations/otel/model/destination.py @@ -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, diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 8b01750b8f2..90b133d945f 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -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 diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index d7b1cadfc92..cbd1be92adb 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -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 diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index e6cb775af1d..275b178050b 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -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"), ) diff --git a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py index 3100ca6fba1..418861f3817 100644 --- a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py +++ b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py @@ -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", } ) diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index b993a5d8d6d..0671651ea82 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -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 diff --git a/litellm/proxy/common_utils/callback_config_validation.py b/litellm/proxy/common_utils/callback_config_validation.py index 6a0fcbe0bb3..cedb64d45f7 100644 --- a/litellm/proxy/common_utils/callback_config_validation.py +++ b/litellm/proxy/common_utils/callback_config_validation.py @@ -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 ), diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py index ac6169d25bd..7567a16136d 100644 --- a/litellm/proxy/management_endpoints/team_callback_endpoints.py +++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py @@ -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 diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 142f9b14a72..5f79a767025 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -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 diff --git a/tests/integration/_support/otlp_sink.py b/tests/integration/_support/otlp_sink.py index eeabe9d886f..a4d2ea30959 100644 --- a/tests/integration/_support/otlp_sink.py +++ b/tests/integration/_support/otlp_sink.py @@ -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) diff --git a/tests/integration/observability/test_otel_team_capture_message_content.py b/tests/integration/observability/test_otel_team_capture_message_content.py new file mode 100644 index 00000000000..5a0340dbf11 --- /dev/null +++ b/tests/integration/observability/test_otel_team_capture_message_content.py @@ -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 diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 3fdc47f130c..d273e0c7897 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -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"}) diff --git a/tests/unit/litellm_core_utils/test_initialize_dynamic_callback_params.py b/tests/unit/litellm_core_utils/test_initialize_dynamic_callback_params.py index a9a6509f6dd..94970fd89d8 100644 --- a/tests/unit/litellm_core_utils/test_initialize_dynamic_callback_params.py +++ b/tests/unit/litellm_core_utils/test_initialize_dynamic_callback_params.py @@ -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" diff --git a/tests/unit/proxy/auth/test_auth_utils.py b/tests/unit/proxy/auth/test_auth_utils.py index 90e0595dc17..d115634b6f8 100644 --- a/tests/unit/proxy/auth/test_auth_utils.py +++ b/tests/unit/proxy/auth/test_auth_utils.py @@ -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 diff --git a/tests/unit/proxy/common_utils/test_callback_config_validation.py b/tests/unit/proxy/common_utils/test_callback_config_validation.py index 33ae986a511..cac60dff0b4 100644 --- a/tests/unit/proxy/common_utils/test_callback_config_validation.py +++ b/tests/unit/proxy/common_utils/test_callback_config_validation.py @@ -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 diff --git a/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py b/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py index e368a26155b..b7c4bdf7cfd 100644 --- a/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py +++ b/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py @@ -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"] diff --git a/tests/unit/proxy/test_credential_slot_registry.py b/tests/unit/proxy/test_credential_slot_registry.py index 02268dba361..63aa7d6f55f 100644 --- a/tests/unit/proxy/test_credential_slot_registry.py +++ b/tests/unit/proxy/test_credential_slot_registry.py @@ -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(), diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 34cde5bbef6..662dea118a3 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -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