refactor(otel/v2): satisfy the type-discipline and strict-lint ceilings

Same ratchet as the credential-model branch: never-rebound names carry Final, the request-state mirror logs at debug instead of swallowing silently, and the empty header and resource fallbacks share one immutable mapping.

Two constructions keep a mutable-ok with a reason: the dedup accumulator consumed on the next line, and the TypedDict rows the resolver returns.

litellm/integrations/otel/plumbing/context.py

@@ -40,7 +40,7 @@ _PROPAGATOR: Final = TraceContextTextMapPropagator()
 # request task, so there is nothing to leak.
 _request_root_span: Final["ContextVar[Span | None]"] = ContextVar("litellm_otel_request_root_span", default=None)
 
-_request_destinations: 'ContextVar[tuple["OtelDestination", ...]]' = ContextVar(
+_request_destinations: Final['ContextVar[tuple["OtelDestination", ...]]'] = ContextVar(
     "litellm_otel_request_destinations", default=()
 )
 

litellm/proxy/auth/user_api_key_auth.py

@@ -11,7 +11,9 @@ import asyncio
 import fnmatch
 import re
 import secrets
+from collections.abc import Mapping
 from datetime import datetime, timezone
+from types import MappingProxyType
 from typing import Any, Final, NamedTuple, Protocol, Union, cast
 
 import fastapi
@@ -108,6 +110,9 @@ except ImportError as e:
 user_api_key_service_logger_obj: Final = ServiceLogging()  # used for tracking latency on OTEL
 
 
+_EMPTY_MAPPING: Final[Mapping[str, str]] = MappingProxyType({})
+
+
 def _normalize_public_auth_route(route: str) -> str:
     if route != "/" and route.endswith("/"):
         return route.rstrip("/")
@@ -1034,22 +1039,22 @@ async def _hoist_request_destinations(request: Request, user_api_key_dict: UserA
         )
 
         destinations_raw, _backends = await _resolve_logging_exporters(user_api_key_dict)
-        destinations = tuple(
+        destinations: Final = tuple(
             OtelDestination(
                 callback_name=item.get("callback_name"),
                 endpoint=item.get("endpoint", ""),
-                headers=item.get("headers") or {},
-                resource_attributes=item.get("resource_attributes") or {},
+                headers=item.get("headers") or _EMPTY_MAPPING,
+                resource_attributes=item.get("resource_attributes") or _EMPTY_MAPPING,
                 protocol=item.get("protocol"),
             )
             for item in destinations_raw
-            if isinstance(item, dict) and item.get("endpoint")
+            if item.get("endpoint")
         )
         set_request_destinations(destinations)
         try:
             request.state.otel_destinations = destinations_raw
-        except Exception:  # noqa: BLE001  # request.state mirror is best-effort; the ContextVar is the source of truth
-            pass
+        except Exception as mirror_exc:  # noqa: BLE001  # the ContextVar is the source of truth
+            verbose_proxy_logger.debug("OTel V2: request.state mirror failed: %s", mirror_exc)
     except Exception as exc:  # noqa: BLE001  # destination hoist is best-effort telemetry setup; it must never fail auth
         verbose_proxy_logger.debug("OTel V2: hoist destination resolution failed: %s", exc)
 

litellm/proxy/litellm_pre_call_utils.py

@@ -719,7 +719,7 @@ async def _effective_org_id(user_api_key_dict: UserAPIKeyAuth) -> str | None:
     """
     if user_api_key_dict.org_id is not None:
         return user_api_key_dict.org_id
-    team_id = user_api_key_dict.team_id
+    team_id: Final = user_api_key_dict.team_id
     if team_id is None:
         return None
     from litellm.proxy import proxy_server
@@ -728,7 +728,7 @@ async def _effective_org_id(user_api_key_dict: UserAPIKeyAuth) -> str | None:
     if proxy_server.prisma_client is None:
         return None
     try:
-        team_obj = await get_team_object(
+        team_obj: Final = await get_team_object(
             team_id=team_id,
             prisma_client=proxy_server.prisma_client,
             user_api_key_cache=proxy_server.user_api_key_cache,
@@ -769,7 +769,7 @@ async def _resolve_logging_exporters(
     if not is_otel_v2_enabled():
         return (), ()
 
-    logging_credentials = tuple(
+    logging_credentials: Final = tuple(
         (credential, info)
         for credential in litellm.credential_list
         if (info := parse_credential_info(credential.credential_info)) is not None and info.credential_type == "logging"
@@ -777,21 +777,21 @@ async def _resolve_logging_exporters(
     if not logging_credentials:
         return (), ()
 
-    team_id = user_api_key_dict.team_id
-    org_id = (
+    team_id: Final = user_api_key_dict.team_id
+    org_id: Final = (
         await _effective_org_id(user_api_key_dict)
         if any(info.access is not None and info.access.orgs for _, info in logging_credentials)
         else None
     )
     team_ids, org_ids = identity_scope(team_id, org_id)
 
-    built = tuple(
+    built: Final = tuple(
         result
         for credential, info in logging_credentials
         if access_grants(info.access, team_ids, org_ids)
         if (result := destination_for_credential(credential)) is not None
     )
-    deduped = {
+    deduped: Final = {  # mutable-ok: dedup accumulator, consumed immediately below
         (
             destination.endpoint,
             tuple(sorted(destination.headers.items())),
@@ -802,8 +802,8 @@ async def _resolve_logging_exporters(
         )
         for backend, destination in built
     }
-    destinations: tuple[OtelDestinationParams, ...] = tuple(
-        {
+    destinations: Final[tuple[OtelDestinationParams, ...]] = tuple(
+        {  # mutable-ok: OtelDestinationParams is a TypedDict, so each entry is a real dict
             "callback_name": backend,
             "endpoint": destination.endpoint,
             "headers": destination.headers,
@@ -812,7 +812,7 @@ async def _resolve_logging_exporters(
         }
         for backend, destination in deduped.values()
     )
-    backends = tuple(dict.fromkeys(backend for backend, _ in deduped.values()))
+    backends: Final = tuple(dict.fromkeys(backend for backend, _ in deduped.values()))
     return destinations, backends
 
 
@@ -883,12 +883,13 @@ async def _apply_admin_logging_exporters(
     if not is_otel_v2_enabled():
         return
     if cached_destinations is not None:
-        destinations = tuple(cached_destinations)
+        destinations = tuple(cached_destinations)  # rebind-ok: the else branch resolves it instead
     else:
         try:
-            destinations, _backends = await _resolve_logging_exporters(user_api_key_dict)
+            resolved, _backends = await _resolve_logging_exporters(user_api_key_dict)
         except Exception:  # noqa: BLE001  # best-effort telemetry setup must never break the request
             return
+        destinations = resolved  # rebind-ok: the cached branch assigns it instead
     _set_request_otel_destinations(destinations)
 
 
@@ -2206,7 +2207,7 @@ async def add_litellm_data_to_request(
             # proxy-owned field instead of the raw request kwargs.
             data[TRUSTED_CALLBACK_VARS_FIELD] = callback_settings_obj.callback_vars
 
-    cached = getattr(getattr(request, "state", None), "otel_destinations", None)
+    cached: Final = getattr(getattr(request, "state", None), "otel_destinations", None)
     await _apply_admin_logging_exporters(user_api_key_dict, cached_destinations=cached)
 
     # Add disabled callbacks from key metadata

litellm/types/utils.py

@@ -39,7 +39,7 @@ from pydantic import (
     field_serializer,
     field_validator,
 )
-from typing_extensions import Required, TypedDict
+from typing_extensions import ReadOnly, Required, TypedDict
 
 from litellm._logging import verbose_logger
 from litellm._uuid import uuid
@@ -3208,11 +3208,11 @@ class OtelDestinationParams(TypedDict, total=False):
     belongs to, so fan-out routes each destination to the right backend's logger.
     """
 
-    callback_name: str
-    endpoint: str
-    headers: Mapping[str, str]
-    resource_attributes: Mapping[str, str]
-    protocol: str | None
+    callback_name: ReadOnly[str]
+    endpoint: ReadOnly[str]
+    headers: ReadOnly[Mapping[str, str]]
+    resource_attributes: ReadOnly[Mapping[str, str]]
+    protocol: ReadOnly[str | None]
 
 
 class StandardCallbackDynamicParams(TypedDict, total=False):

tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py

@@ -299,4 +299,33 @@ def test_resolved_names_keep_every_grant_sharing_one_target(monkeypatch):
             _cred("distinct", access={"teams": ["t1"]}),
         ],
     )
-    assert resolved_logging_exporter_names("t1", None) == ("dup-one", "dup-two", "distinct")
+    assert resolved_logging_exporter_names("t1", None) == ("dup-one", "dup-two", "distinct")
+
+
+
+@pytest.mark.asyncio
+async def test_disclosure_agrees_with_the_resolver_on_a_shared_target(monkeypatch):
+    """Regression: the disclosed names and the resolver's selection are derived from the
+    same grants, so no credential is disclosed that the resolver dropped entirely.
+
+    Pins the two sides together: the resolver collapses the duplicate target to a single
+    export, and every name it kept a destination for is disclosed."""
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.proxy.litellm_pre_call_utils import _resolve_logging_exporters
+
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    is_otel_v2_enabled.cache_clear()
+    same = "http://collector.example/shared/v1/traces"
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            _cred("dup-one", access={"global": True}, endpoint=same),
+            _cred("dup-two", access={"global": True}, endpoint=same),
+        ],
+    )
+
+    destinations, _backends = await _resolve_logging_exporters(UserAPIKeyAuth(api_key="k"))
+
+    assert len(destinations) == 1
+    assert resolved_logging_exporter_names(None, None) == ("dup-one", "dup-two")

tests/test_litellm/proxy/test_litellm_pre_call_utils.py

@@ -5536,27 +5536,6 @@ async def test_client_cannot_control_otel_destinations(_seeded_logging_credentia
     assert not dynamic_params.get("otel_destinations")
 
 
-@pytest.mark.asyncio
-async def test_apply_admin_logging_exporters_registers_on_failure(_seeded_logging_credentials, monkeypatch):
-    """An admin-owned destination must capture a FAILED upstream call, not only a
-    successful one.
-
-    The destination sink is one process-wide logger, so it has to sit on the failure
-    list as well as the success list; registering it on success alone means a
-    401/timeout never reaches the destination and the trace lands with no error
-    gen-AI span. Registration is idempotent.
-    """
-    import litellm
-    from litellm.integrations.otel.destination_logger import admin_destination_logger
-    from litellm.integrations.otel.logger import publish_global_otel_v2_provider
-
-    sink = admin_destination_logger()
-    publish_global_otel_v2_provider([], lambda provider: None)
-    publish_global_otel_v2_provider([], lambda provider: None)
-    for bucket in (litellm._async_success_callback, litellm._async_failure_callback):
-        assert sum(1 for callback in bucket if callback is sink) == 1
-
-
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_merges_metadata_tags_on_responses_route():
     """Regression for #31584: user-supplied metadata.tags must be merged into