fix(otel v2): scope-aware additive dedupe, reject langfuse_span_scope off langfuse_otel, render the scope as a select

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/otel/plumbing/providers.py

@@ -524,7 +524,7 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         self,
         processor_factory: 'Callable[["OtelDestination"], SpanProcessor | None] | None' = None,
         shutdown_drain_seconds: float = _SHUTDOWN_DRAIN_SECONDS,
-        operator_sinks: frozenset[_SinkKey] = frozenset(),
+        operator_sinks: 'Mapping[_SinkKey, "OtelSpanScope"]' = MappingProxyType({}),
         pending_drains: int = _MAX_PENDING_DRAINS,
         drain_pool: _DrainPool | None = None,
     ) -> None:
@@ -544,7 +544,9 @@ class TenantFanOutSpanProcessor(SpanProcessor):
     def on_end(self, span: ReadableSpan) -> None:
         suppressed: Final = suppressed_backends()
         for destination in request_destinations():
-            if self._operator_already_writes(destination, suppressed) or not _in_scope(span, destination.span_scope):
+            if self._operator_already_writes(span, destination, suppressed) or not _in_scope(
+                span, destination.span_scope
+            ):
                 continue
             processor = self._acquire(destination)  # rebind-ok: loop variable; pyright forbids Final in a loop
             if processor is None:
@@ -556,17 +558,22 @@ class TenantFanOutSpanProcessor(SpanProcessor):
             finally:
                 self._release(processor)
 
-    def _operator_already_writes(self, destination: "OtelDestination", suppressed: frozenset[str]) -> bool:
+    def _operator_already_writes(
+        self, span: ReadableSpan, destination: "OtelDestination", suppressed: frozenset[str]
+    ) -> bool:
         """Whether the operator's own exporter is sending this span to the same account.
 
         Only reachable under ``additive``, where nothing is suppressed: a team that
         names the operator's own project would otherwise have every span written
-        there twice, once by the operator's exporter and once by the fan-out.
+        there twice, once by the operator's exporter and once by the fan-out. The
+        operator's exporter may itself be narrowed to the model calls, in which case
+        the rest of the tree is still the fan-out's to deliver.
         """
-        return (
-            destination.callback_name not in suppressed
-            and _sink_key(destination.endpoint, destination.headers) in self._operator_sinks
-        )
+        sink: Final = _sink_key(destination.endpoint, destination.headers)
+        if destination.callback_name in suppressed or sink is None:
+            return False
+        operator_scope: Final = self._operator_sinks.get(sink)
+        return operator_scope is not None and _in_scope(span, operator_scope)
 
     def shutdown(self) -> None:
         """Close every destination processor, once the spans in flight have landed.
@@ -1085,7 +1092,7 @@ def build_tracer_provider(
             (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
         )
         owner = spec.owner.value if tenant_overrides and spec.owner is not None else None
-        scope = config.langfuse_span_scope if spec.owner is ExporterOwner.LANGFUSE_OTEL else "full"
+        scope = _operator_scope(config, spec)
         provider.add_span_processor(
             _OverriddenBackendFilter(processor, owner, scope) if owner is not None or scope != "full" else processor
         )
@@ -1109,7 +1116,7 @@ def attach_tenant_fan_out(provider: TracerProvider, *configs: OpenTelemetryV2Con
     with _FAN_OUT_ATTACH_LOCK:
         if any(isinstance(processor, TenantFanOutSpanProcessor) for processor in _attached_processors(provider)):
             return
-        provider.add_span_processor(TenantFanOutSpanProcessor(operator_sinks=operator_sink_keys(*configs)))
+        provider.add_span_processor(TenantFanOutSpanProcessor(operator_sinks=operator_sink_scopes(*configs)))
 
 
 def deliverable_destinations(
@@ -1134,8 +1141,9 @@ def deliverable_destinations(
     return fan_out.deliverable(destinations) if fan_out is not None else ()
 
 
-def operator_sink_keys(*configs: OpenTelemetryV2Config) -> frozenset[_SinkKey]:
-    """The accounts the operator's own exporters write to, in destination terms.
+def operator_sink_scopes(*configs: OpenTelemetryV2Config) -> 'Mapping[_SinkKey, "OtelSpanScope"]':
+    """The accounts the operator's own exporters write to, in destination terms, and
+    how much of the tree each one receives.
 
     Every v2 logger's config counts, since each logger exports through its own
     provider. An exporter with no endpoint of its own resolves one from the
@@ -1143,14 +1151,20 @@ def operator_sink_keys(*configs: OpenTelemetryV2Config) -> frozenset[_SinkKey]:
     and so is one that never reaches the wire: a console kind ignores the endpoint,
     and a header-gated spec with no credentials is skipped when the provider is built.
     """
-    return frozenset(
-        key
-        for config in configs
-        for spec in config.exporters
-        if _exports_to_the_wire(spec) and (key := _sink_key(spec.endpoint, parse_headers(spec.headers))) is not None
+    return MappingProxyType(
+        {
+            key: _operator_scope(config, spec)
+            for config in configs
+            for spec in config.exporters
+            if _exports_to_the_wire(spec) and (key := _sink_key(spec.endpoint, parse_headers(spec.headers))) is not None
+        }
     )
 
 
+def _operator_scope(config: OpenTelemetryV2Config, spec: ExporterSpec) -> "OtelSpanScope":
+    return config.langfuse_span_scope if spec.owner is ExporterOwner.LANGFUSE_OTEL else "full"
+
+
 def _exports_to_the_wire(spec: ExporterSpec) -> bool:
     """Whether ``build_tracer_provider`` gives ``spec`` an exporter that sends OTLP."""
     return exporter_transport(spec.kind) != "headerless" and not (spec.requires_headers and not spec.headers)

litellm/proxy/common_utils/callback_config_validation.py

@@ -11,12 +11,15 @@ from typing import Final
 
 _NEWRELIC_CALLBACK: Final = "newrelic"
 _NEWRELIC_VAR_PREFIX: Final = "newrelic_"
+_LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel"
 
 
 def callback_config_error(callback_name: str | None, callback_vars: Mapping[str, str] | None) -> str | None:
     if not callback_vars:
         return None
-    langfuse_error: Final = _langfuse_environment_error(callback_vars) or _langfuse_span_scope_error(callback_vars)
+    langfuse_error: Final = _langfuse_environment_error(callback_vars) or _langfuse_span_scope_error(
+        callback_name, callback_vars
+    )
     if langfuse_error is not None:
         return langfuse_error
     if callback_name != _NEWRELIC_CALLBACK:
@@ -44,10 +47,14 @@ def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None:
     return None
 
 
-def _langfuse_span_scope_error(callback_vars: Mapping[str, str]) -> str | None:
+def _langfuse_span_scope_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None:
+    """Only the OTel Langfuse callback reads the scope; on any other callback the
+    value would be stored and then ignored, with the full tree still exported."""
     value: Final = callback_vars.get("langfuse_span_scope")
     if value is None:
         return None
+    if callback_name != _LANGFUSE_OTEL_CALLBACK:
+        return f"langfuse_span_scope applies to the {_LANGFUSE_OTEL_CALLBACK} callback only, not {callback_name!r}"
     from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
         validate_langfuse_span_scope_value,
     )

tests/test_litellm/integrations/otel/test_otel_v2_destinations.py

@@ -42,7 +42,7 @@ from litellm.integrations.otel.plumbing.providers import (
     _sink_key,
     build_tracer_provider,
     deliverable_destinations,
-    operator_sink_keys,
+    operator_sink_scopes,
 )
 from litellm.integrations.otel.plumbing.routing import TenantTracerCache, get_tracer
 from litellm.integrations.otel.presets.arize import arize_preset
@@ -237,7 +237,7 @@ class TestRoutingMode:
         provider.add_span_processor(
             TenantFanOutSpanProcessor(
                 processor_factory=lambda _d: SimpleSpanProcessor(shared),
-                operator_sinks=frozenset({self.OPERATOR_SINK}),
+                operator_sinks=MappingProxyType({self.OPERATOR_SINK: "full"}),
             )
         )
 
@@ -260,7 +260,7 @@ class TestRoutingMode:
         provider.add_span_processor(
             TenantFanOutSpanProcessor(
                 processor_factory=lambda _d: SimpleSpanProcessor(shared),
-                operator_sinks=frozenset({self.OPERATOR_SINK}),
+                operator_sinks=MappingProxyType({self.OPERATOR_SINK: "full"}),
             )
         )
 
@@ -277,7 +277,7 @@ class TestRoutingMode:
         provider.add_span_processor(
             TenantFanOutSpanProcessor(
                 processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter),
-                operator_sinks=frozenset({self.OPERATOR_SINK}),
+                operator_sinks=MappingProxyType({self.OPERATOR_SINK: "full"}),
             )
         )
 
@@ -330,7 +330,7 @@ class TestRoutingMode:
 
         assert global_exporter.get_finished_spans() == ()
 
-    def test_operator_sink_keys_skips_an_exporter_with_no_endpoint_of_its_own(self):
+    def test_operator_sink_scopes_skips_an_exporter_with_no_endpoint_of_its_own(self):
         """Such an exporter resolves its endpoint from the environment at export
         time, so it has no identity to compare a destination against."""
         config = OpenTelemetryV2Config(
@@ -340,9 +340,9 @@ class TestRoutingMode:
             )
         )
 
-        assert operator_sink_keys(config) == frozenset({self.OPERATOR_SINK})
+        assert dict(operator_sink_scopes(config)) == {self.OPERATOR_SINK: "full"}
 
-    def test_operator_sink_keys_skips_exporters_that_never_reach_the_wire(self):
+    def test_operator_sink_scopes_skips_exporters_that_never_reach_the_wire(self):
         """A console kind ignores the endpoint and a header-gated spec with no
         credentials is dropped when the provider is built, so treating either as an
         account the operator writes to would silently withhold a team's own spans
@@ -355,9 +355,9 @@ class TestRoutingMode:
             )
         )
 
-        assert operator_sink_keys(config) == frozenset({self.OPERATOR_SINK})
+        assert dict(operator_sink_scopes(config)) == {self.OPERATOR_SINK: "full"}
 
-    def test_operator_sink_keys_spans_every_config_it_is_handed(self):
+    def test_operator_sink_scopes_spans_every_config_it_is_handed(self):
         first = OpenTelemetryV2Config(
             exporters=(
                 ExporterSpec(
@@ -377,9 +377,9 @@ class TestRoutingMode:
             )
         )
 
-        assert operator_sink_keys(first, second) == {
-            self.OPERATOR_SINK,
-            _sink_key("https://otlp.arize.com/v1/traces", {"space_id": "s", "api_key": "k"}),
+        assert dict(operator_sink_scopes(first, second)) == {
+            self.OPERATOR_SINK: "full",
+            _sink_key("https://otlp.arize.com/v1/traces", {"space_id": "s", "api_key": "k"}): "full",
         }
 
     def test_a_team_pointing_at_a_credential_less_operator_exporter_still_gets_its_spans(self, monkeypatch):
@@ -397,7 +397,7 @@ class TestRoutingMode:
         provider.add_span_processor(
             TenantFanOutSpanProcessor(
                 processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter),
-                operator_sinks=operator_sink_keys(config),
+                operator_sinks=operator_sink_scopes(config),
             )
         )
 
@@ -416,7 +416,7 @@ class TestRoutingMode:
         monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-op")
         monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-op")
         monkeypatch.setattr(litellm, "provider_url_destination_allowed_hosts", ["lf.internal"], raising=False)
-        operator = operator_sink_keys(langfuse_preset())
+        operator = operator_sink_scopes(langfuse_preset())
 
         def sink(public_key, secret_key):
             destination = destination_for(
@@ -450,7 +450,7 @@ class TestRoutingMode:
         monkeypatch.setenv("ARIZE_SPACE_ID", "space-op")
         monkeypatch.setenv("ARIZE_API_KEY", "key-op")
         monkeypatch.delenv("ARIZE_SPACE_KEY", raising=False)
-        operator = operator_sink_keys(arize_preset())
+        operator = operator_sink_scopes(arize_preset())
 
         def sink(space, api_key):
             destination = destination_for(
@@ -1039,7 +1039,9 @@ class TestProviderWiring:
             set_request_destinations(destinations)
             emit(published.tracer_provider)
 
-        in_fresh_context(run, (destination(canonical, dict(pair.split("=") for pair in accounts[canonical][1].split(","))),))
+        in_fresh_context(
+            run, (destination(canonical, dict(pair.split("=") for pair in accounts[canonical][1].split(","))),)
+        )
         in_fresh_context(run, (destination(other, dict(pair.split("=") for pair in accounts[other][1].split(","))),))
         assert shared.get_finished_spans() == (), "an account the operator already writes to was written twice"
 
@@ -1540,6 +1542,56 @@ class TestSpanScope:
         assert operator.get_finished_spans() == ()
         assert names(tenant) == LLM_SPANS
 
+    @staticmethod
+    def _same_account_provider(shared, operator_scope):
+        """The operator's own exporter and a tenant destination naming the same account,
+        both writing one sink, with the operator's exporter narrowed to ``operator_scope``."""
+        provider = TracerProvider()
+        provider.add_span_processor(
+            _OverriddenBackendFilter(SimpleSpanProcessor(shared), "langfuse_otel", operator_scope)
+        )
+        provider.add_span_processor(
+            TenantFanOutSpanProcessor(
+                processor_factory=lambda _d: SimpleSpanProcessor(shared),
+                operator_sinks=MappingProxyType({TestRoutingMode.OPERATOR_SINK: operator_scope}),
+            )
+        )
+        return provider
+
+    @staticmethod
+    def _same_account_destination(span_scope):
+        return OtelDestination(
+            endpoint=TestRoutingMode.SAME_ACCOUNT_ENDPOINT,
+            headers=MappingProxyType({"Authorization": "Basic op"}),
+            callback_name="langfuse_otel",
+            span_scope=span_scope,
+        )
+
+    @pytest.mark.parametrize(
+        ("operator_scope", "tenant_scope", "expected"),
+        [
+            ("llm_only", "full", REQUEST_TREE),
+            ("full", "llm_only", REQUEST_TREE),
+            ("llm_only", "llm_only", LLM_SPANS),
+            ("full", "full", REQUEST_TREE),
+        ],
+    )
+    def test_a_team_naming_the_operators_project_gets_the_wider_of_the_two_scopes_once(
+        self, monkeypatch, operator_scope, tenant_scope, expected
+    ):
+        """Under additive the fan-out stands down for a span the operator's exporter is
+        already sending to that account. When the operator's exporter is narrowed, the
+        spans it drops are not being sent by anyone, so the fan-out still owes them to
+        the team; and no span may land twice."""
+        self._additive(monkeypatch)
+        shared = InMemorySpanExporter()
+
+        self._run(self._same_account_provider(shared, operator_scope), (self._same_account_destination(tenant_scope),))
+
+        finished = [s.name for s in shared.get_finished_spans()]
+        assert frozenset(finished) == expected
+        assert len(finished) == len(expected), "the same account received a span twice"
+
     def test_a_kept_generation_still_hangs_off_the_request_trace_with_its_trace_controls(self, monkeypatch):
         self._additive(monkeypatch)
         operator, tenant = InMemorySpanExporter(), InMemorySpanExporter()
@@ -1679,7 +1731,9 @@ class TestSpanScope:
         assert destination_for("langfuse_otel", creds).span_scope == "full"
 
     def test_only_langfuse_honours_the_scope_var(self):
-        arize = destination_for("arize", {"arize_api_key": "k", "arize_space_id": "s", "langfuse_span_scope": "llm_only"})
+        arize = destination_for(
+            "arize", {"arize_api_key": "k", "arize_space_id": "s", "langfuse_span_scope": "llm_only"}
+        )
 
         assert arize is not None and arize.span_scope == "full"
 
@@ -2365,7 +2419,9 @@ class TestEvictionSafety:
 
             assert len(built) == _MAX_CACHED_DESTINATION_PROCESSORS + 3, "a processor per request during the outage"
             assert sum(1 for accepted in anchored if accepted) == len(built), "anchored what it could not build"
-            assert fan_out.deliverable((self._dest(999),)) == (), "the span would vanish instead of staying with the operator"
+            assert fan_out.deliverable((self._dest(999),)) == (), (
+                "the span would vanish instead of staying with the operator"
+            )
         finally:
             release.set()
         for _ in range(500):

tests/test_litellm/proxy/common_utils/test_callback_config_validation.py

@@ -21,6 +21,17 @@ def test_callback_config_error_rejects_an_unknown_langfuse_span_scope():
     assert callback_config_error("langfuse_otel", {"langfuse_span_scope": "full"}) is None
 
 
+def test_a_span_scope_on_a_callback_that_does_not_read_it_is_rejected():
+    """Only langfuse_otel filters on the scope. Accepting it on the classic Langfuse
+    callback or on an unrelated backend would store a setting that never takes
+    effect, with the full tree still exported."""
+    for callback_name in ["langfuse", "datadog", "otel", None]:
+        error = callback_config_error(callback_name, {"langfuse_span_scope": "llm_only"})
+        assert error is not None and "langfuse_span_scope" in error and "langfuse_otel" in error
+
+    assert callback_config_error("langfuse", {"langfuse_environment": "team-a-prod"}) is None
+
+
 def test_a_bad_span_scope_is_reported_even_when_the_environment_is_fine():
     error = callback_config_error(
         "langfuse_otel", {"langfuse_environment": "team-a-prod", "langfuse_span_scope": "everything"}

ui/litellm-dashboard/src/components/callback_info_helpers.tsx

@@ -17,6 +17,7 @@ interface CallbackConfig {
   logo?: string;
   supports_key_team_logging: boolean;
   dynamic_params: Record<string, "text" | "password" | "select" | "upload" | "number">;
+  dynamic_param_options?: Record<string, readonly string[]>;
   description: string;
 }
 
@@ -126,6 +127,9 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [
       langfuse_environment: "text",
       langfuse_span_scope: "select",
     },
+    dynamic_param_options: {
+      langfuse_span_scope: ["full", "llm_only"],
+    },
     description: "Langfuse v3 OTEL Logging Integration",
   },
   {

ui/litellm-dashboard/src/components/team/LoggingSettings.test.tsx

@@ -216,6 +216,29 @@ describe("LoggingSettings", () => {
     expect(mockOnChange).toHaveBeenCalledWith([expect.objectContaining({ callback_type: "failure" })]);
   });
 
+  it("offers the Langfuse OTEL span scope as a pick between full and llm_only rather than free text", async () => {
+    const user = userEvent.setup({ delay: null });
+    const mockOnChange = vi.fn();
+    const initialValue = [
+      {
+        callback_name: "langfuse_otel",
+        callback_type: "success",
+        callback_vars: {},
+      },
+    ];
+
+    renderWithProviders(<LoggingSettings value={initialValue} onChange={mockOnChange} />);
+
+    expect(screen.queryByPlaceholderText("os.environ/LANGFUSE_SPAN_SCOPE")).not.toBeInTheDocument();
+    await user.click(screen.getByRole("combobox", { name: "langfuse span scope" }));
+    expect((await screen.findAllByRole("option")).map((option) => option.textContent)).toEqual(["full", "llm_only"]);
+    await user.click(screen.getByRole("option", { name: "llm_only" }));
+
+    expect(mockOnChange).toHaveBeenCalledWith([
+      expect.objectContaining({ callback_vars: expect.objectContaining({ langfuse_span_scope: "llm_only" }) }),
+    ]);
+  });
+
   it("correctly handles numerical input with decimal values", () => {
     const mockOnChange = vi.fn();
 

ui/litellm-dashboard/src/components/team/LoggingSettings.tsx

@@ -135,6 +135,55 @@ const LoggingSettings: React.FC<LoggingSettingsProps> = ({
     handleChange(updatedConfigs);
   };
 
+  const renderParamControl = (
+    config: LoggingConfig,
+    configIndex: number,
+    paramName: string,
+    param: { type: string; options: readonly string[] },
+  ) => {
+    const { type: paramType, options } = param;
+    const label = paramName.replace(/_/g, " ");
+    if (options.length > 0) {
+      return (
+        <Select
+          items={options.map((option) => ({ label: option, value: option }))}
+          value={config.callback_vars[paramName] || null}
+          onValueChange={(selected: string | null) => updateCallbackVar(configIndex, paramName, selected ?? "")}
+        >
+          <SelectTrigger aria-label={label} className="w-full">
+            <SelectValue placeholder={`Select ${label}`} />
+          </SelectTrigger>
+          <SelectContent>
+            {options.map((option) => (
+              <SelectItem key={option} value={option}>
+                {option}
+              </SelectItem>
+            ))}
+          </SelectContent>
+        </Select>
+      );
+    }
+    if (paramType === "number") {
+      return (
+        <NumericalInput
+          step={0.01}
+          width={400}
+          placeholder={`os.environ/${paramName.toUpperCase()}`}
+          value={config.callback_vars[paramName] || ""}
+          onChange={(e: any) => updateCallbackVar(configIndex, paramName, e.target.value)}
+        />
+      );
+    }
+    return (
+      <CallbackVarInput
+        sensitive={paramType === "password"}
+        placeholder={`os.environ/${paramName.toUpperCase()}`}
+        value={config.callback_vars[paramName] || ""}
+        onValueChange={(newValue) => updateCallbackVar(configIndex, paramName, newValue)}
+      />
+    );
+  };
+
   const renderDynamicParams = (config: LoggingConfig, configIndex: number) => {
     if (!config.callback_name) return null;
 
@@ -144,6 +193,7 @@ const LoggingSettings: React.FC<LoggingSettingsProps> = ({
     if (!callbackDisplayName) return null;
 
     const dynamicParams = callbackInfo[callbackDisplayName]?.dynamic_params || {};
+    const paramOptions = callbackInfo[callbackDisplayName]?.dynamic_param_options || {};
 
     if (Object.keys(dynamicParams).length === 0) return null;
 
@@ -166,22 +216,10 @@ const LoggingSettings: React.FC<LoggingSettingsProps> = ({
               {paramType === "number" && (
                 <span className="text-xs text-muted-foreground">Value must be between 0 and 1</span>
               )}
-              {paramType === "number" ? (
-                <NumericalInput
-                  step={0.01}
-                  width={400}
-                  placeholder={`os.environ/${paramName.toUpperCase()}`}
-                  value={config.callback_vars[paramName] || ""}
-                  onChange={(e: any) => updateCallbackVar(configIndex, paramName, e.target.value)}
-                />
-              ) : (
-                <CallbackVarInput
-                  sensitive={paramType === "password"}
-                  placeholder={`os.environ/${paramName.toUpperCase()}`}
-                  value={config.callback_vars[paramName] || ""}
-                  onValueChange={(newValue) => updateCallbackVar(configIndex, paramName, newValue)}
-                />
-              )}
+              {renderParamControl(config, configIndex, paramName, {
+                type: paramType,
+                options: paramType === "select" ? paramOptions[paramName] || [] : [],
+              })}
             </div>
           ))}
         </div>