feat(credentials): admin-owned logging credential, access shape, and destination mapping

.github/workflows/test-unit-proxy-endpoints.yml

@@ -32,6 +32,7 @@ jobs:
         tests/test_litellm/proxy/analytics_endpoints
         tests/test_litellm/proxy/management_endpoints
         tests/test_litellm/proxy/memory
+        tests/test_litellm/proxy/credential_endpoints
         tests/test_litellm/proxy/guardrails
         tests/test_litellm/proxy/management_helpers
         tests/test_litellm/proxy/anthropic_endpoints

litellm/constants.py

@@ -464,6 +464,8 @@ DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE: Final = os.getenv(
 
 LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED: Final = 499
 
+LITELLM_LOGGING_CREDENTIAL_NAME_KEY: Final = "litellm_logging_credential_name"
+
 EMAIL_BUDGET_ALERT_TTL: Final = int(os.getenv("EMAIL_BUDGET_ALERT_TTL", 24 * 60 * 60))  # 24 hours in seconds
 EMAIL_BUDGET_ALERT_MAX_SPEND_ALERT_PERCENTAGE: Final = float(
     os.getenv("EMAIL_BUDGET_ALERT_MAX_SPEND_ALERT_PERCENTAGE", 0.8)

litellm/integrations/otel/model/destination.py

@@ -0,0 +1,34 @@
+"""The resolved OTLP destination a request's traces export to.
+
+A destination is a backend-agnostic target: an endpoint plus the auth headers the
+exporter sends. The proxy builds it from the named logging credential bound to the
+request's identity chain, and the v2 logger exports through it. Every OTEL backend
+-- Langfuse, Arize, Weave, a self-hosted collector -- reduces to this shape; the
+per-backend field mapping lives in ``litellm.integrations.otel.presets.destinations``.
+"""
+
+from collections.abc import Mapping
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class OtelDestination(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    endpoint: str
+    headers: Mapping[str, str] = Field(default_factory=dict)
+    resource_attributes: Mapping[str, str] = Field(default_factory=dict)
+    callback_name: str | None = None
+    protocol: str | None = Field(
+        default=None,
+        description=(
+            "OTLP transport for this endpoint (``otlp_http`` / ``otlp_grpc``). The "
+            "backend's intrinsic default is used when unset. A backend whose own cloud "
+            "endpoint is gRPC can still be pointed at an HTTP collector, which the "
+            "scheme alone cannot express: Arize's own ``https://otlp.arize.com/v1`` is gRPC."
+        ),
+    )
+
+    def header_string(self) -> str:
+        """Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects."""
+        return ",".join(f"{key}={value}" for key, value in self.headers.items())

litellm/integrations/otel/presets/destinations.py

@@ -0,0 +1,118 @@
+"""Resolve an admin-owned named credential into a typed OTLP destination.
+
+The destination (endpoint + auth headers) is admin infrastructure config. Each
+OTEL backend stores its own fields on the named credential's free-form
+``credential_values``; the adapter here maps those fields to the universal
+``OtelDestination`` the v2 router exports through. A backend with no bespoke
+adapter is still reachable through the generic ``otel_endpoint`` / ``otel_headers``
+passthrough, so the registry covers every OTEL destination rather than an
+enumerated few. Nothing here reads request data; callers pass admin-resolved
+credential values only.
+"""
+
+import os
+from collections.abc import Callable, Mapping
+
+from litellm.constants import LITELLM_LOGGING_CREDENTIAL_NAME_KEY
+from litellm.integrations.langfuse.langfuse_otel import (
+    LANGFUSE_CLOUD_US_ENDPOINT,
+    LangfuseOtelLogger,
+)
+from litellm.integrations.otel.model.destination import OtelDestination
+from litellm.integrations.weave.weave_otel import _get_weave_authorization_header
+
+LOGGING_CREDENTIAL_NAME_KEY = LITELLM_LOGGING_CREDENTIAL_NAME_KEY
+
+
+def _parse_header_string(raw: str) -> Mapping[str, str]:
+    pairs = (item.split("=", 1) for item in raw.split(",") if "=" in item)
+    return {key.strip(): value.strip() for key, value in pairs}
+
+
+def _langfuse_endpoint(host: str) -> str:
+    normalized = host if host.startswith("http") else f"https://{host}"
+    return f"{normalized.rstrip('/')}/api/public/otel"
+
+
+def _langfuse_destination(values: Mapping[str, str]) -> OtelDestination | None:
+    public_key = values.get("langfuse_public_key")
+    secret_key = values.get("langfuse_secret_key")
+    if not public_key or not secret_key:
+        return None
+    host = values.get("langfuse_host")
+    endpoint = _langfuse_endpoint(host) if host else LANGFUSE_CLOUD_US_ENDPOINT
+    auth = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key)
+    return OtelDestination(endpoint=endpoint, headers={"Authorization": auth})
+
+
+def _arize_destination(values: Mapping[str, str]) -> OtelDestination | None:
+    space = values.get("arize_space_id") or values.get("arize_space_key")
+    api_key = values.get("arize_api_key")
+    if not space or not api_key:
+        return None
+    # Mirrors the global config's ARIZE_ENDPOINT / ARIZE_HTTP_ENDPOINT split: Arize's
+    # own endpoint is gRPC, but a destination may point at an HTTP collector, and the
+    # URL scheme cannot express that (the gRPC endpoint is also https://).
+    http_endpoint = values.get("arize_http_endpoint")
+    endpoint = values.get("arize_endpoint") or http_endpoint or "https://otlp.arize.com/v1"
+    project = values.get("arize_project_name") or values.get("project_name") or os.environ.get("ARIZE_PROJECT_NAME")
+    resource_attributes = {"model_id": project, "arize.project.name": project} if project else {}
+    return OtelDestination(
+        endpoint=endpoint,
+        headers={"space_id": space, "api_key": api_key},
+        resource_attributes=resource_attributes,
+        protocol="otlp_http" if http_endpoint and not values.get("arize_endpoint") else None,
+    )
+
+
+def _weave_destination(values: Mapping[str, str]) -> OtelDestination | None:
+    api_key = values.get("wandb_api_key")
+    if not api_key:
+        return None
+    from litellm.integrations.weave.weave_otel import (
+        WEAVE_BASE_URL,
+        WEAVE_OTEL_ENDPOINT,
+    )
+
+    base = (values.get("weave_endpoint") or WEAVE_BASE_URL).rstrip("/")
+    endpoint = base if base.endswith("/v1/traces") else base.removesuffix("/otel") + WEAVE_OTEL_ENDPOINT
+    headers = {"Authorization": _get_weave_authorization_header(api_key=api_key)}
+    project_id = values.get("weave_project_id")
+    if project_id:
+        headers["project_id"] = project_id
+    return OtelDestination(endpoint=endpoint, headers=headers)
+
+
+def _generic_destination(values: Mapping[str, str]) -> OtelDestination | None:
+    """Any OTLP backend: an explicit endpoint plus raw headers. The catch-all that
+    makes the registry cover self-hosted collectors / Phoenix / Honeycomb / etc."""
+    endpoint = values.get("otel_endpoint")
+    if not endpoint:
+        return None
+    return OtelDestination(endpoint=endpoint, headers=_parse_header_string(values.get("otel_headers", "")))
+
+
+_ADAPTERS: Mapping[str, Callable[[Mapping[str, str]], OtelDestination | None]] = {
+    "langfuse_otel": _langfuse_destination,
+    "arize": _arize_destination,
+    "weave_otel": _weave_destination,
+}
+
+OTEL_V2_DESTINATION_CALLBACKS = frozenset(_ADAPTERS)
+
+
+def build_destination(callback_name: str, values: Mapping[str, str]) -> OtelDestination | None:
+    """Map an admin credential's ``values`` to an ``OtelDestination`` for
+    ``callback_name``, falling back to the generic OTLP passthrough.
+
+    Values are trimmed first: a stray leading/trailing space in an endpoint or
+    host (an easy slip in the create form) yields a malformed OTLP URL the
+    exporter rejects with a 404, so whitespace is never significant here.
+    """
+    trimmed = {key: value.strip() if isinstance(value, str) else value for key, value in values.items()}
+    adapter = _ADAPTERS.get(callback_name)
+    if adapter is not None:
+        destination = adapter(trimmed)
+        if destination is not None:
+            return destination
+    return _generic_destination(trimmed)

litellm/litellm_core_utils/litellm_logging.py

@@ -3603,6 +3603,7 @@ def _get_masked_values(
         "credentials",
         "password",
         "passwd",
+        "otel_headers",
     ]
 
     def _mask_value(v: Any) -> Any:

litellm/models/credentials.py

@@ -5,7 +5,9 @@ These are the canonical credential types for the proxy. They live in the model
 layer; ``litellm.types.utils`` re-exports them for backwards compatibility.
 """
 
-from pydantic import BaseModel, model_validator
+from collections.abc import Mapping
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 
 class CredentialBase(BaseModel):
@@ -18,7 +20,7 @@ class CredentialItem(CredentialBase):
 
 
 class CreateCredentialItem(CredentialBase):
-    credential_values: dict | None = None
+    credential_values: Mapping[str, object] | None = None
     model_id: str | None = None
 
     @model_validator(mode="before")
@@ -27,3 +29,34 @@ class CreateCredentialItem(CredentialBase):
         if not values.get("credential_values") and not values.get("model_id"):
             raise ValueError("Either credential_values or model_id must be set")
         return values
+
+
+class CredentialAccess(BaseModel):
+    """Destination-side access list on a logging credential.
+
+    ``global`` is exposed via the JSON name "global" through a field alias since
+    that's a Python keyword. ``populate_by_name`` keeps internal Python code
+    using ``global_`` working while JSON in/out uses "global".
+    """
+
+    model_config = ConfigDict(populate_by_name=True, extra="forbid")
+
+    global_: bool = Field(default=False, alias="global")
+    teams: tuple[str, ...] = ()
+    orgs: tuple[str, ...] = ()
+
+
+class CredentialInfo(BaseModel):
+    """Typed shape of ``credential_info`` as read by the request-time resolver.
+
+    Existing stored credentials carry arbitrary extra fields (e.g.
+    ``custom_llm_provider``); ``extra="allow"`` preserves them. Only the fields
+    the resolver consumes are typed: ``credential_type`` selects logging
+    destinations, and ``access`` decides which identities the destination fires
+    for.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    credential_type: str | None = None
+    access: CredentialAccess | None = None

litellm/proxy/credential_endpoints/endpoints.py

@@ -13,6 +13,13 @@ from litellm.litellm_core_utils.litellm_logging import _get_masked_values
 from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
+from litellm.proxy.management_endpoints.logging_exporter_access import (
+    destination_for_credential,
+    is_logging_credential,
+)
+from litellm.proxy.management_endpoints.logging_exporter_validation import (
+    validate_credential_access,
+)
 from litellm.proxy.utils import handle_exception_on_proxy, jsonify_object
 from litellm.repositories.credentials_repository import CredentialsRepository
 from litellm.types.utils import CreateCredentialItem, CredentialItem
@@ -55,6 +62,9 @@ async def create_credential(
     """
     from litellm.proxy.proxy_server import llm_router, prisma_client
 
+    if is_logging_credential(credential.credential_info):
+        validate_credential_access(credential.credential_info)
+
     try:
         if prisma_client is None:
             raise HTTPException(
@@ -125,6 +135,11 @@ async def get_credentials(
                 "credential_name": credential.credential_name,
                 "credential_values": _get_masked_values(credential.credential_values),
                 "credential_info": credential.credential_info,
+                **(
+                    {"resolves_to_destination": destination_for_credential(credential) is not None}
+                    if is_logging_credential(credential.credential_info)
+                    else {}
+                ),
             }
             for credential in litellm.credential_list
         ]
@@ -272,12 +287,8 @@ def update_db_credential(
 
         merged_credential.credential_values.update(encrypted_params)
 
-    # update model info
     if encrypted_credential.credential_info:
-        """Update credential info"""
-        if "credential_info" not in merged_credential.credential_info:
-            merged_credential.credential_info = {}
-        merged_credential.credential_info.update(encrypted_credential.credential_info)
+        merged_credential.credential_info = encrypted_credential.credential_info
 
     return merged_credential
 
@@ -309,6 +320,8 @@ async def update_credential(
         db_credential: Final = await credentials_repository.find_by_name(credential_name)
         if db_credential is None:
             raise HTTPException(status_code=404, detail="Credential not found in DB.")
+        if is_logging_credential(db_credential.credential_info) or is_logging_credential(credential.credential_info):
+            validate_credential_access(credential.credential_info)
         merged_credential: Final = update_db_credential(db_credential, credential)
         credential_object_jsonified: Final = jsonify_object(merged_credential.model_dump())
         await credentials_repository.update_by_name(

litellm/proxy/management_endpoints/logging_exporter_access.py

@@ -0,0 +1,161 @@
+"""Request-time routing predicate for admin-owned logging destinations.
+
+``credential_info.access`` answers "which identities' traces may this destination
+receive". It is the sole routing determinant: at call time the resolver in
+``litellm_pre_call_utils`` fires a destination for a request exactly when the
+request's team/org is granted by that destination's ``access``.
+
+``access_grants`` is the primitive: does this ``access`` reach an identity whose
+scope is the given set of team ids and org ids. The resolver passes a
+one-element scope built with ``identity_scope``.
+"""
+
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel, ConfigDict, ValidationError
+
+import litellm
+from litellm.integrations.otel.model.config import is_otel_v2_enabled
+from litellm.models.credentials import CredentialAccess, CredentialInfo, CredentialItem
+
+if TYPE_CHECKING:
+    from litellm.integrations.otel.model.destination import OtelDestination
+
+
+class _LoggingDestinationTag(BaseModel):
+    """Lenient read of just the ``credential_type`` tag, ignoring the rest of
+    ``credential_info`` (including a possibly-malformed ``access``)."""
+
+    model_config = ConfigDict(extra="ignore")
+    credential_type: str | None = None
+
+
+def parse_credential_info(raw: object) -> CredentialInfo | None:
+    """Parse stored ``credential_info`` into the typed model, or ``None`` when it is
+    absent or malformed.
+
+    Callers fail closed on ``None``: a destination whose stored ``access`` cannot be
+    parsed (a legacy shape the strict read model rejects) is treated as granted to
+    no one rather than granted to everyone.
+    """
+    if not isinstance(raw, dict):
+        return None
+    try:
+        return CredentialInfo.model_validate(raw)
+    except ValidationError:
+        return None
+
+
+def is_logging_credential(raw: object) -> bool:
+    """Whether ``credential_info`` is tagged as an admin-owned logging destination.
+
+    The ``access`` shape validation and the ``credential_info`` subfield merge are
+    scoped to these; a provider credential is left on its base replace-and-accept path.
+
+    This keys off the ``credential_type`` tag alone and does not parse ``access``: a
+    destination carrying a malformed ``access`` is still a logging destination, and the
+    point of the gate is to route it into ``validate_credential_access`` so that bad
+    ``access`` is rejected rather than stored.
+    """
+    try:
+        return _LoggingDestinationTag.model_validate(raw).credential_type == "logging"
+    except ValidationError:
+        return False
+
+
+def identity_scope(team_id: str | None, org_id: str | None) -> tuple[frozenset[str], frozenset[str]]:
+    """A single request identity's scope as ``(team_ids, org_ids)`` for
+    ``access_grants``."""
+    return (
+        frozenset({team_id}) if team_id else frozenset(),
+        frozenset({org_id}) if org_id else frozenset(),
+    )
+
+
+def resolved_logging_exporter_names(
+    team_id: str | None,
+    org_id: str | None,
+) -> tuple[str, ...]:
+    """Destination names that will receive this identity's traces, for disclosure on
+    the team/org info pages.
+
+    Mirrors the request-time resolver's selection so it never advertises an exporter that
+    receives no traces: a destination is disclosed only when its ``access`` grants the
+    identity AND it actually builds. Names only; endpoints, headers, and the access map
+    itself stay proxy-admin information.
+
+    Every granting destination is named, including several that resolve to one export
+    target. This answers "which destinations route my traces", not "how many distinct
+    exports happen": the resolver collapses a shared target so the spans are sent once,
+    but each credential named here genuinely grants this identity. Picking one winner
+    per target would have to agree with the resolver's choice, and disagreeing named a
+    credential whose backend the request path never activated.
+
+    Gated on ``is_otel_v2_enabled`` for parity with the resolver, which returns nothing
+    when the flag is off: disclosing a destination the request path would never fire
+    would claim traces are exported when none are.
+    """
+    if not is_otel_v2_enabled():
+        return ()
+    team_ids, org_ids = identity_scope(team_id, org_id)
+    return tuple(
+        credential.credential_name
+        for credential in litellm.credential_list
+        if (info := parse_credential_info(credential.credential_info)) is not None
+        and info.credential_type == "logging"
+        and access_grants(info.access, team_ids, org_ids)
+        if destination_for_credential(credential) is not None
+    )
+
+
+def destination_for_credential(credential: CredentialItem) -> 'tuple[str, "OtelDestination"] | None':
+    """The ``(backend, destination)`` this logging credential resolves to, or ``None`` when it
+    resolves to nothing.
+
+    A credential builds only when it names a backend (``credential_info.description``) and
+    ``build_destination`` accepts its values. Shared by the request-time resolver (which fans
+    out to the built destinations) and the team/org disclosure (which must not advertise a
+    destination that resolves to nothing), so the two cannot drift apart.
+
+    The returned backend is the name the span is routed under, which is the stored one only
+    when ``PRESET_BY_CALLBACK`` has it. A backend outside that registry gets no
+    ``OpenTelemetryV2`` logger, and the logger is what emits the gen-AI span to its
+    destinations, so routing an unregistered name under itself delivered the surrounding
+    trace without the LLM call. ``generic`` is the registered name for a plain OTLP
+    passthrough, which is what ``build_destination`` already fell back to for these.
+    """
+    from litellm.integrations.otel.presets import PRESET_BY_CALLBACK
+    from litellm.integrations.otel.presets.destinations import build_destination
+
+    backend = (credential.credential_info or {}).get("description")
+    if not backend:
+        return None
+    # Drop unset (``None``) values rather than stringifying them: ``str(None)`` is the
+    # literal ``"None"``, which would land in the exporter endpoint/headers and break
+    # the export (e.g. an empty ``otel_endpoint`` becoming the URL ``"None"``).
+    values = {str(key): str(value) for key, value in (credential.credential_values or {}).items() if value is not None}
+    destination = build_destination(backend, values)
+    if destination is None:
+        return None
+    return (backend if backend in PRESET_BY_CALLBACK else "generic", destination)
+
+
+def access_grants(
+    access: CredentialAccess | None,
+    team_ids: frozenset[str],
+    org_ids: frozenset[str],
+) -> bool:
+    """Whether ``access`` grants a destination to an identity scoped to
+    ``team_ids`` / ``org_ids``.
+
+    ``global`` reaches everyone; otherwise one of the identity's teams or orgs
+    must be granted. A missing ``access`` grants no one (fail closed): routing is
+    an explicit admin grant, never the accident of an absent field.
+    """
+    if access is None:
+        return False
+    if access.global_:
+        return True
+    if not team_ids.isdisjoint(access.teams):
+        return True
+    return not org_ids.isdisjoint(access.orgs)

litellm/proxy/management_endpoints/logging_exporter_validation.py

@@ -0,0 +1,45 @@
+"""Shape validation for an admin-owned logging destination's ``credential_info.access``.
+
+Which identities a destination fires for is governed entirely by its
+``credential_info.access``; the resolver (``litellm_pre_call_utils``) evaluates that at
+request time. This module only checks that a write sets a well-formed ``access`` object.
+"""
+
+from collections.abc import Mapping
+
+from fastapi import HTTPException, status
+
+
+def validate_credential_access(credential_info: Mapping[str, object] | None) -> None:
+    """Validate ``credential_info.access`` shape when the write sets one.
+
+    No-op when ``access`` is absent. Otherwise it must be an object whose ``global`` (if
+    present) is a bool and whose ``teams``/``orgs`` (if present) are lists of strings.
+    Per-key access is intentionally unsupported on a destination.
+    """
+    if not isinstance(credential_info, dict) or "access" not in credential_info:
+        return
+    access = credential_info["access"]
+    if not isinstance(access, dict):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={"error": "credential_info.access must be an object"},
+        )
+    if "global" in access and not isinstance(access["global"], bool):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={"error": "access.global must be a boolean"},
+        )
+    for field in ("teams", "orgs"):
+        bucket = access.get(field)
+        if bucket is not None and not (isinstance(bucket, list) and all(isinstance(item, str) for item in bucket)):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail={"error": f"access.{field} must be a list of strings"},
+            )
+    unknown = set(access) - {"global", "teams", "orgs"}
+    if unknown:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={"error": f"access contains unknown field(s): {sorted(unknown)}"},
+        )

tests/test_litellm/integrations/otel/test_presets_destinations.py

@@ -0,0 +1,252 @@
+"""``build_destination`` maps an admin credential to a generic OTLP destination.
+
+The point of these tests is that the resolution is backend-agnostic: Langfuse,
+Arize, Weave, and any raw collector all resolve to an ``{endpoint, headers}``
+the router exports through, and an incomplete credential resolves to nothing.
+"""
+
+import base64
+import os
+import sys
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+from litellm.integrations.otel.presets.destinations import (
+    OTEL_V2_DESTINATION_CALLBACKS,
+    build_destination,
+)
+
+
+def test_langfuse_endpoint_derived_from_host_with_basic_auth():
+    dest = build_destination(
+        "langfuse_otel",
+        {
+            "langfuse_host": "https://cloud.langfuse.com",
+            "langfuse_public_key": "pk-eu",
+            "langfuse_secret_key": "sk-eu",
+        },
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://cloud.langfuse.com/api/public/otel"
+    scheme, b64 = dest.headers["Authorization"].split(" ", 1)
+    assert scheme == "Basic"
+    assert base64.b64decode(b64).decode() == "pk-eu:sk-eu"
+
+
+def test_langfuse_bare_host_gets_https_and_path():
+    dest = build_destination(
+        "langfuse_otel",
+        {
+            "langfuse_host": "my-langfuse.internal",
+            "langfuse_public_key": "pk",
+            "langfuse_secret_key": "sk",
+        },
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://my-langfuse.internal/api/public/otel"
+
+
+def test_langfuse_without_host_defaults_to_us_cloud():
+    dest = build_destination(
+        "langfuse_otel",
+        {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://us.cloud.langfuse.com/api/public/otel"
+
+
+def test_langfuse_incomplete_returns_none():
+    assert build_destination("langfuse_otel", {"langfuse_public_key": "pk"}) is None
+
+
+def test_arize_space_and_api_key_headers(monkeypatch):
+    monkeypatch.delenv("ARIZE_PROJECT_NAME", raising=False)
+    dest = build_destination("arize", {"arize_space_id": "S", "arize_api_key": "K"})
+    assert dest is not None
+    assert dest.endpoint == "https://otlp.arize.com/v1"
+    assert dest.headers == {"space_id": "S", "api_key": "K"}
+    assert dest.protocol is None, "Arize's own endpoint keeps the backend's intrinsic gRPC default"
+
+
+def test_arize_http_endpoint_selects_http_transport(monkeypatch):
+    """Regression: transport was chosen from the backend name alone, so an Arize
+    destination pointed at an HTTP collector was still exported over gRPC -- the
+    collector received nothing and every request drove an exponential-backoff retry
+    storm. The URL scheme cannot express this, since Arize's own gRPC endpoint is
+    also https://, so the destination carries the transport explicitly.
+    """
+    monkeypatch.delenv("ARIZE_PROJECT_NAME", raising=False)
+    dest = build_destination(
+        "arize",
+        {
+            "arize_space_id": "S",
+            "arize_api_key": "K",
+            "arize_http_endpoint": "http://collector.internal/v1/traces",
+        },
+    )
+    assert dest is not None
+    assert dest.endpoint == "http://collector.internal/v1/traces"
+    assert dest.protocol == "otlp_http"
+
+
+def test_arize_grpc_endpoint_wins_over_http_endpoint(monkeypatch):
+    """Mirrors the global config's precedence: ARIZE_ENDPOINT (gRPC) is checked before
+    ARIZE_HTTP_ENDPOINT, so a destination supplying both keeps the gRPC default."""
+    monkeypatch.delenv("ARIZE_PROJECT_NAME", raising=False)
+    dest = build_destination(
+        "arize",
+        {
+            "arize_space_id": "S",
+            "arize_api_key": "K",
+            "arize_endpoint": "https://grpc.internal/v1",
+            "arize_http_endpoint": "http://collector.internal/v1/traces",
+        },
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://grpc.internal/v1"
+    assert dest.protocol is None
+
+
+# --- per-backend Resource declaration -------------------------------------- #
+#
+# Each backend declares the Resource attributes its ingestion needs, in its own
+# builder. Arize is the only first-class backend that routes by a Resource
+# attribute (``model_id``); langfuse / weave / generic route by auth header and
+# declare none. The shared ``destination_resource_attrs`` just reads whatever the
+# builder put on the destination, so the model generalizes: a new backend that
+# needs Resource-level routing only populates ``resource_attributes`` here.
+
+
+def test_arize_project_from_credential_sets_resource_attrs(monkeypatch):
+    monkeypatch.delenv("ARIZE_PROJECT_NAME", raising=False)
+    dest = build_destination(
+        "arize",
+        {"arize_space_id": "S", "arize_api_key": "K", "arize_project_name": "team-x"},
+    )
+    assert dest is not None
+    assert dest.resource_attributes == {
+        "model_id": "team-x",
+        "arize.project.name": "team-x",
+    }
+
+
+def test_arize_project_falls_back_to_env(monkeypatch):
+    monkeypatch.setenv("ARIZE_PROJECT_NAME", "env-proj")
+    dest = build_destination("arize", {"arize_space_id": "S", "arize_api_key": "K"})
+    assert dest is not None
+    assert dest.resource_attributes == {
+        "model_id": "env-proj",
+        "arize.project.name": "env-proj",
+    }
+
+
+def test_arize_credential_project_wins_over_env(monkeypatch):
+    monkeypatch.setenv("ARIZE_PROJECT_NAME", "env-proj")
+    dest = build_destination(
+        "arize",
+        {
+            "arize_space_id": "S",
+            "arize_api_key": "K",
+            "arize_project_name": "cred-proj",
+        },
+    )
+    assert dest is not None
+    assert dest.resource_attributes["model_id"] == "cred-proj"
+
+
+def test_arize_no_project_anywhere_has_empty_resource_attrs(monkeypatch):
+    monkeypatch.delenv("ARIZE_PROJECT_NAME", raising=False)
+    dest = build_destination("arize", {"arize_space_id": "S", "arize_api_key": "K"})
+    assert dest is not None
+    assert dest.resource_attributes == {}
+
+
+def test_header_routed_backends_declare_no_resource_attrs():
+    """langfuse / weave / generic route the project via auth headers, so they
+    declare no Resource attributes -- the generalization counterpart to arize."""
+    langfuse = build_destination(
+        "langfuse_otel",
+        {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
+    )
+    weave = build_destination(
+        "weave_otel",
+        {
+            "wandb_api_key": "w",
+            "weave_endpoint": "https://trace.wandb.ai/otel/v1/traces",
+        },
+    )
+    generic = build_destination(
+        "self_hosted", {"otel_endpoint": "https://collector:4318/v1/traces"}
+    )
+    for dest in (langfuse, weave, generic):
+        assert dest is not None
+        assert dest.resource_attributes == {}
+
+
+def test_weave_requires_only_api_key_and_defaults_endpoint():
+    # No API key -> nothing.
+    assert build_destination("weave_otel", {}) is None
+    # The API key alone is enough: Weave cloud's endpoint is fixed, so it defaults
+    # to the cloud OTLP path and the endpoint field is optional.
+    dest = build_destination(
+        "weave_otel",
+        {"wandb_api_key": "w", "weave_project_id": "entity/project"},
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://trace.wandb.ai/otel/v1/traces"
+    assert dest.headers["project_id"] == "entity/project"
+    assert "Authorization" in dest.headers
+
+
+def test_generic_passthrough_covers_any_backend():
+    dest = build_destination(
+        "some_self_hosted_collector",
+        {
+            "otel_endpoint": "https://collector.internal:4318/v1/traces",
+            "otel_headers": "x-api-key=abc,x-team=42",
+        },
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://collector.internal:4318/v1/traces"
+    assert dest.headers == {"x-api-key": "abc", "x-team": "42"}
+
+
+def test_unknown_backend_without_generic_fields_returns_none():
+    assert build_destination("mystery", {"foo": "bar"}) is None
+
+
+def test_registry_lists_the_first_class_backends():
+    assert OTEL_V2_DESTINATION_CALLBACKS == frozenset(
+        {"langfuse_otel", "arize", "weave_otel"}
+    )
+
+
+def test_endpoint_whitespace_is_trimmed():
+    # A stray leading/trailing space in the endpoint (an easy create-form slip)
+    # makes a malformed OTLP URL the exporter rejects with a 404, so values are
+    # trimmed before the destination is built.
+    dest = build_destination(
+        "some_collector",
+        {"otel_endpoint": "  https://collector.internal:4318/v1/traces  "},
+    )
+    assert dest is not None
+    assert dest.endpoint == "https://collector.internal:4318/v1/traces"
+
+
+def test_weave_endpoint_completed_to_otel_path():
+    # Weave's OTLP path is /otel/v1/traces, not the bare /v1/traces the generic
+    # exporter would append; a host must be completed here or the export 404s.
+    # Idempotent when the full path or the /otel prefix is already supplied.
+    for given, expected in (
+        ("https://trace.wandb.ai", "https://trace.wandb.ai/otel/v1/traces"),
+        ("https://trace.wandb.ai/otel", "https://trace.wandb.ai/otel/v1/traces"),
+        (
+            "https://trace.wandb.ai/otel/v1/traces",
+            "https://trace.wandb.ai/otel/v1/traces",
+        ),
+    ):
+        dest = build_destination(
+            "weave_otel", {"wandb_api_key": "w", "weave_endpoint": given}
+        )
+        assert dest is not None
+        assert dest.endpoint == expected

tests/test_litellm/proxy/credential_endpoints/test_endpoints.py

@@ -10,6 +10,9 @@ from fastapi.testclient import TestClient
 sys.path.insert(0, os.path.abspath("../../../.."))
 
 from litellm.proxy._types import UserAPIKeyAuth
+import litellm
+import litellm.proxy.credential_endpoints.endpoints as endpoints
+from litellm.proxy._types import LitellmUserRoles
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.proxy_server import app
 from litellm.types.utils import CredentialItem
@@ -89,3 +92,168 @@ def test_update_credential_still_answers_200_on_a_successful_write():
 
     assert response.status_code == 200, response.text
     assert response.json()["success"] is True
+
+
+def _admin():
+    return UserAPIKeyAuth(api_key="k", user_role=LitellmUserRoles.PROXY_ADMIN)
+
+
+# --- credential_info replace semantics ---
+
+
+@pytest.mark.asyncio
+async def test_create_credential_validates_access_only_for_logging(monkeypatch):
+    """validate_credential_access runs for a logging destination but never for a
+    provider credential. A provider cred carrying an unrelated `access` key must not be
+    rejected by the destination access-shape validator (that would 400 a valid
+    provider credential the validator was never meant to see)."""
+    import litellm.proxy.proxy_server as proxy_server
+    from litellm.types.utils import CreateCredentialItem
+
+    monkeypatch.setattr(proxy_server, "prisma_client", None, raising=False)
+
+    class _Validated(Exception):
+        pass
+
+    def _spy(_info):
+        raise _Validated()
+
+    monkeypatch.setattr(endpoints, "validate_credential_access", _spy)
+
+    async def _create(credential):
+        return await endpoints.create_credential(
+            request=MagicMock(),
+            fastapi_response=MagicMock(),
+            credential=credential,
+            user_api_key_dict=_admin(),
+        )
+
+    logging_cred = CreateCredentialItem(
+        credential_name="dest",
+        credential_values={"otel_endpoint": "http://collector:4318"},
+        credential_info={"credential_type": "logging", "description": "generic", "access": {"global": True}},
+    )
+    with pytest.raises(_Validated):
+        await _create(logging_cred)
+
+    provider_cred = CreateCredentialItem(
+        credential_name="openai",
+        credential_values={"api_key": "sk"},
+        credential_info={"custom_llm_provider": "openai", "access": {"bogus": True}},
+    )
+    # Validator is skipped; the handler proceeds and fails on the (None) prisma client,
+    # a 500 -- never the _Validated sentinel.
+    with pytest.raises(Exception) as excinfo:
+        await _create(provider_cred)
+    assert not isinstance(excinfo.value, _Validated)
+
+
+# --- PATCH routing regression ------------------------------------------------
+
+def test_patch_credentials_route_targets_update_credential():
+    """Regression: the @router.patch decorator on /credentials/{name:path} must
+    decorate update_credential, not one of the extracted helpers. A misplaced
+    decorator landed once during the 7ecc1d49 split and the unit tests didn't
+    catch it because they import the handler function directly; this asserts
+    the FastAPI routing table actually points at update_credential.
+    """
+    from fastapi.routing import APIRoute
+
+    patch_route = next(
+        route
+        for route in endpoints.router.routes
+        if isinstance(route, APIRoute)
+        and route.path == "/credentials/{credential_name:path}"
+        and "PATCH" in route.methods
+    )
+    assert patch_route.endpoint is endpoints.update_credential
+
+
+@pytest.mark.asyncio
+async def test_get_credentials_masks_secret_values(monkeypatch):
+    """GET /credentials masks secret-bearing values; in particular a destination's
+    otel_headers (which carries the collector auth token) is never returned raw."""
+    raw_headers = "Authorization=Bearer collector-secret,x-api-key=api-secret"
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            CredentialItem(
+                credential_name="openai",
+                credential_values={"api_key": "sk-secret"},
+                credential_info={"custom_llm_provider": "openai"},
+            ),
+            CredentialItem(
+                credential_name="generic-otel",
+                credential_values={"otel_headers": raw_headers},
+                credential_info={"credential_type": "logging", "description": "generic"},
+            ),
+        ],
+    )
+    response = await endpoints.get_credentials(
+        request=MagicMock(),
+        fastapi_response=MagicMock(),
+        user_api_key_dict=_admin(),
+    )
+    names = sorted(c["credential_name"] for c in response["credentials"])
+    assert names == ["generic-otel", "openai"]
+    generic = next(c for c in response["credentials"] if c["credential_name"] == "generic-otel")
+    # otel_headers carries the collector auth token, so the masker treats it as a
+    # secret key: readable prefix only, never the full value.
+    assert generic["credential_values"]["otel_headers"] != raw_headers
+    assert "collector-secret" not in str(response)
+    assert "sk-secret" not in str(response)
+
+
+@pytest.mark.asyncio
+async def test_get_credentials_reports_whether_each_destination_actually_builds(monkeypatch):
+    """The dashboard needs the resolver's own verdict, not a second implementation of it.
+
+    Its Scope column read ``credential_info.access`` alone, so a destination the resolver
+    excludes (no backend name, or values its adapter rejects) still rendered a scope badge
+    and read as live. Reproducing the adapter rules in the frontend would drift from them;
+    this field is computed by ``destination_for_credential``, the same function the
+    request-time resolver and the team/org disclosure use.
+    """
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            CredentialItem(
+                credential_name="builds",
+                credential_values={"otel_endpoint": "http://collector.internal:4318/v1/traces"},
+                credential_info={"credential_type": "logging", "description": "generic", "access": {"global": True}},
+            ),
+            CredentialItem(
+                credential_name="no-backend",
+                credential_values={"otel_endpoint": "http://collector.internal:4318/v1/traces"},
+                credential_info={"credential_type": "logging", "access": {"global": True}},
+            ),
+            CredentialItem(
+                credential_name="adapter-rejects",
+                credential_values={"langfuse_public_key": "pk-only"},
+                credential_info={
+                    "credential_type": "logging",
+                    "description": "langfuse_otel",
+                    "access": {"global": True},
+                },
+            ),
+            CredentialItem(
+                credential_name="openai",
+                credential_values={"api_key": "sk-secret"},
+                credential_info={"custom_llm_provider": "openai"},
+            ),
+        ],
+    )
+    response = await endpoints.get_credentials(
+        request=MagicMock(),
+        fastapi_response=MagicMock(),
+        user_api_key_dict=_admin(),
+    )
+    verdicts = {c["credential_name"]: c.get("resolves_to_destination") for c in response["credentials"]}
+    assert verdicts["builds"] is True
+    assert verdicts["no-backend"] is False
+    assert verdicts["adapter-rejects"] is False
+    # A provider credential is not a destination and gets no verdict at all, rather than a
+    # False that would render it as a broken destination.
+    assert verdicts["openai"] is None

tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py

@@ -0,0 +1,361 @@
+"""The request-time routing predicate for admin-owned logging destinations.
+
+``access_grants`` is the chokepoint the resolver routes through at call time, so a
+mutation here would route an identity's traces to a destination outside its scope.
+Each case is written to fail if the corresponding branch is flipped.
+"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+import litellm
+from litellm.models.credentials import CredentialAccess, CredentialInfo, CredentialItem
+from litellm.proxy.management_endpoints.logging_exporter_access import (
+    access_grants,
+    destination_for_credential,
+    identity_scope,
+    is_logging_credential,
+    parse_credential_info,
+    resolved_logging_exporter_names,
+)
+
+import pytest
+
+from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+
+@pytest.fixture(autouse=True)
+def _reset_otel_v2_flag_cache():
+    """``is_otel_v2_enabled`` is lru-cached; clear it around each test so ``LITELLM_OTEL_V2``
+    toggles take effect and don't leak between tests."""
+    is_otel_v2_enabled.cache_clear()
+    yield
+    is_otel_v2_enabled.cache_clear()
+
+
+# --- is_logging_credential: the access-validation + merge gate ---------------
+
+
+def test_is_logging_credential_true_for_logging_type():
+    assert is_logging_credential({"credential_type": "logging", "description": "arize"}) is True
+
+
+def test_is_logging_credential_true_even_with_malformed_access():
+    """A destination carrying an invalid access shape is still a logging destination.
+    The gate must route it into validate_credential_access (which rejects it), not skip
+    validation because the strict access model can't parse it."""
+    assert is_logging_credential({"credential_type": "logging", "access": {"nonsense_field": True}}) is True
+
+
+def test_is_logging_credential_false_for_provider_and_malformed():
+    assert is_logging_credential({"custom_llm_provider": "openai"}) is False
+    assert is_logging_credential({"custom_llm_provider": "openai", "access": {"global": True}}) is False
+    assert is_logging_credential(None) is False
+    assert is_logging_credential("nope") is False
+
+
+# --- parse_credential_info: fail closed on bad input -----------------------
+
+
+def test_parse_none_for_non_dict():
+    assert parse_credential_info(None) is None
+    assert parse_credential_info("not a dict") is None
+    assert parse_credential_info(["a"]) is None
+
+
+def test_parse_typed_access():
+    info = parse_credential_info(
+        {
+            "credential_type": "logging",
+            "description": "arize",
+            "access": {"global": True, "teams": ["t1"], "orgs": ["o1"]},
+        }
+    )
+    assert info is not None
+    assert info.credential_type == "logging"
+    assert info.access is not None
+    assert info.access.global_ is True
+    assert info.access.teams == ("t1",)
+    assert info.access.orgs == ("o1",)
+
+
+def test_parse_missing_access_is_none_not_error():
+    info = parse_credential_info({"credential_type": "logging"})
+    assert info is not None
+    assert info.access is None
+
+
+def test_parse_malformed_access_fails_closed():
+    """A stored access with an unknown field is rejected by the strict read model;
+    the parse must return None (invisible) rather than raise or grant."""
+    assert parse_credential_info({"access": {"legacy_field": "x"}}) is None
+    assert parse_credential_info({"access": "not-an-object"}) is None
+
+
+# --- access_grants: the primitive ------------------------------------------
+
+
+def _access(**kw) -> CredentialAccess:
+    return CredentialAccess.model_validate(kw)
+
+
+def test_access_grants_global_reaches_empty_scope():
+    assert access_grants(_access(**{"global": True}), frozenset(), frozenset()) is True
+
+
+def test_access_grants_none_denies():
+    assert access_grants(None, frozenset({"t1"}), frozenset({"o1"})) is False
+
+
+def test_access_grants_team_match():
+    a = _access(teams=["t1", "t2"])
+    assert access_grants(a, frozenset({"t2"}), frozenset()) is True
+    assert access_grants(a, frozenset({"t3"}), frozenset()) is False
+
+
+def test_access_grants_org_match():
+    a = _access(orgs=["o1"])
+    assert access_grants(a, frozenset(), frozenset({"o1"})) is True
+    assert access_grants(a, frozenset(), frozenset({"o2"})) is False
+
+
+def test_access_grants_disjoint_denies():
+    a = _access(teams=["t1"], orgs=["o1"])
+    assert access_grants(a, frozenset({"t9"}), frozenset({"o9"})) is False
+
+
+def test_access_grants_not_global_when_false():
+    """global=False must not short-circuit to visible."""
+    a = _access(**{"global": False})
+    assert access_grants(a, frozenset({"t1"}), frozenset({"o1"})) is False
+
+
+# --- routing scope decided entirely by access -------------------------------
+def test_empty_access_is_deny_all():
+    """Empty access grants no one: not proxy-wide."""
+    info = CredentialInfo(credential_type="logging")
+    assert access_grants(info.access, frozenset(), frozenset()) is False
+    assert access_grants(info.access, frozenset({"any-team"}), frozenset()) is False
+    assert access_grants(info.access, frozenset(), frozenset({"any-org"})) is False
+
+
+def test_global_access_is_proxy_wide():
+    """access.global=True reaches every identity."""
+    info = CredentialInfo(credential_type="logging", access=_access(global_=True))
+    assert access_grants(info.access, frozenset({"t1"}), frozenset()) is True
+    assert access_grants(info.access, frozenset(), frozenset()) is True
+
+
+def test_access_team_scoped():
+    """access.teams=[t1] fires only for t1 identities."""
+    info = CredentialInfo(credential_type="logging", access=_access(teams=["t1"]))
+    assert access_grants(info.access, frozenset({"t1"}), frozenset()) is True
+    assert access_grants(info.access, frozenset({"t2"}), frozenset()) is False
+    assert access_grants(info.access, frozenset(), frozenset()) is False
+
+
+def test_access_org_scoped():
+    """access.orgs=[o1] fires only for o1 identities."""
+    info = CredentialInfo(credential_type="logging", access=_access(orgs=["o1"]))
+    assert access_grants(info.access, frozenset(), frozenset({"o1"})) is True
+    assert access_grants(info.access, frozenset(), frozenset({"o2"})) is False
+
+
+def test_denies_when_no_access():
+    info = CredentialInfo(credential_type="logging")
+    assert access_grants(info.access, frozenset({"t1"}), frozenset({"o1"})) is False
+
+
+# --- identity_scope --------------------------------------------------------
+
+
+def test_identity_scope_single_elements():
+    teams, orgs = identity_scope("t1", "o1")
+    assert teams == frozenset({"t1"})
+    assert orgs == frozenset({"o1"})
+
+
+def test_identity_scope_empty_for_none():
+    teams, orgs = identity_scope(None, None)
+    assert teams == frozenset()
+    assert orgs == frozenset()
+
+
+# --- resolved_logging_exporter_names: the /team/info + /organization/info disclosure --
+
+
+def _cred(name, access=None, ctype="logging", buildable=True, endpoint=None):
+    info = {"credential_type": ctype}
+    if access is not None:
+        info["access"] = access
+    values = {}
+    if buildable:
+        # a generic OTLP backend with an endpoint builds a destination, so disclosure
+        # (which now mirrors the resolver's buildability) includes it. The endpoint is
+        # per-name by default so these fixtures are distinct destinations; disclosure
+        # dedupes ones that resolve to the same target, which is covered separately.
+        info["description"] = "generic"
+        values = {"otel_endpoint": endpoint or f"http://collector.example/{name}/v1/traces"}
+    return CredentialItem(credential_name=name, credential_values=values, credential_info=info)
+
+
+def test_resolved_names_are_access_only(monkeypatch):
+    """A destination name appears iff its access grants the (team_id, org_id).
+    Included: team-granted, org-granted, global. Excluded: empty-access,
+    granted-but-not-logging (provider) credentials, access for another team."""
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            _cred("team-granted", access={"teams": ["t1"]}),
+            _cred("team-other", access={"teams": ["other"]}),
+            _cred("org-granted", access={"orgs": ["o1"]}),
+            _cred("empty-access"),
+            _cred("global-access", access={"global": True}),
+            _cred("provider", access={"global": True}, ctype=None),
+        ],
+    )
+    names = resolved_logging_exporter_names("t1", "o1")
+    assert names == ("team-granted", "org-granted", "global-access")
+
+
+def test_resolved_names_empty_scope_gets_global_only(monkeypatch):
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            _cred("global-access", access={"global": True}),
+            _cred("team-scoped", access={"teams": ["t1"]}),
+        ],
+    )
+    assert resolved_logging_exporter_names(None, None) == ("global-access",)
+
+
+def test_resolved_names_empty_registry(monkeypatch):
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    monkeypatch.setattr(litellm, "credential_list", [])
+    assert resolved_logging_exporter_names("t1", "o1") == ()
+
+
+def test_resolved_names_gated_off_when_v2_disabled(monkeypatch):
+    """Disclosure mirrors the resolver, which returns nothing with the v2 flag off.
+    A granting destination is disclosed only when ``LITELLM_OTEL_V2`` is enabled, so
+    ``/team/info`` never claims traces are exported while the feature is inert."""
+    monkeypatch.setattr(litellm, "credential_list", [_cred("team-granted", access={"teams": ["t1"]})])
+
+    monkeypatch.setenv("LITELLM_OTEL_V2", "false")
+    is_otel_v2_enabled.cache_clear()
+    assert resolved_logging_exporter_names("t1", None) == ()
+
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    is_otel_v2_enabled.cache_clear()
+    assert resolved_logging_exporter_names("t1", None) == ("team-granted",)
+
+
+def test_resolved_names_excludes_unbuildable(monkeypatch):
+    """Disclosure mirrors the resolver's buildability, not access alone: a granted
+    destination that names no backend, or a backend with incomplete values, resolves to
+    nothing at request time and must not be advertised on /team/info."""
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            _cred("buildable-generic", access={"teams": ["t1"]}),
+            _cred("no-backend", access={"teams": ["t1"]}, buildable=False),
+            CredentialItem(
+                credential_name="langfuse-missing-secret",
+                credential_values={"langfuse_public_key": "pk-only"},
+                credential_info={
+                    "credential_type": "logging",
+                    "access": {"teams": ["t1"]},
+                    "description": "langfuse_otel",
+                },
+            ),
+        ],
+    )
+    assert resolved_logging_exporter_names("t1", None) == ("buildable-generic",)
+
+
+def test_backend_without_a_preset_routes_under_generic():
+    """Regression: a backend outside ``PRESET_BY_CALLBACK`` is routed as ``generic``.
+
+    Only a registered name gets an ``OpenTelemetryV2`` logger, and that logger is what
+    emits the gen-AI span to its destinations. Routing an unregistered name under itself
+    delivered the proxy-internal spans but never the LLM call. Registered names keep
+    their own routing so each backend's attribute vocabulary is preserved."""
+    from litellm.integrations.otel.presets import PRESET_BY_CALLBACK
+
+    unknown = CredentialItem(
+        credential_name="self-hosted",
+        credential_values={"otel_endpoint": "http://collector.example/v1/traces"},
+        credential_info={"credential_type": "logging", "description": "honeycomb", "access": {"global": True}},
+    )
+    resolved = destination_for_credential(unknown)
+    assert resolved is not None
+    assert resolved[0] == "generic"
+    assert resolved[1].endpoint == "http://collector.example/v1/traces"
+
+    for registered in ("arize", "langfuse_otel", "generic"):
+        assert registered in PRESET_BY_CALLBACK
+    known = CredentialItem(
+        credential_name="lf",
+        credential_values={"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
+        credential_info={"credential_type": "logging", "description": "langfuse_otel", "access": {"global": True}},
+    )
+    known_resolved = destination_for_credential(known)
+    assert known_resolved is not None
+    assert known_resolved[0] == "langfuse_otel"
+
+
+def test_resolved_names_keep_every_grant_sharing_one_target(monkeypatch):
+    """Regression: two credentials resolving to one export target are both named.
+
+    Disclosure once kept the first credential per target while the resolver's dict
+    comprehension kept the last, so /team/info named a credential whose backend the
+    request path never activated. Both grant the team, so both are disclosed and there
+    is no winner to disagree about; the resolver still collapses the shared target."""
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    same = "http://collector.example/shared/v1/traces"
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            _cred("dup-one", access={"teams": ["t1"]}, endpoint=same),
+            _cred("dup-two", access={"teams": ["t1"]}, endpoint=same),
+            _cred("distinct", access={"teams": ["t1"]}),
+        ],
+    )
+    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/management_endpoints/test_logging_exporter_validation.py

@@ -0,0 +1,48 @@
+"""Tests for ``validate_credential_access`` -- the shape check on a logging
+destination's ``credential_info.access`` at create/update time.
+
+Which identities a destination fires for is governed entirely by ``access`` and
+evaluated by the request-time resolver; there is no separate assignment/enable
+surface, so this module only guards that a write stores a well-formed ``access``.
+"""
+
+import pytest
+from fastapi import HTTPException
+
+from litellm.proxy.management_endpoints.logging_exporter_validation import (
+    validate_credential_access,
+)
+
+
+def test_validate_credential_access_accepts_valid_object():
+    validate_credential_access({"access": {"global": False, "teams": ["t1", "t2"], "orgs": ["o1"]}})
+
+
+def test_validate_credential_access_noop_without_access():
+    validate_credential_access({"credential_type": "logging"})
+    validate_credential_access(None)
+
+
+@pytest.mark.parametrize(
+    "access",
+    [
+        5,  # not an object
+        {"global": "yes"},  # global must be bool
+        {"teams": "t1"},  # teams must be a list
+        {"orgs": [1, 2]},  # orgs must be strings
+    ],
+)
+def test_validate_credential_access_rejects_bad_shape(access):
+    with pytest.raises(HTTPException) as exc:
+        validate_credential_access({"access": access})
+    assert exc.value.status_code == 400
+
+
+def test_validate_credential_access_rejects_unknown_field():
+    """Unknown access keys must be rejected at write time so a destination can never
+    be stored in a shape the strict ``CredentialAccess`` read model later refuses to
+    parse (which would 500 every subsequent PATCH)."""
+    with pytest.raises(HTTPException) as exc:
+        validate_credential_access({"access": {"global": True, "legacy_field": "x"}})
+    assert exc.value.status_code == 400
+    assert "legacy_field" in exc.value.detail["error"]