diff --git a/litellm/integrations/otel/README.md b/litellm/integrations/otel/README.md
index 38f96f954f1..4b963ca3415 100644
--- a/litellm/integrations/otel/README.md
+++ b/litellm/integrations/otel/README.md
@@ -295,7 +295,9 @@ nothing here imports outside it:
 - [`config.py`](./model/config.py) — `OpenTelemetryV2Config`, a pydantic-settings
   model that reads `OTEL_*` / `LITELLM_OTEL_*` env vars, plus the feature gate.
   `capture_span_content` gates whether prompt/response bodies may be written as
-  span attributes; it defaults **off** (`no_content`). The Baggage allowlists are
+  span attributes; it defaults **off** (`no_content`). A team or key can override
+  the global mode per OTel v2 destination with `capture_message_content`; omitted
+  values follow the proxy's own callback for that backend, else the global mode. The Baggage allowlists are
   configurable, not hard-coded: set `LITELLM_OTEL_BAGGAGE_PROMOTED_KEYS` /
   `LITELLM_OTEL_BAGGAGE_METADATA_KEYS` /
   `LITELLM_OTEL_BAGGAGE_TEAM_METADATA_KEYS` (comma-separated) as env vars, or
diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py
index d029b153c52..8f6a7d5eb51 100644
--- a/litellm/integrations/otel/langfuse_logger.py
+++ b/litellm/integrations/otel/langfuse_logger.py
@@ -51,6 +51,10 @@ class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
         response: "AsyncIterator[ModelResponseStream]",
         request_data: Mapping[str, object],
     ) -> "AsyncGenerator[ModelResponseStream, None]":
+        if not self._capture_span_content():
+            async for chunk in response:
+                yield chunk
+            return
         relayed: Final[list[ModelResponseStream]] = []  # mutable-ok: relayed as they arrive, assembled at end of stream
         async for chunk in response:
             relayed.append(chunk)
@@ -58,6 +62,8 @@ class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
         self._stamp_root_io(request_data, lambda: stream_output(tuple(relayed), request_data))
 
     def _stamp_root_io(self, data: Mapping[str, object], render_output: Callable[[], str | None]) -> None:
+        if not self._capture_span_content():
+            return
         root: Final = request_root_span()
         if root is None or not root.is_recording():
             return
diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py
index 498ebe655da..d8dccc39c13 100644
--- a/litellm/integrations/otel/logger.py
+++ b/litellm/integrations/otel/logger.py
@@ -58,6 +58,7 @@ from litellm.integrations.otel.plumbing.context import (
     is_recordable_span,
     mcp_message_transport_span,
     post_response_root,
+    request_destinations,
     request_root_http_route,
     request_root_span,
     resolve_internal_call_span_context,
@@ -82,6 +83,7 @@ from litellm.integrations.otel.plumbing.providers import (
     resolve_meter_provider,
 )
 from litellm.integrations.otel.plumbing.routing import TenantTracerCache
+from litellm.types.utils import captures_span_content
 
 if TYPE_CHECKING:
     from opentelemetry.metrics import MeterProvider
@@ -254,6 +256,11 @@ class OpenTelemetryV2(CustomLogger):
             return None
         return GenAIEventRecorder(get_event_logger(provider, LITELLM_TRACER_NAME), provider.resource)
 
+    def _capture_span_content(self) -> bool:
+        return self.config.capture_span_content or any(
+            captures_span_content(destination.capture_message_content) for destination in request_destinations()
+        )
+
     # ====================================================================== #
     #  Proxy global registration
     # ====================================================================== #
@@ -433,7 +440,7 @@ class OpenTelemetryV2(CustomLogger):
             return False
         payload: Final = cast("StandardLoggingPayload", raw_payload)
         data: Final = MCPToolCallSpanData.from_standard_logging_payload(
-            payload, capture_content=self.config.capture_span_content
+            payload, capture_content=self._capture_span_content()
         )
         # A stray LLM carrier from a ``pre_call`` that mis-fired for this id would
         # otherwise linger until evicted; drop it so it's neither leaked nor closed
@@ -477,7 +484,7 @@ class OpenTelemetryV2(CustomLogger):
             return False
         payload: Final = cast("StandardLoggingPayload", raw_payload)
         data: Final = MCPListToolsSpanData.from_standard_logging_payload(
-            payload, capture_content=self.config.capture_span_content
+            payload, capture_content=self._capture_span_content()
         )
         if data.identity.call_id:
             self._release_carrier(self._open_llm_calls.pop(data.identity.call_id, None))
@@ -569,7 +576,7 @@ class OpenTelemetryV2(CustomLogger):
             return None
         data: Final = LLMCallSpanData.from_standard_logging_payload(
             payload,
-            capture_content=self.config.capture_span_content,
+            capture_content=self._capture_span_content(),
             time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
             request_route=request_root_http_route(),
             request_purpose=call.purpose,
@@ -1076,6 +1083,6 @@ def build_otel_v2_logger(
 def _logger_class(config: OpenTelemetryV2Config) -> type[OpenTelemetryV2]:
     if "langfuse" not in config.mapper_names:
         return OpenTelemetryV2
-    from litellm.integrations.otel.langfuse_logger import LangfuseContentOpenTelemetryV2, LangfuseOpenTelemetryV2
+    from litellm.integrations.otel.langfuse_logger import LangfuseContentOpenTelemetryV2
 
-    return LangfuseContentOpenTelemetryV2 if config.capture_span_content else LangfuseOpenTelemetryV2
+    return LangfuseContentOpenTelemetryV2
diff --git a/litellm/integrations/otel/model/config.py b/litellm/integrations/otel/model/config.py
index 3cf49ea2f11..55ef28bf470 100644
--- a/litellm/integrations/otel/model/config.py
+++ b/litellm/integrations/otel/model/config.py
@@ -16,19 +16,17 @@ from litellm.integrations.otel.model.baggage import (
 )
 from litellm.integrations.otel.model.spans import POSTGRESQL, db_system
 from litellm.types.llms.base import LiteLLMBaseModel
-from litellm.types.utils import OtelSpanScope
+from litellm.types.utils import (
+    CAPTURE_MESSAGE_CONTENT_VALUES,
+    CaptureMessageContent,
+    OtelSpanScope,
+    captures_span_content,
+)
 
 #: Master feature-flag env var. The logger is inert until this is truthy.
 OTEL_V2_ENV: Final = "LITELLM_OTEL_V2"
 
 
-class CaptureMessageContent(str):
-    NO_CONTENT = "no_content"
-    SPAN_ONLY = "span_only"
-    EVENT_ONLY = "event_only"
-    SPAN_AND_EVENT = "span_and_event"
-
-
 class ExporterOwner(str, Enum):
     """The preset that contributed an exporter. Values match the callback names
     in ``presets.PRESET_BY_CALLBACK`` so per-request dynamic-credential routing
@@ -297,9 +295,7 @@ class OpenTelemetryV2Config(BaseSettings):
         here are lower_snake_case; normalizing at the boundary keeps both
         spellings working and lets every downstream comparison stay exact.
         """
-        if isinstance(value, str):
-            return value.lower()
-        return value
+        return _normalized_capture_message_content(value)
 
     @field_validator("langfuse_span_scope", mode="before")
     @classmethod
@@ -371,22 +367,39 @@ class OpenTelemetryV2Config(BaseSettings):
 
     @property
     def capture_span_content(self) -> bool:
-        """Whether prompt/response content may be stamped as span attributes.
-
-        Defaults off (``no_content``): an operator must opt in before message
-        bodies leave the process, so a user request can never force its prompt
-        or completion into the configured backend while capture is disabled.
-        """
-        return self.capture_message_content in (
-            CaptureMessageContent.SPAN_ONLY,
-            CaptureMessageContent.SPAN_AND_EVENT,
-        )
+        """Whether the global setting permits prompt/response span attributes."""
+        return captures_span_content(self.capture_message_content)
 
     @classmethod
     def from_env(cls) -> "OpenTelemetryV2Config":
         return cls()
 
 
+def _normalized_capture_message_content(value: object) -> object:
+    if isinstance(value, str):
+        return value.lower()
+    return value
+
+
+class _CaptureMessageContentSettings(BaseSettings):
+    model_config = SettingsConfigDict(extra="ignore", frozen=True)
+
+    capture_message_content: str = Field(
+        default=CaptureMessageContent.NO_CONTENT,
+        validation_alias=AliasChoices("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
+    )
+
+    @field_validator("capture_message_content", mode="before")
+    @classmethod
+    def _normalize_capture_message_content(cls, value: object) -> object:
+        return _normalized_capture_message_content(value)
+
+
+def capture_message_content_from_env() -> str:
+    """The global capture mode, read without building a settings model that rereads every other env var"""
+    return _CaptureMessageContentSettings().capture_message_content
+
+
 _EXCLUDED_SERVICES_INPUT: Final[TypeAdapter[str | tuple[object, ...]]] = TypeAdapter(str | tuple[object, ...])
 
 
@@ -395,6 +408,10 @@ def excluded_db_systems_from(value: object) -> frozenset[str]:
     return _normalize_excluded_services(excluded_service_names(value))
 
 
+def parse_capture_message_content(value: object) -> str | None:
+    return value if isinstance(value, str) and value in CAPTURE_MESSAGE_CONTENT_VALUES else None
+
+
 def excluded_service_names(value: object) -> frozenset[str]:
     """Read a YAML list or comma-separated string of service names, logging and dropping unusable input
     so a malformed value cannot stop the OTel logger from being built"""
diff --git a/litellm/integrations/otel/model/destination.py b/litellm/integrations/otel/model/destination.py
index 56b3123660c..b0c22dea292 100644
--- a/litellm/integrations/otel/model/destination.py
+++ b/litellm/integrations/otel/model/destination.py
@@ -11,7 +11,7 @@ from urllib.parse import quote
 from pydantic import ConfigDict, Field
 
 from litellm.types.llms.base import LiteLLMBaseModel
-from litellm.types.utils import OtelSpanScope
+from litellm.types.utils import OtelSpanScope, captures_span_content
 
 
 class OtelDestination(LiteLLMBaseModel):
@@ -50,6 +50,17 @@ class OtelDestination(LiteLLMBaseModel):
         default=None,
         description="Same, for the requests with a failed span in their tree; ``None`` forwards every one.",
     )
+    capture_message_content: str | None = Field(
+        default=None,
+        description=(
+            "An explicit mode overrides the global capture policy for this destination. "
+            "Omitted follows the global setting."
+        ),
+    )
+
+    def captures_content(self, default: str) -> bool:
+        setting: Final = self.capture_message_content
+        return captures_span_content(default if setting is None else setting)
 
     def header_string(self) -> str:
         """Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects.
@@ -64,9 +75,9 @@ class OtelDestination(LiteLLMBaseModel):
     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`` and the sampling rates are left out on purpose: they decide 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``, the sampling rates and ``capture_message_content`` are left out on
+        purpose: they decide which spans reach the processor and what they carry, not how
+        the processor exports them, 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 37a8f1951d9..4e7d4f231ed 100644
--- a/litellm/integrations/otel/plumbing/providers.py
+++ b/litellm/integrations/otel/plumbing/providers.py
@@ -5,7 +5,7 @@ import random
 import threading
 import time
 from collections import OrderedDict
-from collections.abc import Callable, Iterable, Mapping, Sequence
+from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Final, Literal
 
@@ -40,8 +40,19 @@ 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.model.config import ExporterOwner, ExporterSpec, OpenTelemetryV2Config
+from litellm.integrations.otel.mappers.langfuse import (
+    LANGFUSE_OBSERVATION_INPUT,
+    LANGFUSE_OBSERVATION_OUTPUT,
+    LANGFUSE_TRACE_NAME,
+)
+from litellm.integrations.otel.model.config import (
+    CaptureMessageContent,
+    ExporterOwner,
+    ExporterSpec,
+    OpenTelemetryV2Config,
+    capture_message_content_from_env,
+    captures_span_content,
+)
 from litellm.integrations.otel.model.semconv import (
     DB,
     MCP,
@@ -385,6 +396,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):
@@ -415,6 +446,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)
 
@@ -553,7 +597,11 @@ def _tenant_resource(resource: Resource, defaults: Mapping[str, str], extra: Map
     return filled.merge(Resource(extra)) if extra else filled
 
 
-def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan:
+def _for_destination(
+    span: ReadableSpan,
+    destination: "OtelDestination",
+    default_capture: str,
+) -> ReadableSpan:
     """The view of ``span`` a tenant destination receives.
 
     A span the tenant's own call produced keeps its error text. Every other span is
@@ -564,8 +612,10 @@ 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 whose effective setting disables span 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
     defaults: Final = destination.resource_defaults
@@ -573,11 +623,12 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read
     database: Final = _is_database_span(attributes)
     owned: Final = _is_tenant_owned_span(attributes)
     unreachable: Final = _guardrail_unreachable(attributes)
+    redacted: Final = not destination.captures_content(default_capture)
     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
@@ -612,6 +663,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         shutdown_drain_seconds: float = _SHUTDOWN_DRAIN_SECONDS,
         operator_sinks: 'Mapping[_SinkKey, "OtelSpanScope"]' = MappingProxyType({}),
         excluded_db_systems: frozenset[str] = frozenset(),
+        default_capture: str = CaptureMessageContent.NO_CONTENT,
+        default_capture_by_backend: Mapping[str, str] = MappingProxyType({}),
         pending_drains: int = _MAX_PENDING_DRAINS,
         drain_pool: _DrainPool | None = None,
         sampling_draw: Callable[[], float] | None = None,
@@ -619,6 +672,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         self._operator_sinks: Final = operator_sinks
         self._draw: Final = sampling_draw if sampling_draw is not None else random.random
         self._excluded_db_systems: Final = excluded_db_systems
+        self._default_capture: Final = default_capture
+        self._default_capture_by_backend: Final = default_capture_by_backend
         self._drain_seconds: Final = shutdown_drain_seconds
         self._lock: Final = threading.Condition()
         self._closed = False  # guarded by ``_lock``: an unlocked read races the teardown it gates
@@ -661,12 +716,22 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         for held_span, held_destination in self._settle(trace_id, failed):
             self._forward(held_span, held_destination)
 
+    def _default_capture_for(self, destination: "OtelDestination") -> str:
+        backend: Final = destination.callback_name
+        return (
+            self._default_capture_by_backend.get(backend, self._default_capture) if backend else self._default_capture
+        )
+
     def _forward(self, span: ReadableSpan, destination: "OtelDestination") -> None:
         processor: Final = self._acquire(destination)
         if processor is None:
             return
         try:
-            processor.on_end(_scoped(_for_destination(span, destination), destination.span_scope))
+            processor.on_end(
+                _scoped(
+                    _for_destination(span, destination, self._default_capture_for(destination)), destination.span_scope
+                )
+            )
         except Exception as exc:  # noqa: BLE001  # one destination's failure must not cost the others their span
             verbose_logger.debug("OTel V2 fan-out: forwarding to %s failed: %s", destination.endpoint, exc)
         finally:
@@ -963,14 +1028,15 @@ class _OverriddenBackendFilter(SpanProcessor):
     Wrapping is the only place this works: ``SynchronousMultiSpanProcessor.on_end``
     ignores return values, so a sibling processor can never veto the export.
 
-    Under ``additive`` mode nothing is suppressed, so the wrapper passes every span
-    straight through and the operator keeps its copy.
+    Under ``additive`` mode spans are not suppressed, though global content policy
+    still strips message content when disabled.
 
     ``scope`` narrows what the exporter receives independently of that: under
     ``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`` sets this exporter's content policy; otherwise it follows global capture.
     """
 
     def __init__(
@@ -979,11 +1045,15 @@ class _OverriddenBackendFilter(SpanProcessor):
         owner: str | None,
         scope: "OtelSpanScope" = "full",
         sink: _SinkKey | None = None,
+        global_captures: bool = True,
+        filter_content: bool = True,
     ) -> None:
         self._inner: Final = inner
         self._owner: Final = owner
         self._scope: Final = scope
         self._sink: Final = sink
+        self._global_captures: Final = global_captures
+        self._filter_content: Final = filter_content
 
     def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
         self._inner.on_start(span, parent_context)
@@ -991,7 +1061,28 @@ 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._filter_content and not self._account_captures_content() else scoped
+        )
+
+    def _account_captures_content(self) -> bool:
+        """Whether a destination of this request that names ``sink`` captures 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 content mode.
+        """
+        explicit: Final = next(
+            (
+                destination.capture_message_content
+                for destination in request_destinations()
+                if self._sink is not None
+                and destination.capture_message_content is not None
+                and _sink_key(destination.endpoint, destination.headers) == self._sink
+            ),
+            None,
+        )
+        return captures_span_content(explicit) if explicit is not None else self._global_captures
 
     def _account_scope(self) -> "OtelSpanScope":
         if self._scope == "full" or self._sink is None:
@@ -1010,6 +1101,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.
 
@@ -1261,12 +1371,23 @@ def build_resource(config: OpenTelemetryV2Config) -> Resource:
     return Resource.create(attributes)
 
 
+def _spec_processor(
+    spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool
+) -> 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),
+    )
+    return _MessageContentFilter(exporting) if strip_message_content 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,
+    strip_message_content: bool = False,
 ) -> TracerProvider:
     """Build the shared :class:`TracerProvider`.
 
@@ -1285,6 +1406,9 @@ 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.
+    ``strip_message_content`` removes prompt and response content before every
+    exporter, for a routed provider whose spans can carry content another
+    destination asked for.
     """
     provider: Final = TracerProvider(resource=build_resource(config))
     if baggage_processor is None:
@@ -1300,17 +1424,23 @@ 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, strip_message_content)
         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"
+            _OverriddenBackendFilter(
+                processor,
+                owner,
+                scope,
+                sink,
+                config.capture_span_content,
+                tenant_overrides,
+            )
+            if owner is not None
+            or scope != "full"
+            or sink is not None
+            or (tenant_overrides and not config.capture_span_content)
             else processor
         )
     return provider
@@ -1340,10 +1470,29 @@ def attach_tenant_fan_out(
             TenantFanOutSpanProcessor(
                 operator_sinks=operator_sink_scopes(*configs),
                 excluded_db_systems=excluded_db_systems,
+                default_capture=_operator_capture_mode(configs),
+                default_capture_by_backend=MappingProxyType(dict(_owned_capture_modes(configs))),
             )
         )
 
 
+def _operator_capture_mode(configs: Sequence[OpenTelemetryV2Config]) -> str:
+    """The mode of the operator's generic ``otel`` callback, so ``callback_settings.otel`` counts as the global default"""
+    return next(
+        (config.capture_message_content for config in configs if any(spec.owner is None for spec in config.exporters)),
+        capture_message_content_from_env(),
+    )
+
+
+def _owned_capture_modes(
+    configs: Sequence[OpenTelemetryV2Config],
+) -> Iterator[tuple[str, str]]:
+    for config in configs:
+        for spec in config.exporters:
+            if spec.owner is not None:
+                yield spec.owner.value, config.capture_message_content
+
+
 def deliverable_destinations(
     destinations: Iterable["OtelDestination"],
     provider: trace.TracerProvider | None = None,
diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py
index d7b1cadfc92..eb90b9035c5 100644
--- a/litellm/integrations/otel/plumbing/routing.py
+++ b/litellm/integrations/otel/plumbing/routing.py
@@ -282,7 +282,8 @@ class TenantTracerCache:
             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),
+            strip_message_content=not self._config.capture_span_content,
         )
         self._providers[cache_key] = built
         return built
diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py
index 91341a57634..c4273fbb337 100644
--- a/litellm/integrations/otel/presets/destinations.py
+++ b/litellm/integrations/otel/presets/destinations.py
@@ -187,6 +187,7 @@ def destination_for(
     callback_name: str,
     params: StandardCallbackDynamicParams,
     service_name: str | None = None,
+    capture_message_content: str | None = None,
 ) -> OtelDestination | None:
     """The destination ``params`` names for ``callback_name``, or ``None``.
 
@@ -217,6 +218,7 @@ def destination_for(
         callback_name=callback_name,
         protocol=protocol,
         span_scope=_span_scope(callback_name, params),
+        capture_message_content=capture_message_content,
         success_sampling_rate=success_sampling_rate,
         error_sampling_rate=error_sampling_rate,
     )
diff --git a/litellm/integrations/otel/presets/newrelic.py b/litellm/integrations/otel/presets/newrelic.py
index 771b3f643c8..9d8d04d8f0d 100644
--- a/litellm/integrations/otel/presets/newrelic.py
+++ b/litellm/integrations/otel/presets/newrelic.py
@@ -9,6 +9,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from litellm._logging import verbose_logger
 from litellm.integrations.otel.model.config import (
+    CaptureMessageContent,
     ExporterOwner,
     ExporterSpec,
     OpenTelemetryV2Config,
@@ -66,7 +67,11 @@ def newrelic_preset(
             # New Relic ingests the OTLP GenAI semantic conventions natively.
             "mapper_names": ensure_mappers(base.mapper_names, "genai"),
             **(
-                {"capture_message_content": ("span_only" if settings.record_content else "no_content")}
+                {
+                    "capture_message_content": (
+                        CaptureMessageContent.SPAN_ONLY if settings.record_content else CaptureMessageContent.NO_CONTENT
+                    )
+                }
                 if settings.record_content is not None
                 else {}
             ),
diff --git a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py
index d2ba17e169a..2efeead6bb2 100644
--- a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py
+++ b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py
@@ -6,6 +6,7 @@ from typing import Any, Final
 
 from litellm.types.utils import (
     ARIZE_OTLP_PROTOCOLS,
+    CAPTURE_MESSAGE_CONTENT_VALUES,
     OTEL_SPAN_SCOPES,
     TRUSTED_CALLBACK_VARS_FIELD,
     StandardCallbackDynamicParams,
@@ -91,6 +92,13 @@ def validate_arize_otlp_protocol_value(value: str) -> None:
         raise ValueError(f"Invalid arize_otlp_protocol {value!r}: must be one of {sorted(ARIZE_OTLP_PROTOCOLS)}")
 
 
+def validate_capture_message_content_value(value: str) -> None:
+    if value not in CAPTURE_MESSAGE_CONTENT_VALUES:
+        raise ValueError(
+            f"Invalid capture_message_content {value!r}: must be one of {sorted(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",
diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py
index 0715a4cae97..39892bcc29c 100644
--- a/litellm/proxy/_types.py
+++ b/litellm/proxy/_types.py
@@ -33,6 +33,7 @@ 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_arize_otlp_protocol_value,
+    validate_capture_message_content_value,
     validate_langfuse_environment_value,
     validate_langfuse_span_scope_value,
     validate_no_callback_env_reference,
@@ -79,6 +80,7 @@ from litellm.types.router import AllowedModelRegion, RouterErrors, UpdateRouterC
 from litellm.types.router_weights import validate_router_settings_dict
 from litellm.types.secret_managers.main import KeyManagementSystem
 from litellm.types.utils import (
+    CAPTURE_MESSAGE_CONTENT_VAR,
     AzureSpillover,
     CallTypes,
     CostBreakdown,
@@ -2202,7 +2204,7 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
     @classmethod
     def validate_callback_vars(cls, values):
         callback_vars: Final = values.get("callback_vars", {})
-        valid_keys: Final = set(StandardCallbackDynamicParams.__annotations__.keys())
+        valid_keys: Final = {*StandardCallbackDynamicParams.__annotations__, CAPTURE_MESSAGE_CONTENT_VAR}
         for key, value in callback_vars.items():
             if key not in valid_keys:
                 raise ValueError(f"Invalid callback variable: {key}. Must be one of {valid_keys}")
@@ -2214,6 +2216,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
                 validate_langfuse_span_scope_value(callback_vars[key])
             if key == "arize_otlp_protocol":
                 validate_arize_otlp_protocol_value(callback_vars[key])
+            if key == CAPTURE_MESSAGE_CONTENT_VAR:
+                validate_capture_message_content_value(callback_vars[key])
         return values
 
 
diff --git a/litellm/proxy/common_utils/callback_config_validation.py b/litellm/proxy/common_utils/callback_config_validation.py
index 5b9fb2e3977..4956da853c6 100644
--- a/litellm/proxy/common_utils/callback_config_validation.py
+++ b/litellm/proxy/common_utils/callback_config_validation.py
@@ -10,6 +10,8 @@ from collections.abc import Mapping, Sequence
 from types import MappingProxyType
 from typing import Final
 
+from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR
+
 _NEWRELIC_CALLBACK: Final = "newrelic"
 _NEWRELIC_VAR_PREFIX: Final = "newrelic_"
 _LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel"
@@ -36,6 +38,9 @@ def callback_config_error(
     )
     if langfuse_error is not None:
         return langfuse_error
+    capture_error: Final = _capture_message_content_error(callback_name, callback_vars, callback_type)
+    if capture_error is not None:
+        return capture_error
     if callback_name != _NEWRELIC_CALLBACK:
         return None
     return _newrelic_config_error(callback_vars)
@@ -80,6 +85,32 @@ 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], callback_type: str | None
+) -> str | None:
+    value: Final = callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)
+    if value is None:
+        return None
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    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}"
+    if not is_otel_v2_enabled():
+        return f"Per-destination {CAPTURE_MESSAGE_CONTENT_VAR} requires the proxy to run with LITELLM_OTEL_V2=true."
+    try:
+        validate_capture_message_content_value(value)
+    except ValueError as e:
+        return str(e)
+    if callback_type == "failure":
+        return f"{CAPTURE_MESSAGE_CONTENT_VAR} needs callback_type 'success' or 'success_and_failure'"
+    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,
@@ -190,20 +221,41 @@ def conflicting_span_scope_error(
     )
 
 
+def conflicting_capture_error(
+    callback_name: str,
+    callback_vars: Mapping[str, str] | None,
+    stored_entries: Sequence[tuple[str | None, Mapping[str, str]]],
+) -> str | None:
+    incoming: Final = None if callback_vars is None else callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)
+    if incoming is None:
+        return None
+    return next(
+        (
+            f"{CAPTURE_MESSAGE_CONTENT_VAR} is already set to {stored!r} by another {callback_name} entry. "
+            f"Every {callback_name} entry shares one value: remove that entry or send the same value."
+            for name, entry_vars in stored_entries
+            if name == callback_name and (stored := entry_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)) not in (None, incoming)
+        ),
+        None,
+    )
+
+
 def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None:
     """Validate every ``logging`` entry of a team/key metadata payload."""
     if not metadata:
         return None
-    entries: Final = metadata.get("logging")
-    if not isinstance(entries, Sequence) or isinstance(entries, (str, bytes)):
+    entries: Final = _object_sequence(metadata.get("logging"))
+    if entries is None:
         return None
     entry_vars: Final = tuple(_entry_callback_vars(entry) for entry in entries)
+    named_vars: Final = tuple(zip((_entry_callback_name(entry) for entry in entries), entry_vars))
     return next(
         (
             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_entry_capture_error(named_vars[i], named_vars[:i]) for i in range(len(named_vars))),
             )
             if error is not None
         ),
@@ -211,6 +263,31 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str
     )
 
 
+def stored_capture_entries(logging_entries: object) -> tuple[tuple[str | None, Mapping[str, str]], ...]:
+    """The callback name and vars of every stored ``logging`` entry, for ``conflicting_capture_error``."""
+    entries: Final = _object_sequence(logging_entries) or ()
+    return tuple((_entry_callback_name(entry), _entry_callback_vars(entry)) for entry in entries)
+
+
+def _object_sequence(value: object) -> Sequence[object] | None:
+    return value if isinstance(value, Sequence) and not isinstance(value, (str, bytes)) else None
+
+
+def _entry_callback_name(entry: object) -> str | None:
+    match entry:
+        case {"callback_name": str() as callback_name}:
+            return callback_name
+        case _:
+            return None
+
+
+def _conflicting_entry_capture_error(
+    entry: tuple[str | None, Mapping[str, str]], earlier: Sequence[tuple[str | None, Mapping[str, str]]]
+) -> str | None:
+    callback_name, callback_vars = entry
+    return None if callback_name is None else conflicting_capture_error(callback_name, callback_vars, earlier)
+
+
 def _entry_callback_vars(entry: object) -> Mapping[str, str]:
     callback_vars: Final = entry.get("callback_vars") if isinstance(entry, Mapping) else None
     if not isinstance(callback_vars, Mapping):
diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index 0d693b85ce8..fdc2e6d94b8 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -227,6 +227,7 @@ from litellm.types.litellm_params import AGENTIC_LOOP_KWARG_NAMES
 from litellm.types.llms.anthropic import ANTHROPIC_API_HEADERS
 from litellm.types.services import ServiceTypes
 from litellm.types.utils import (
+    CAPTURE_MESSAGE_CONTENT_VAR,
     CustomPricingLiteLLMParams,
     LlmProviders,
     ProviderSpecificHeader,
@@ -975,6 +976,8 @@ def convert_key_logging_metadata_to_callback(
             team_callback_settings_obj.callbacks.append(data.callback_name)
 
     for var, value in data.callback_vars.items():
+        if var == CAPTURE_MESSAGE_CONTENT_VAR:
+            continue
         # New Relic routing reads these from the trusted-vars overlay with no
         # callback-name check, so scope them to the newrelic entry: a team that
         # put newrelic_* under a different callback never asked for New Relic and
@@ -1162,7 +1165,6 @@ def resolve_tenant_otel_destinations(
     span still routes to the tenant's credentials the way it did then.
     """
     from litellm.integrations.otel.model.config import is_otel_v2_enabled
-    from litellm.integrations.otel.presets.destinations import destination_for
 
     if not is_otel_v2_enabled():
         return ()
@@ -1186,23 +1188,28 @@ def resolve_tenant_otel_destinations(
         for name in dict.fromkeys(
             callback.callback_name for callback in callbacks if callback.callback_type != "failure"
         )
-        if (
-            destination := destination_for(
-                name,
-                _tenant_otel_params(
-                    MappingProxyType(
-                        {
-                            var: value
-                            for callback in callbacks
-                            if callback.callback_name == name
-                            for var, value in callback.callback_vars.items()
-                        }
-                    )
-                ),
-                _tenant_service_name(user_api_key_dict),
+        if (destination := _tenant_destination(name, callbacks, _tenant_service_name(user_api_key_dict))) is not None
+    )
+
+
+def _tenant_destination(
+    name: str, callbacks: Sequence[AddTeamCallback], service_name: str | None
+) -> "OtelDestination | None":
+    from litellm.integrations.otel.model.config import parse_capture_message_content
+    from litellm.integrations.otel.presets.destinations import destination_for
+
+    callback_vars: Final = MappingProxyType(
+        dict(
+            itertools.chain.from_iterable(
+                callback.callback_vars.items() for callback in callbacks if callback.callback_name == name
             )
         )
-        is not None
+    )
+    return destination_for(
+        name,
+        _tenant_otel_params(callback_vars),
+        service_name,
+        parse_capture_message_content(callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)),
     )
 
 
diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py
index 91fd447fb8d..3908ab3c927 100644
--- a/litellm/proxy/management_endpoints/team_callback_endpoints.py
+++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py
@@ -31,8 +31,10 @@ 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_capture_error,
     conflicting_span_scope_error,
     cross_entry_family_error,
+    stored_capture_entries,
 )
 from litellm.proxy.common_utils.callback_utils import (  # noqa: F401  # legacy module exports
     _CALLBACK_VAR_ENCRYPTED_PREFIX,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
@@ -288,6 +290,9 @@ 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 OTel v2 callbacks, "no_content" or "span_only".
+          An explicit value overrides the global setting for this destination; omitted follows the proxy's own callback for that backend, else the global setting.
+          "span_only" puts prompt and response content on the destination's spans, "no_content" leaves it out
         - 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
@@ -351,11 +356,19 @@ async def add_team_callbacks(
         # Decrypted, because the checks compare the incoming values against
         # the stored ones and the credentials are encrypted at rest.
         decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging")
+        stored_capture: Final = stored_capture_entries(decrypted_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)
         if scope_error is not None:
             raise _callback_config_error(scope_error)
+        capture_error: Final = conflicting_capture_error(
+            data.callback_name,
+            data.callback_vars,
+            stored_capture,
+        )
+        if capture_error is not None:
+            raise _callback_config_error(capture_error)
         # One entry has to own a credential family end to end. The entries are
         # flattened into one dict before a request reads them, so an entry
         # naming only a destination would pair with a key written on another
diff --git a/litellm/types/utils.py b/litellm/types/utils.py
index 4a962e5b3bd..d6751610e7a 100644
--- a/litellm/types/utils.py
+++ b/litellm/types/utils.py
@@ -3726,6 +3726,28 @@ ArizeOtlpProtocol = Literal["grpc", "http/protobuf"]
 ARIZE_OTLP_PROTOCOLS: Final[frozenset[str]] = frozenset(get_args(ArizeOtlpProtocol))
 
 
+CAPTURE_MESSAGE_CONTENT_VAR: Final = "capture_message_content"
+
+
+class CaptureMessageContent(str):
+    NO_CONTENT = "no_content"
+    SPAN_ONLY = "span_only"
+    EVENT_ONLY = "event_only"
+    SPAN_AND_EVENT = "span_and_event"
+
+
+# The values a team or key destination may set. OTel v2 writes content only to span attributes,
+# so the event modes stay global-only settings
+CAPTURE_MESSAGE_CONTENT_VALUES: Final[frozenset[str]] = frozenset(
+    {CaptureMessageContent.NO_CONTENT, CaptureMessageContent.SPAN_ONLY}
+)
+
+
+def captures_span_content(mode: str | None) -> bool:
+    """Whether a capture mode puts prompt and response content on spans"""
+    return mode in (CaptureMessageContent.SPAN_ONLY, CaptureMessageContent.SPAN_AND_EVENT)
+
+
 class StandardCallbackDynamicParams(TypedDict, total=False):
     # Langfuse dynamic params
     langfuse_public_key: str | 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..84da9a40486
--- /dev/null
+++ b/tests/integration/observability/test_otel_team_capture_message_content.py
@@ -0,0 +1,720 @@
+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_a_failure_only_entry_rejects_the_setting(rig: Rig) -> None:
+    team: Final = rig.scenario.team()
+    response: Final = rig.attach(
+        team, "langfuse_otel", {**rig.langfuse("fail"), "capture_message_content": "no_content"}, "failure"
+    )
+    assert response.status_code == 400, response.text
+    assert "success_and_failure" in response.text, response.text
+
+
+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_span_only_team_lifts_global_no_content_only_for_its_destination(dark_rig: Rig) -> None:
+    """The sibling Arize entry gets a fan-out destination and the SigNoz entry rides the routed tracer."""
+    team, key = dark_rig.team_key("span_only", "b")
+    siblings: Final = (
+        dark_rig.attach(team, "arize", {"arize_space_key": "space-sibling", "arize_api_key": "arize-sibling"}),
+        dark_rig.attach(
+            team,
+            "signoz",
+            {"signoz_ingestion_endpoint": dark_rig.sinks.arize + "/v1/traces", "signoz_ingestion_key": "signoz-team"},
+        ),
+    )
+    assert [response.status_code for response in siblings] == [200, 200], [response.text for response in siblings]
+    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 = _with_content(dark_rig.sinks.tenant, cursors.tenant, sent)
+    sibling: Final = _arrived(
+        dark_rig.sinks.arize,
+        cursors.arize,
+        sent,
+        frozenset(),
+        ("arize-space-id", "space-sibling"),
+    )
+    signoz: Final = _arrived(
+        dark_rig.sinks.arize,
+        cursors.arize,
+        sent,
+        frozenset(),
+        ("signoz-ingestion-key", "signoz-team"),
+    )
+    assert any(_carries(span, sent.marker) for span in tenant), _names(tenant)
+    assert not any(_carries(span, sent.marker) for span in operator), _names(operator)
+    assert not any(_carries(span, sent.marker) for span in sibling), _names(sibling)
+    assert not any(_carries(span, sent.marker) for span in signoz), _names(signoz)
+
+
+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_second_entry_for_the_same_backend_cannot_disagree(rig: Rig) -> None:
+    team, _ = rig.team_key("no_content", "a")
+    response: Final = rig.attach(team, "langfuse_otel", {"capture_message_content": "span_only"}, "success_and_failure")
+    assert response.status_code == 400, response.text
+    assert "already set to 'no_content' by another langfuse_otel entry" 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, callbacks: tuple[str, ...]) -> Path:
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config["litellm_settings"] = {
+        **config.get("litellm_settings", {}),
+        "callbacks": list(callbacks),
+        "otel_tenant_destination_mode": "additive",
+        "provider_url_destination_allowed_hosts": [
+            urlparse(sinks.tenant).netloc,
+            urlparse(sinks.operator).netloc,
+            urlparse(sinks.arize).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,
+    callbacks: tuple[str, ...] = ("langfuse_otel",),
+) -> 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,
+        "SIGNOZ_INGESTION_ENDPOINT": sinks.arize + "/v1/traces",
+        "SIGNOZ_INGESTION_KEY": "signoz-operator",
+        "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", callbacks),
+            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,
+        ("langfuse_otel", "signoz"),
+    ) as started:
+        yield started
diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py
index aca9dcc8a5e..ff2060d91c2 100644
--- a/tests/unit/integrations/otel/test_langfuse_logger.py
+++ b/tests/unit/integrations/otel/test_langfuse_logger.py
@@ -16,6 +16,7 @@ import litellm  # noqa: E402
 from litellm.caching.dual_cache import DualCache  # noqa: E402
 from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger  # noqa: E402
 from litellm.integrations.otel.model.config import OpenTelemetryV2Config, is_otel_v2_enabled  # noqa: E402
+from litellm.integrations.otel.model.destination import OtelDestination  # noqa: E402
 from litellm.integrations.otel.model.spans import LITELLM_PROXY_REQUEST_SPAN_NAME, SpanRole  # noqa: E402
 from litellm.integrations.otel.plumbing import context as otel_context  # noqa: E402
 from litellm.integrations.otel.plumbing import providers  # noqa: E402
@@ -155,6 +156,25 @@ def test_chat_request_stamps_root_observation_input_and_output():
     assert [(turn["role"], turn["content"]) for turn in output] == [("assistant", "pong")]
 
 
+def test_content_on_destination_stamps_langfuse_root_io_when_global_capture_is_off() -> None:
+    logger, exporter = _logger(capture="no_content")
+    destination: Final = OtelDestination(
+        endpoint="https://team.example.com/api/public/otel",
+        callback_name="langfuse_otel",
+        capture_message_content="span_only",
+    )
+    token: Final = otel_context.set_request_destinations((destination,))
+    try:
+        response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))])
+        _run_request(logger, CHAT_DATA, "acompletion", response)
+    finally:
+        otel_context.reset_request_destinations(token)
+
+    attrs: Final = _root_attrs(exporter)
+    assert json.loads(attrs[INPUT_ATTR]) == [{"role": "user", "content": "ping"}]
+    assert json.loads(attrs[OUTPUT_ATTR]) == [{"role": "assistant", "content": "pong"}]
+
+
 def test_responses_request_folds_instructions_into_input_and_stamps_output_items():
     logger, exporter = _logger()
     data = {"model": "gpt-5.4-mini", "instructions": "be terse", "input": "ping"}
@@ -461,7 +481,9 @@ def test_a_request_without_trace_controls_stamps_none_of_them():
     logger, exporter = _logger()
 
     root_attrs, generation_attrs = _run_named_request(
-        logger, exporter, {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}}
+        logger,
+        exporter,
+        {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}},
     )
 
     assert set(TRACE_CONTROL_ATTRS).isdisjoint(root_attrs)
@@ -472,7 +494,7 @@ def test_a_request_without_trace_controls_stamps_none_of_them():
     ("capture", "mappers"),
     [("no_content", ("genai", "langfuse")), ("span_only", ("genai",))],
 )
-def test_factory_keeps_the_base_logger_unless_langfuse_content_capture_is_on(capture, mappers):
+def test_langfuse_root_io_requires_global_or_destination_capture(capture: str, mappers: tuple[str, ...]) -> None:
     logger, exporter = _logger(capture=capture, mappers=mappers)
 
     _run_request(logger, CHAT_DATA, "acompletion", ModelResponse())
@@ -484,13 +506,13 @@ def test_factory_keeps_the_base_logger_unless_langfuse_content_capture_is_on(cap
     ("capture", "mappers", "relays_streams"),
     [
         ("span_only", ("genai", "langfuse"), True),
-        ("no_content", ("genai", "langfuse"), False),
+        ("no_content", ("genai", "langfuse"), True),
         ("span_only", ("genai",), False),
     ],
 )
-def test_only_langfuse_content_capture_takes_proxy_streams_off_the_fast_path(
-    monkeypatch, capture, mappers, relays_streams
-):
+def test_langfuse_mapper_relays_streams_for_per_request_content_capture(
+    monkeypatch: pytest.MonkeyPatch, capture: str, mappers: tuple[str, ...], relays_streams: bool
+) -> None:
     logger, _ = _logger(capture=capture, mappers=mappers)
     monkeypatch.setattr(litellm, "callbacks", [logger])
 
diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py
index 2d7e0745039..5c9cab88784 100644
--- a/tests/unit/integrations/otel/test_otel_v2_destinations.py
+++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py
@@ -5,6 +5,7 @@ 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
@@ -13,13 +14,14 @@ from typing import Final
 import pytest
 from opentelemetry import trace as trace_api
 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,
@@ -27,13 +29,24 @@ 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 (
+    CaptureMessageContent,
     ExporterOwner,
     ExporterSpec,
     OpenTelemetryV2Config,
     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,
@@ -44,9 +57,11 @@ from litellm.integrations.otel.plumbing.providers import (
     TenantFanOutSpanProcessor,
     _OverriddenBackendFilter,
     _sink_key,
+    attach_tenant_fan_out,
     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
@@ -55,7 +70,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,
@@ -4228,6 +4243,569 @@ class TestTenantHostSsrfGuard:
         assert sum("provider_url_destination_allowed_hosts" in record.message for record in caplog.records) == 1
 
 
+_ALL_MAPPERS: Final = ("genai", "legacy", "openinference", "langfuse", "weave", "langtrace")
+_SECRET: Final = "SECRET-MARKER"
+_MODEL_CALL_WITH_CONTENT: Final = 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: Final = replace(_MODEL_CALL_WITH_CONTENT, messages_in=(), choices_out=())
+_TOOL_CALL_WITH_CONTENT: Final = 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: Final = 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: Final = 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 destination's explicit capture mode overrides the global default."""
+
+    @staticmethod
+    def _fan_out(
+        operator: InMemorySpanExporter,
+        exporters: Mapping[str, InMemorySpanExporter],
+        default_capture: CaptureMessageContent = CaptureMessageContent.SPAN_ONLY,
+    ) -> TracerProvider:
+        provider: Final = TracerProvider()
+        provider.add_span_processor(SimpleSpanProcessor(operator))
+        provider.add_span_processor(
+            TenantFanOutSpanProcessor(
+                processor_factory=lambda d: SimpleSpanProcessor(exporters[d.endpoint]),
+                default_capture=default_capture,
+            )
+        )
+        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(
+        ("setting", "content_exported"),
+        [
+            (None, True),
+            ("no_content", False),
+            ("span_only", True),
+            ("event_only", False),
+            ("span_and_event", True),
+        ],
+    )
+    def test_a_destination_setting_controls_its_copy_and_omission_uses_the_global_default(
+        self, setting: str | None, content_exported: bool
+    ) -> None:
+        attributes: Final = mapped(_MODEL_CALL_WITH_CONTENT)
+        destination: Final = 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: Final = by_name(tenant)["chat gpt-4o"]
+        assert carries_content(exported) is content_exported
+        assert carries_content(by_name(operator)["chat gpt-4o"])
+        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
+
+    def test_a_span_only_team_lifts_global_no_content_only_for_its_destination(self) -> None:
+        operator_exporter, team_exporter, sibling_exporter = (
+            InMemorySpanExporter(),
+            InMemorySpanExporter(),
+            InMemorySpanExporter(),
+        )
+        kind: Final = "lit8244_global_no_content_capture"
+        register_exporter_factory(kind, lambda _spec: operator_exporter)
+        config: Final = OpenTelemetryV2Config(
+            capture_message_content=CaptureMessageContent.NO_CONTENT,
+            exporters=[ExporterSpec(kind=kind)],
+        )
+        provider: Final = build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True)
+        team: Final = OtelDestination(
+            endpoint="http://team.local/api/public/otel",
+            headers=MappingProxyType({"Authorization": "Basic dA=="}),
+            callback_name="langfuse_otel",
+            capture_message_content="span_only",
+        )
+        sibling: Final = INHERITING_DEST
+        provider.add_span_processor(
+            TenantFanOutSpanProcessor(
+                processor_factory=lambda destination: SimpleSpanProcessor(
+                    team_exporter if destination.endpoint == team.endpoint else sibling_exporter
+                ),
+                default_capture=config.capture_message_content,
+            )
+        )
+
+        self._run(
+            provider,
+            (team, sibling),
+            mapped(_MODEL_CALL_WITH_CONTENT),
+        )
+
+        assert carries_content(by_name(team_exporter)["chat gpt-4o"])
+        assert not any(carries_content(span) for span in operator_exporter.get_finished_spans())
+        assert not any(carries_content(span) for span in sibling_exporter.get_finished_spans())
+
+    @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: LLMCallSpanData | MCPToolCallSpanData,
+        without_content: LLMCallSpanData | MCPToolCallSpanData,
+    ) -> None:
+        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: Final = 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) -> None:
+        attributes: Final = mapped(_MODEL_CALL_WITH_CONTENT)
+        operator, restricted, inheriting = InMemorySpanExporter(), InMemorySpanExporter(), InMemorySpanExporter()
+        exporters: Final = {NO_CONTENT_DEST.endpoint: restricted, INHERITING_DEST.endpoint: inheriting}
+
+        self._run(self._fan_out(operator, exporters), (NO_CONTENT_DEST, INHERITING_DEST), attributes)
+
+        original: Final = by_name(operator)
+        kept: Final = by_name(inheriting)
+        stripped: Final = 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) -> None:
+        attributes: Final = mapped(_MODEL_CALL_WITH_CONTENT)
+        span_only: Final = INHERITING_DEST.model_copy(
+            update={"capture_message_content": CaptureMessageContent.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(
+        ("globally_captured", "setting", "content_exported"),
+        [
+            (True, "no_content", False),
+            (True, None, True),
+            (False, "span_only", True),
+            (False, None, False),
+        ],
+    )
+    def test_the_operators_exporter_on_the_same_account_honors_the_teams_setting_once(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        globally_captured: bool,
+        setting: str | None,
+        content_exported: bool,
+    ) -> None:
+        """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: Final = InMemorySpanExporter()
+        provider: Final = TracerProvider()
+        provider.add_span_processor(
+            _OverriddenBackendFilter(
+                SimpleSpanProcessor(shared),
+                "langfuse_otel",
+                "full",
+                TestRoutingMode.OPERATOR_SINK,
+                global_captures=globally_captured,
+            )
+        )
+        provider.add_span_processor(
+            TenantFanOutSpanProcessor(
+                processor_factory=lambda _d: SimpleSpanProcessor(shared),
+                operator_sinks=MappingProxyType({TestRoutingMode.OPERATOR_SINK: "full"}),
+            )
+        )
+        destination: Final = 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: Final = [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: pytest.MonkeyPatch, mode: str
+    ) -> None:
+        monkeypatch.setattr(litellm, "otel_tenant_destination_mode", mode, raising=False)
+        shared: Final = InMemorySpanExporter()
+        kind: Final = f"lit8244_collector_{mode}"
+        register_exporter_factory(kind, lambda _spec: shared)
+        config: Final = OpenTelemetryV2Config(
+            exporters=[
+                ExporterSpec(
+                    kind=kind,
+                    endpoint=TestRoutingMode.OPERATOR_SINK[0],
+                    headers="authorization=Basic op",
+                )
+            ]
+        )
+        provider: Final = 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: Final = 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: Final = [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(
+        ("global_capture", "content_exported"),
+        [(CaptureMessageContent.NO_CONTENT, False), (CaptureMessageContent.SPAN_ONLY, True)],
+    )
+    def test_a_routed_provider_exports_content_only_when_the_global_setting_captures(
+        self, global_capture: str, content_exported: bool
+    ) -> None:
+        """A sibling destination's span_only makes the logger collect content, and a request
+        routed to this callback's team credentials must not carry it under a global no_content."""
+        exporters: dict[ExporterOwner | None, InMemorySpanExporter] = {}
+        kind: Final = f"lit8244_routed_{global_capture}"
+        register_exporter_factory(kind, lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()))
+        config: Final = OpenTelemetryV2Config(
+            capture_message_content=global_capture,
+            exporters=[
+                ExporterSpec(
+                    kind=kind,
+                    endpoint="http://op.local",
+                    owner=ExporterOwner.LANGFUSE_OTEL,
+                    use_simple_processor=True,
+                ),
+                ExporterSpec(kind=kind),
+            ],
+        )
+        cache: Final = TenantTracerCache(config, "langfuse_otel", "litellm")
+
+        route: Final = cache.route_for(
+            get_tracer(TracerProvider(), "litellm"),
+            {"langfuse_public_key": "pk-team", "langfuse_secret_key": "sk-team"},
+        )
+        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
+        for exporter in (exporters[ExporterOwner.LANGFUSE_OTEL], exporters[None]):
+            copy = by_name(exporter)["chat gpt-4o"]
+            assert carries_content(copy) is content_exported
+            assert copy.attributes["gen_ai.usage.input_tokens"] == 12
+
+    @pytest.mark.parametrize("first", ["newrelic", "langfuse_otel"])
+    def test_an_omitted_setting_follows_its_own_callbacks_mode_whatever_the_callback_order(
+        self, monkeypatch: pytest.MonkeyPatch, first: str
+    ) -> None:
+        monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
+        tenant: Final = InMemorySpanExporter()
+        monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, "otlp_http", lambda _spec: tenant)
+        langfuse: Final = OpenTelemetryV2Config(
+            capture_message_content=CaptureMessageContent.SPAN_ONLY,
+            exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.LANGFUSE_OTEL)],
+        )
+        newrelic: Final = OpenTelemetryV2Config(
+            capture_message_content=CaptureMessageContent.NO_CONTENT,
+            exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.NEWRELIC)],
+        )
+        provider: Final = TracerProvider()
+        attach_tenant_fan_out(provider, *((newrelic, langfuse) if first == "newrelic" else (langfuse, newrelic)))
+        destination: Final = OtelDestination(
+            endpoint="http://team.local/api/public/otel",
+            headers=MappingProxyType({"Authorization": "Basic dA=="}),
+            callback_name="langfuse_otel",
+        )
+
+        self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
+        provider.force_flush()
+
+        assert carries_content(by_name(tenant)["chat gpt-4o"])
+
+    def test_an_omitted_setting_follows_the_capture_mode_set_in_callback_settings_otel(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
+        tenant: Final = InMemorySpanExporter()
+        monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, "otlp_http", lambda _spec: tenant)
+        operator: Final = OpenTelemetryV2Config(**{"capture_message_content": "span_only", "exporter": "in_memory"})
+        provider: Final = TracerProvider()
+        attach_tenant_fan_out(provider, operator)
+        destination: Final = OtelDestination(
+            endpoint="http://team.local/api/public/otel",
+            headers=MappingProxyType({"Authorization": "Basic dA=="}),
+            callback_name="langfuse_otel",
+        )
+
+        self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
+        provider.force_flush()
+
+        assert carries_content(by_name(tenant)["chat gpt-4o"]), "the YAML value is the global default, not only the env"
+
+    def test_a_published_presets_capture_mode_is_the_default_for_an_unowned_destination(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A destination for a backend the proxy does not export to itself falls back to the published logger's mode,
+        here New Relic with record_content on, so it keeps the content."""
+        monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
+        tenant: Final = InMemorySpanExporter()
+        monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, "otlp_http", lambda _spec: tenant)
+        newrelic: Final = OpenTelemetryV2Config(
+            capture_message_content=CaptureMessageContent.SPAN_ONLY,
+            exporters=[
+                ExporterSpec(kind="console"),
+                ExporterSpec(kind="in_memory", owner=ExporterOwner.NEWRELIC),
+            ],
+        )
+        provider: Final = TracerProvider()
+        attach_tenant_fan_out(provider, newrelic)
+        destination: Final = OtelDestination(
+            endpoint="http://team.local/api/public/otel",
+            headers=MappingProxyType({"Authorization": "Basic dA=="}),
+            callback_name="langfuse_otel",
+        )
+
+        self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
+        provider.force_flush()
+
+        assert carries_content(by_name(tenant)["chat gpt-4o"])
+
+    @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: pytest.MonkeyPatch, setting: str | None
+    ) -> None:
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        is_otel_v2_enabled.cache_clear()
+        callback_vars: Final = {
+            "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: Final = 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: Final = {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.usefixtures("allow_test_hosts")
+    @pytest.mark.parametrize("setting", ["event_only", "span_and_event"])
+    def test_a_stored_event_mode_is_skipped_like_any_invalid_entry(
+        self, monkeypatch: pytest.MonkeyPatch, setting: str
+    ) -> None:
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        is_otel_v2_enabled.cache_clear()
+        auth: Final = UserAPIKeyAuth(
+            team_metadata={
+                "logging": [
+                    {
+                        "callback_name": "langfuse_otel",
+                        "callback_type": "success",
+                        "callback_vars": {
+                            "langfuse_public_key": "pk-team",
+                            "langfuse_secret_key": "sk-team",
+                            "langfuse_host": "http://team.local",
+                            "capture_message_content": setting,
+                        },
+                    },
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "success",
+                        "callback_vars": {"arize_api_key": "k", "arize_space_id": "s"},
+                    },
+                ]
+            }
+        )
+
+        destinations: Final = [d.callback_name for d in resolve_tenant_otel_destinations(auth)]
+
+        assert destinations == ["arize"]
+
+    @pytest.mark.parametrize("value", ["full", "NO_CONTENT", "", "true", "event_only", "span_and_event"])
+    def test_an_unsupported_value_fails_registration(self, value: str) -> None:
+        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: str) -> None:
+        saved: Final = 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_a_team_wide_callback_vars_map_cannot_carry_the_setting(self) -> None:
+        """The flattened map is shared by every callback, so a value there would apply to all of them."""
+        with pytest.raises(ValueError, match="Invalid callback variable: capture_message_content"):
+            TeamCallbackMetadata(
+                success_callback=["langfuse_otel", "signoz"], callback_vars={"capture_message_content": "span_only"}
+            )
+
+    def test_flattening_the_entries_leaves_the_setting_on_its_own_entry(self) -> None:
+        flattened: Final = convert_key_logging_metadata_to_callback(
+            AddTeamCallback(
+                callback_name="langfuse_otel",
+                callback_type="success",
+                callback_vars={"langfuse_public_key": "pk", "capture_message_content": "span_only"},
+            ),
+            None,
+        )
+
+        assert flattened.callback_vars == {"langfuse_public_key": "pk"}
+
+
 class TestArizeProjectRouting:
     """Arize rejects an export that names no project (400: set x-project-name,
     arize.project.name, openinference.project.name, or model_id), so an Arize
diff --git a/tests/unit/integrations/otel/test_otel_v2_logger.py b/tests/unit/integrations/otel/test_otel_v2_logger.py
index 3b1dd592d08..6a9a963948e 100644
--- a/tests/unit/integrations/otel/test_otel_v2_logger.py
+++ b/tests/unit/integrations/otel/test_otel_v2_logger.py
@@ -9,11 +9,13 @@ hooks, proxy SERVER span lifecycle (start + setters), parent-context resolution
 import ast
 import asyncio
 import contextlib
+import contextvars
 import os
 import re
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from types import MappingProxyType
 from typing import Final
 from unittest.mock import patch
 
@@ -43,7 +45,8 @@ from litellm.integrations.otel import (  # noqa: E402
     OpenTelemetryV2Config,
 )
 from litellm.integrations.otel.logger import OpenTelemetryV2  # noqa: E402
-from litellm.integrations.otel.model.config import ExporterSpec  # noqa: E402
+from litellm.integrations.otel.model.config import CaptureMessageContent, ExporterSpec  # noqa: E402
+from litellm.integrations.otel.model.destination import OtelDestination  # noqa: E402
 from litellm.integrations.otel.model.spans import (  # noqa: E402
     LITELLM_PROXY_REQUEST_SPAN_NAME,
     SpanRole,
@@ -55,6 +58,7 @@ from litellm.integrations.otel.plumbing.context import (  # noqa: E402
     reset_mcp_message_transport_span,
     set_mcp_message_trace_carrier,
     set_mcp_message_transport_span,
+    set_request_destinations,
     set_request_root_span,
 )
 from litellm.types.utils import AUTOROUTER_CLASSIFIER_CALL_ORIGIN  # noqa: E402
@@ -617,8 +621,6 @@ def _mcp_payload(**overrides):
 
 
 def _logger_capturing():
-    from litellm.integrations.otel.model.config import CaptureMessageContent
-
     cfg = OpenTelemetryV2Config(
         exporter="in_memory",
         legacy_compat=False,
@@ -684,6 +686,57 @@ def test_mcp_tool_call_captures_io_when_enabled():
     assert "21" in span.attributes["gen_ai.tool.call.result"]
 
 
+@pytest.mark.parametrize(
+    ("capture_mode", "captures_content"),
+    [
+        (None, False),
+        (CaptureMessageContent.NO_CONTENT, False),
+        (CaptureMessageContent.SPAN_ONLY, True),
+        (CaptureMessageContent.EVENT_ONLY, False),
+        (CaptureMessageContent.SPAN_AND_EVENT, True),
+    ],
+)
+def test_only_a_request_destination_can_turn_on_llm_payload_content(
+    capture_mode: CaptureMessageContent | None, captures_content: bool
+) -> None:
+    config: Final = OpenTelemetryV2Config(
+        exporter="in_memory",
+        capture_message_content=CaptureMessageContent.NO_CONTENT,
+    )
+    exporter: Final = InMemorySpanExporter()
+    logger: Final = OpenTelemetryV2(
+        config=config, tracer_provider=providers.build_tracer_provider(config, exporter=exporter)
+    )
+    payload: Final = _payload(
+        messages=[{"role": "user", "content": "prompt marker"}],
+        response={
+            "id": "resp_1",
+            "model": "gpt-4o-2024",
+            "choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "answer marker"}}],
+        },
+    )
+    kwargs: Final = {
+        **_kwargs(payload),
+        "standard_callback_dynamic_params": {"capture_message_content": "span_only"},
+    }
+    destination: Final = OtelDestination(
+        endpoint="http://team.local/api/public/otel",
+        headers=MappingProxyType({"Authorization": "Basic dA=="}),
+        callback_name="langfuse_otel",
+        capture_message_content=capture_mode,
+    )
+
+    def emit() -> None:
+        set_request_destinations((destination,))
+        _emit_llm(logger, kwargs)
+
+    contextvars.copy_context().run(emit)
+
+    (span,) = exporter.get_finished_spans()
+    assert ("gen_ai.input.messages" in span.attributes) is captures_content
+    assert ("gen_ai.output.messages" in span.attributes) is captures_content
+
+
 def test_mcp_tool_call_failure_marks_error():
     logger, exporter = _logger()
     payload = _mcp_payload(
diff --git a/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py b/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py
index ea0416cecbd..318859cb3ed 100644
--- a/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py
+++ b/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py
@@ -1692,6 +1692,19 @@ def test_capture_message_content_normalizer_only_touches_strings():
         OpenTelemetryV2Config(capture_message_content=123)
 
 
+def test_unknown_capture_message_content_env_is_kept_as_a_plain_string_that_captures_nothing(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "Invalid-Mode")
+
+    config: Final = OpenTelemetryV2Config()
+
+    assert type(config.capture_message_content) is str
+    assert config.capture_message_content == "invalid-mode"
+    assert config.capture_span_content is False
+    assert config.model_dump()["capture_message_content"] == "invalid-mode"
+
+
 def test_v2_flag_is_off_by_default(monkeypatch):
     monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
     is_otel_v2_enabled.cache_clear()
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 d10a96e2aef..a6544e56c57 100644
--- a/tests/unit/proxy/common_utils/test_callback_config_validation.py
+++ b/tests/unit/proxy/common_utils/test_callback_config_validation.py
@@ -1,7 +1,11 @@
+from collections.abc import Iterator
+from typing import Final
+
 import pytest
 
 from litellm.proxy.common_utils.callback_config_validation import (
     callback_config_error,
+    conflicting_capture_error,
     conflicting_span_scope_error,
     cross_entry_family_error,
     logging_metadata_config_error,
@@ -259,3 +263,114 @@ 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.fixture
+def otel_v2_on(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    is_otel_v2_enabled.cache_clear()
+    yield
+    is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+@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: str, value: str) -> None:
+    assert callback_config_error(callback_name, {"capture_message_content": value}) is None
+
+
+def test_capture_message_content_is_rejected_while_otel_v2_is_off(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+    monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+    is_otel_v2_enabled.cache_clear()
+    try:
+        error: Final = callback_config_error("langfuse_otel", {"capture_message_content": "no_content"})
+    finally:
+        is_otel_v2_enabled.cache_clear()
+    assert error is not None and "capture_message_content" in error and "LITELLM_OTEL_V2" in error
+
+
+@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: str | None) -> None:
+    error: Final = 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
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+@pytest.mark.parametrize("value", ["full", "event_only", "span_and_event"])
+def test_an_unsupported_capture_message_content_is_rejected_on_key_logging_metadata(value: str) -> None:
+    error: Final = logging_metadata_config_error(
+        {
+            "logging": [
+                {
+                    "callback_name": "langfuse_otel",
+                    "callback_type": "success",
+                    "callback_vars": {"capture_message_content": value},
+                }
+            ]
+        }
+    )
+    assert error is not None and "Invalid capture_message_content" in error
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+def test_each_entry_keeps_its_own_capture_message_content() -> None:
+    error: Final = logging_metadata_config_error(
+        {
+            "logging": [
+                {
+                    "callback_name": "langfuse_otel",
+                    "callback_type": "success",
+                    "callback_vars": {"capture_message_content": "span_only"},
+                },
+                {
+                    "callback_name": "arize",
+                    "callback_type": "success",
+                    "callback_vars": {"capture_message_content": "no_content"},
+                },
+            ]
+        }
+    )
+    assert error is None
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+def test_a_failure_only_entry_rejects_capture_message_content() -> None:
+    error: Final = callback_config_error("langfuse_otel", {"capture_message_content": "no_content"}, "failure")
+    assert error is not None and "capture_message_content" in error and "success_and_failure" in error
+
+
+@pytest.mark.parametrize(
+    ("stored", "rejected"),
+    [
+        ([("langfuse_otel", {"capture_message_content": "no_content"})], True),
+        ([("langfuse_otel", {"capture_message_content": "span_only"})], False),
+        ([("langfuse_otel", {"langfuse_public_key": "pk"})], False),
+        ([("newrelic", {"capture_message_content": "no_content"})], False),
+    ],
+)
+def test_entries_for_one_backend_share_one_capture_message_content(
+    stored: list[tuple[str, dict[str, str]]], rejected: bool
+) -> None:
+    """The backend's entries merge into one destination, so a second value would silently win or lose."""
+    error: Final = conflicting_capture_error("langfuse_otel", {"capture_message_content": "span_only"}, stored)
+    assert (error is not None) is rejected
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+def test_key_logging_entries_for_one_backend_may_not_disagree_on_capture_message_content() -> None:
+    def entry(callback_type: str, capture: str) -> dict[str, str | dict[str, str]]:
+        return {
+            "callback_name": "langfuse_otel",
+            "callback_type": callback_type,
+            "callback_vars": {"capture_message_content": capture},
+        }
+
+    error: Final = logging_metadata_config_error(
+        {"logging": [entry("success", "no_content"), entry("success_and_failure", "span_only")]}
+    )
+    assert error is not None and "already set to 'no_content' by another langfuse_otel entry" 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 f576bac4742..ad5b6903ece 100644
--- a/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py
+++ b/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py
@@ -7,6 +7,7 @@ redacted audit rows for callback mutations.
 """
 
 import json
+from collections.abc import Iterator
 from typing import Final
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
@@ -1712,3 +1713,62 @@ 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: Final = 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.fixture
+def otel_v2_on(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    is_otel_v2_enabled.cache_clear()
+    yield
+    is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("capture", ["no_content", "span_only"])
+async def test_add_team_callbacks_stores_capture_message_content_in_the_existing_callback_metadata(
+    patched_prisma, capture: str
+) -> None:
+    data: Final = AddTeamCallback(
+        callback_name="langfuse_otel",
+        callback_type="success",
+        callback_vars={
+            "langfuse_public_key": "pk",
+            "langfuse_secret_key": "sk",
+            "capture_message_content": capture,
+        },
+    )
+    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: Final = 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"]] == [capture]
diff --git a/tests/unit/types/test_utils.py b/tests/unit/types/test_utils.py
new file mode 100644
index 00000000000..09063d894e7
--- /dev/null
+++ b/tests/unit/types/test_utils.py
@@ -0,0 +1,40 @@
+import pytest
+
+from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VALUES, CaptureMessageContent, captures_span_content
+
+
+@pytest.mark.parametrize(
+    "mode",
+    [
+        CaptureMessageContent.NO_CONTENT,
+        CaptureMessageContent.SPAN_ONLY,
+        CaptureMessageContent.EVENT_ONLY,
+        CaptureMessageContent.SPAN_AND_EVENT,
+    ],
+)
+def test_capture_message_content_members_are_plain_strings(mode: str) -> None:
+    assert type(mode) is str
+    assert (str(mode), f"{mode}", repr(mode)) == (mode, mode, repr(str(mode)))
+
+
+def test_a_destination_may_only_set_the_span_capture_modes() -> None:
+    assert CAPTURE_MESSAGE_CONTENT_VALUES == {"no_content", "span_only"}
+
+
+def test_capture_message_content_constructor_accepts_any_string() -> None:
+    assert CaptureMessageContent("custom") == "custom"
+
+
+@pytest.mark.parametrize(
+    ("mode", "captures"),
+    [
+        ("no_content", False),
+        ("span_only", True),
+        ("event_only", False),
+        ("span_and_event", True),
+        ("bogus", False),
+        (None, False),
+    ],
+)
+def test_captures_span_content(mode: str | None, captures: bool) -> None:
+    assert captures_span_content(mode) is captures
diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts
index 731a92d8b98..47df4593a65 100644
--- a/ui/litellm-dashboard/src/lib/http/schema.d.ts
+++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts
@@ -17246,6 +17246,9 @@ 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 OTel v2 callbacks, "no_content" or "span_only".
+         *           An explicit value overrides the global setting for this destination; omitted follows the proxy's own callback for that backend, else the global setting.
+         *           "span_only" puts prompt and response content on the destination's spans, "no_content" leaves it out
          *         - 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