fix(otel): give a key's Langfuse destination the prompt and response the legacy logger sent, only where it ran

litellm/integrations/otel/presets/destinations.py

@@ -15,7 +15,7 @@ import litellm
 from litellm._logging import verbose_logger
 from litellm.integrations.otel.model.destination import OtelDestination
 from litellm.litellm_core_utils.url_utils import is_url_destination_allowed_by_host
-from litellm.types.utils import OtelSpanScope, StandardCallbackDynamicParams
+from litellm.types.utils import CaptureMessageContent, OtelSpanScope, StandardCallbackDynamicParams
 
 #: An endpoint plus the OTLP transport to reach it with, or ``None`` when the backend
 #: names no destination. The transport is ``None`` where the backend has only one.
@@ -124,10 +124,19 @@ def _arize_default_project() -> Mapping[str, str]:
     return MappingProxyType({"model_id": ARIZE_DEFAULT_PROJECT})
 
 
+def _stands_in_for_the_legacy_langfuse_logger() -> bool:
+    """Whether a Langfuse destination replaces the older Langfuse logger's copy, which a proxy without Langfuse keys
+    of its own sends for a key's or team's ``langfuse_otel``."""
+    return not (os.environ.get("LANGFUSE_PUBLIC_KEY") and os.environ.get("LANGFUSE_SECRET_KEY"))
+
+
 def _langfuse_environment(params: StandardCallbackDynamicParams) -> Mapping[str, str]:
-    """The key's or team's ``langfuse_environment``, else ``LANGFUSE_TRACING_ENVIRONMENT``."""
+    """The key's or team's ``langfuse_environment``, else ``LANGFUSE_TRACING_ENVIRONMENT``, where the older Langfuse
+    logger wrote it."""
     from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes
 
+    if not _stands_in_for_the_legacy_langfuse_logger():
+        return _NO_ATTRS
     environment: Final = params.get("langfuse_environment") or os.environ.get("LANGFUSE_TRACING_ENVIRONMENT")
     return MappingProxyType({LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value: environment} if environment else {})
 
@@ -162,6 +171,14 @@ def _resource_defaults(callback_name: str) -> Mapping[str, str]:
     return backend() if backend is not None else _NO_ATTRS
 
 
+def _capture_message_content(callback_name: str, configured: str | None) -> str | None:
+    """The key's or team's own setting, else the prompt and response where the older Langfuse logger sent them
+    whatever the global setting."""
+    if configured is not None or callback_name != "langfuse_otel" or not _stands_in_for_the_legacy_langfuse_logger():
+        return configured
+    return CaptureMessageContent.SPAN_ONLY
+
+
 def _span_scope(callback_name: str, params: StandardCallbackDynamicParams) -> OtelSpanScope:
     if callback_name != "langfuse_otel":
         return "full"
@@ -229,7 +246,7 @@ def destination_for(
         callback_name=callback_name,
         protocol=protocol,
         span_scope=_span_scope(callback_name, params),
-        capture_message_content=capture_message_content,
+        capture_message_content=_capture_message_content(callback_name, capture_message_content),
         excluded_db_systems=excluded_db_systems,
         success_sampling_rate=success_sampling_rate,
         error_sampling_rate=error_sampling_rate,

tests/integration/observability/test_langfuse_otel_tenant_only.py

@@ -135,7 +135,11 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k
         for span in tenant
     ] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant]
     assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"])
+    assert f"lf-tenant-only-{trace_id}" in str(tenant[0]["attributes"].get("langfuse.observation.input")), sorted(
+        tenant[0]["attributes"]
+    )
     assert len(operator) == 1, [span["name"] for span in operator]
+    assert f"lf-tenant-only-{trace_id}" not in json.dumps(operator[0]["attributes"]), sorted(operator[0]["attributes"])
     assert [
         name for name in operator[0]["attributes"] if name in ("user.id", "session.id") or name.startswith("langfuse.")
     ] == [], sorted(operator[0]["attributes"])

tests/unit/integrations/otel/test_otel_v2_destinations.py

@@ -3483,7 +3483,10 @@ class TestLangfuseDestinationFields:
 
     @staticmethod
     def _received_by_account(
-        monkeypatch: pytest.MonkeyPatch, config: OpenTelemetryV2Config, destinations: tuple[OtelDestination, ...]
+        monkeypatch: pytest.MonkeyPatch,
+        config: OpenTelemetryV2Config,
+        destinations: tuple[OtelDestination, ...],
+        with_content: bool = False,
     ) -> Mapping[str | None, Mapping[str, object]]:
         """The attributes of the model call each account received, keyed by the headers it was sent with."""
         credential_less_proxy(monkeypatch)
@@ -3494,7 +3497,7 @@ class TestLangfuseDestinationFields:
         attach_tenant_fan_out(provider, config)
         emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config)
         call: Final = replace(
-            _MODEL_CALL_WITHOUT_CONTENT,
+            _MODEL_CALL_WITH_CONTENT if with_content else _MODEL_CALL_WITHOUT_CONTENT,
             trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)),
             request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}),
         )
@@ -3554,6 +3557,7 @@ class TestLangfuseDestinationFields:
         tracing_environment: str | None,
         expected: Mapping[str, str],
     ) -> None:
+        credential_less_proxy(monkeypatch)
         if tracing_environment is None:
             monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False)
         else:
@@ -3569,6 +3573,76 @@ class TestLangfuseDestinationFields:
         assert destination is not None
         assert dict(destination.resource_attributes) == expected
 
+    def test_a_langfuse_destination_on_a_proxy_with_langfuse_keys_names_no_environment(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """With operator keys the older Langfuse logger never runs, so there is no environment of its to keep."""
+        monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator")
+        monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator")
+
+        destination: Final = destination_for(
+            "langfuse_otel",
+            StandardCallbackDynamicParams(
+                langfuse_public_key="pk", langfuse_secret_key="sk", langfuse_environment="prod"
+            ),
+        )
+
+        assert destination is not None
+        assert dict(destination.resource_attributes) == {}
+
+    @pytest.mark.parametrize(
+        ("configured", "operator_keys", "expected"),
+        [
+            (None, False, "span_only"),
+            ("no_content", False, "no_content"),
+            (None, True, None),
+            ("span_only", True, "span_only"),
+        ],
+        ids=["unset", "key-no-content", "operator-keys", "operator-keys-key-span-only"],
+    )
+    def test_a_langfuse_destination_captures_the_prompt_where_the_legacy_logger_sent_it(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        configured: str | None,
+        operator_keys: bool,
+        expected: str | None,
+    ) -> None:
+        credential_less_proxy(monkeypatch)
+        if operator_keys:
+            monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator")
+            monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator")
+
+        destination: Final = destination_for(
+            "langfuse_otel",
+            StandardCallbackDynamicParams(langfuse_public_key="pk", langfuse_secret_key="sk"),
+            capture_message_content=configured,
+        )
+
+        assert destination is not None
+        assert destination.capture_message_content == expected
+
+    def test_only_the_langfuse_destination_gets_the_prompt_it_captures_by_default(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        credential_less_proxy(monkeypatch)
+        destination: Final = destination_for(
+            "langfuse_otel", StandardCallbackDynamicParams(langfuse_public_key="pk", langfuse_secret_key="sk")
+        )
+        arize: Final = destination_for("arize", StandardCallbackDynamicParams(arize_space_id="s", arize_api_key="k"))
+        assert destination is not None
+        assert arize is not None
+
+        received: Final = self._received_by_account(
+            monkeypatch, OpenTelemetryV2Config(), (destination, arize), with_content=True
+        )
+
+        assert {headers: "langfuse.observation.input" in attributes for headers, attributes in received.items()} == {
+            destination.header_string(): True,
+            arize.header_string(): False,
+            None: False,
+        }
+        assert not [key for key in received[None] if key.startswith("gen_ai.input")]
+
     def test_only_the_langfuse_destination_gets_the_keys_environment(self, monkeypatch: pytest.MonkeyPatch) -> None:
         credential_less_proxy(monkeypatch)
         capture: Final = _ExporterCapture()
@@ -5303,7 +5377,10 @@ class TestCaptureMessageContent:
     def test_the_setting_rides_the_teams_destination_and_omission_stays_omitted(
         self, monkeypatch: pytest.MonkeyPatch, setting: str | None
     ) -> None:
+        """On a proxy with Langfuse keys of its own; without them see ``TestLangfuseDestinationFields``."""
         monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator")
+        monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator")
         is_otel_v2_enabled.cache_clear()
         callback_vars: Final = {
             "langfuse_public_key": "pk-team",