Merge origin/litellm_internal_staging into litellm_cli_skip_cost_map_fetch

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

basedpyright-code-budget.json

@@ -105,7 +105,7 @@
     "limit": 109
   },
   "reportUnknownMemberType": {
-    "limit": 38271
+    "limit": 38269
   },
   "reportUnknownParameterType": {
     "limit": 19584

litellm/__init__.py

@@ -546,7 +546,7 @@ _key_management_system: Optional["KeyManagementSystem"] = None
 #### PII MASKING ####
 output_parse_pii: bool = False
 #############################################
-from litellm.litellm_core_utils.get_model_cost_map import get_model_cost_map
+from litellm.litellm_core_utils.get_model_cost_map import get_model_cost_map, mark_litellm_import_complete
 
 model_cost = get_model_cost_map(url=model_cost_map_url)
 cost_discount_config: Dict[str, float] = {}  # Provider-specific cost discounts {"vertex_ai": 0.05} = 5% discount
@@ -2405,3 +2405,5 @@ def __getattr__(name: str) -> Any:
 
 
 # ALL_LITELLM_RESPONSE_TYPES is lazy-loaded via __getattr__ to avoid loading utils at import time
+
+mark_litellm_import_complete()

litellm/constants.py

@@ -143,6 +143,7 @@ DEFAULT_MCP_SEMANTIC_FILTER_SIMILARITY_THRESHOLD: Final = float(
     os.getenv("DEFAULT_MCP_SEMANTIC_FILTER_SIMILARITY_THRESHOLD", 0.3)
 )
 MAX_MCP_SEMANTIC_FILTER_TOOLS_HEADER_LENGTH: Final = int(os.getenv("MAX_MCP_SEMANTIC_FILTER_TOOLS_HEADER_LENGTH", 150))
+MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH: Final = 2048
 
 DEFAULT_AUTO_ROUTER_MAX_INPUT_CHARS: Final = 2000
 
@@ -197,6 +198,7 @@ LITELLM_UI_ALLOW_HEADERS: Final = [
     "x-litellm-adaptive-router-model",
     "x-litellm-applied-guardrails",
     "x-litellm-guardrail-scan-id",
+    "x-litellm-guardrail-scan-metadata",
     "x-litellm-cache-key",
 ]
 

litellm/integrations/custom_guardrail.py

@@ -850,20 +850,24 @@ class CustomGuardrail(CustomLogger):
         if self.should_run_guardrail(data=request_data, event_type=GuardrailEventHooks.post_call) is not True:
             return None
 
-        # CHECK IF GUARDRAIL REJECTS THE REQUEST
         target: Final = self._deployment_hook_target()
-        hook_request_data: Final = {**request_data, "guardrail_to_apply": self} if target is not self else request_data
-        result: Final = await target.async_post_call_success_hook(
-            user_api_key_dict=UserAPIKeyAuth(
-                user_id=request_data.get("user_api_key_user_id"),
-                team_id=request_data.get("user_api_key_team_id"),
-                end_user_id=request_data.get("user_api_key_end_user_id"),
-                api_key=request_data.get("user_api_key_hash"),
-                request_route=request_data.get("user_api_key_request_route"),
-            ),
-            data=hook_request_data,
-            response=response,
-        )
+        try:
+            if target is not self:
+                request_data["guardrail_to_apply"] = self  # rebind-ok: dispatch consumes this key
+            result: Final = await target.async_post_call_success_hook(
+                user_api_key_dict=UserAPIKeyAuth(
+                    user_id=request_data.get("user_api_key_user_id"),
+                    team_id=request_data.get("user_api_key_team_id"),
+                    end_user_id=request_data.get("user_api_key_end_user_id"),
+                    api_key=request_data.get("user_api_key_hash"),
+                    request_route=request_data.get("user_api_key_request_route"),
+                ),
+                data=request_data,
+                response=response,
+            )
+        finally:
+            if target is not self:
+                request_data.pop("guardrail_to_apply", None)
 
         if not self._is_valid_response_type(result):
             return None

litellm/litellm_core_utils/get_model_cost_map.py

@@ -15,6 +15,7 @@ import json
 import os
 import random
 import sys
+import threading
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, replace
@@ -185,6 +186,11 @@ class GetModelCostMap:
 RETRYABLE_FETCH_STATUS_CODES: Final = frozenset({429, 500, 502, 503, 504})
 MODEL_COST_MAP_FETCH_MAX_ATTEMPTS: Final = 3
 MODEL_COST_MAP_FETCH_MAX_WAIT_SECONDS: Final = 30.0
+_litellm_import_complete = threading.Event()
+
+
+def mark_litellm_import_complete() -> None:
+    _litellm_import_complete.set()
 
 
 @dataclass(frozen=True, slots=True)
@@ -323,12 +329,13 @@ async def _fetch_remote_model_cost_map_with_retry(
 def _fetch_remote_model_cost_map_with_retry_sync(
     url: str,
     timeout: int,
-    max_attempts: int,
+    attempts: range,
     sleep: Callable[[float], None],
     rng: random.Random,
     client: _SyncGetClient,
 ) -> ModelCostMapReloadResult:
-    for attempt in range(1, max_attempts + 1):
+    max_attempts: Final = attempts.stop - 1
+    for attempt in attempts:
         outcome = _attempt_fetch_sync(client=client, url=url, timeout=timeout)
         if not isinstance(outcome, _FetchAttemptRetryable):
             return outcome
@@ -529,6 +536,68 @@ def _finalize_loaded_model_cost_map(loaded: ModelCostMapReloaded) -> ModelCostMa
     return replace(loaded, model_cost_map=_finalize_model_cost_map(loaded.model_cost_map))
 
 
+def adopt_model_cost_map(
+    new_model_cost_map: dict,  # mutable-ok: public API preserves the mutable cost-map contract
+) -> int:
+    import litellm
+    from litellm import utils
+
+    litellm.model_cost = new_model_cost_map
+    utils._invalidate_model_cost_lowercase_map()  # pyright: ignore[reportPrivateUsage]  # required cache invalidation
+    litellm.add_known_models(model_cost_map=new_model_cost_map)
+    fetched_model_count: Final = len(new_model_cost_map) if new_model_cost_map else 0
+    utils.reapply_runtime_model_cost_registrations()
+    return fetched_model_count
+
+
+def _retry_remote_fetch_in_background(
+    url: str,
+    timeout: int,
+    max_attempts: int,
+    sleep: Callable[[float], None],
+    rng: random.Random,
+    client: _SyncGetClient,
+    first_outcome: _FetchAttemptRetryable,
+) -> None:
+    try:
+        first_wait: Final = _next_retry_wait(outcome=first_outcome, attempt=1, max_attempts=max_attempts, rng=rng)
+        if isinstance(first_wait, ModelCostMapReloadUnavailable):
+            return
+        sleep(first_wait)
+        result: Final = _fetch_remote_model_cost_map_with_retry_sync(
+            url=url,
+            timeout=timeout,
+            attempts=range(2, max_attempts + 1),
+            sleep=sleep,
+            rng=rng,
+            client=client,
+        )
+        if isinstance(result, ModelCostMapReloadUnavailable):
+            verbose_logger.warning(
+                "LiteLLM: Failed to fetch remote model cost map from %s after %d attempts; keeping local backup",
+                url,
+                max_attempts,
+            )
+            return
+        _litellm_import_complete.wait()
+        if not GetModelCostMap.validate_model_cost_map(
+            fetched_map=result.model_cost_map,
+            backup_model_count=GetModelCostMap._get_backup_model_count(),  # pyright: ignore[reportPrivateUsage]  # integrity cache
+        ):
+            verbose_logger.warning(
+                "LiteLLM: Fetched model cost map failed integrity check. Using local backup instead. url=%s",
+                url,
+            )
+            return
+        finalized: Final = _finalize_loaded_model_cost_map(result).model_cost_map
+        _cost_map_source_info.source = "remote"
+        _cost_map_source_info.fallback_reason = None
+        _cost_map_source_info.loaded_at = datetime.now(timezone.utc)
+        adopt_model_cost_map(finalized)
+    except Exception as e:
+        verbose_logger.warning("LiteLLM: Background model cost map retry failed: %s", e)
+
+
 def get_model_cost_map(
     url: str,
     timeout: int = 5,
@@ -543,8 +612,9 @@ def get_model_cost_map(
     1. If ``LITELLM_LOCAL_MODEL_COST_MAP`` is set or this is a ``lite`` /
        ``litellm-proxy`` CLI process, uses the local backup only.
     2. Otherwise fetches from ``url``, retrying transient HTTP errors
-       (429/5xx/transport) with Retry-After-aware backoff, validates
-       integrity, and falls back to the local backup on any failure.
+       (429/5xx/transport) with Retry-After-aware backoff in a background
+       thread, validates integrity, and falls back to the local backup on any
+       failure.
 
     Only the backup model count is cached (a single int) for validation.
     The full backup dict is only parsed when it must be *returned* as a
@@ -563,24 +633,34 @@ def get_model_cost_map(
     _cost_map_source_info.url = url
     _cost_map_source_info.is_env_forced = False
 
-    result: Final = _fetch_remote_model_cost_map_with_retry_sync(
-        url=url,
-        timeout=timeout,
-        max_attempts=max_attempts,
-        sleep=sleep,
-        rng=rng if rng is not None else random.Random(),
-        client=client if client is not None else httpx,
-    )
-    if isinstance(result, ModelCostMapReloadUnavailable):
+    fetch_client: Final = client if client is not None else httpx
+    fetch_rng: Final = rng if rng is not None else random.Random()
+    outcome: Final = _attempt_fetch_sync(client=fetch_client, url=url, timeout=timeout)
+    if isinstance(outcome, _FetchAttemptRetryable) and max_attempts > 1:
+        threading.Thread(
+            target=_retry_remote_fetch_in_background,
+            kwargs={  # mutable-ok: threading requires a mutable keyword-arguments mapping
+                "url": url,
+                "timeout": timeout,
+                "max_attempts": max_attempts,
+                "sleep": sleep,
+                "rng": fetch_rng,
+                "client": fetch_client,
+                "first_outcome": outcome,
+            },
+            name="litellm-model-cost-map-retry",
+            daemon=True,
+        ).start()
+    if not isinstance(outcome, ModelCostMapReloaded):
         verbose_logger.warning(
             "LiteLLM: Failed to fetch remote model cost map from %s: %s. Falling back to local backup.",
             url,
-            result.reason,
+            outcome.reason,
         )
         _cost_map_source_info.source = "local"
-        _cost_map_source_info.fallback_reason = f"Remote fetch failed: {result.reason}"
+        _cost_map_source_info.fallback_reason = f"Remote fetch failed: {outcome.reason}"
         return _finalize_loaded_model_cost_map(GetModelCostMap.load_local_model_cost_map_with_revision()).model_cost_map
-    content: Final = result.model_cost_map
+    content: Final = outcome.model_cost_map
 
     # Validate using cached count (cheap int comparison, no file I/O)
     if not GetModelCostMap.validate_model_cost_map(
@@ -597,4 +677,4 @@ def get_model_cost_map(
 
     _cost_map_source_info.source = "remote"
     _cost_map_source_info.fallback_reason = None
-    return _finalize_loaded_model_cost_map(result).model_cost_map
+    return _finalize_loaded_model_cost_map(outcome).model_cost_map

litellm/proxy/auth/auth_checks.py

@@ -475,6 +475,7 @@ def _is_model_cost_zero(model: str | list[str] | None, llm_router: Router | None
 
 
 _NO_MODEL_INFO: Final[Mapping[str, object]] = MappingProxyType({})
+_TEAM_GRANT_RELATIONS: Final[Mapping[str, object]] = MappingProxyType({"litellm_model_table": True})
 
 
 def _has_ptu_flat_cost(model: str, llm_router: "Router") -> bool:
@@ -2858,7 +2859,9 @@ class TeamNotFoundError(HTTPException):
 async def _get_team_db_check(
     team_id: str, prisma_client: PrismaClient, team_id_upsert: bool | None = None
 ) -> "_PrismaTeamRow | None":
-    response = await _team_table(TeamRepository(prisma_client)).find_unique(where={"team_id": team_id})
+    response = await _team_table(TeamRepository(prisma_client)).find_unique(
+        where={"team_id": team_id}, include=_TEAM_GRANT_RELATIONS
+    )
 
     if response is None and team_id_upsert:
         from litellm.proxy.management_endpoints.team_endpoints import new_team
@@ -3158,7 +3161,9 @@ async def get_team_object_by_alias(
 
     # Query database by team_alias
     try:
-        teams: Final = await _team_table(TeamRepository(prisma_client)).find_many(where={"team_alias": team_alias})
+        teams: Final = await _team_table(TeamRepository(prisma_client)).find_many(
+            where={"team_alias": team_alias}, include=_TEAM_GRANT_RELATIONS
+        )
 
         if not teams:
             raise HTTPException(

litellm/proxy/auth/handle_jwt.py

@@ -53,6 +53,7 @@ from litellm.proxy._types import (
 )
 from litellm.proxy.auth.auth_checks import can_team_access_model
 from litellm.proxy.auth.route_checks import RouteChecks
+from litellm.proxy.auth.team_grants import team_model_aliases
 from litellm.proxy.common_utils.user_api_key_cache import (
     UserApiKeyCache,
     get_management_object_ttl,
@@ -1595,7 +1596,7 @@ class JWTAuthManager:
                             model=requested_model,
                             team_object=team_object,
                             llm_router=llm_router,
-                            team_model_aliases=None,
+                            team_model_aliases=team_model_aliases(team_object),
                         )
                     ):
                         is_allowed = allowed_routes_check(
@@ -2132,7 +2133,7 @@ class JWTAuthManager:
                         model=requested_model,
                         team_object=team_object,
                         llm_router=llm_router,
-                        team_model_aliases=None,
+                        team_model_aliases=team_model_aliases(team_object),
                     )
                 except ProxyException:
                     continue

litellm/proxy/auth/team_grants.py

@@ -0,0 +1,122 @@
+"""Project a team row (plus the caller's membership in it) onto the ``team_*`` fields of ``UserAPIKeyAuth``.
+
+The virtual-key path gets these fields for free from the combined-view SQL join. Every other auth path
+starts from a ``LiteLLM_TeamTable`` object instead and has to copy them over by hand, which is how JWT
+callers kept losing grants (aliases, permissions, limits) one field at a time. Build the badge through
+``team_grants`` and the two paths cannot drift.
+"""
+
+from collections.abc import Mapping, Sequence
+from types import MappingProxyType
+from typing import Annotated, Final
+
+from pydantic import BaseModel, BeforeValidator, ConfigDict, TypeAdapter, ValidationError
+from pydantic.main import IncEx
+from typing_extensions import ReadOnly, TypedDict
+
+from litellm.proxy._types import (
+    LiteLLM_ObjectPermissionTable,
+    LiteLLM_TeamMembership,
+    LiteLLM_TeamTable,
+    Member,
+)
+
+_MODEL_ALIASES_ADAPTER: Final = TypeAdapter(dict[str, str])
+_JSON_COLUMNS: Final[Mapping[str, IncEx | bool]] = MappingProxyType(
+    {"metadata": True, "litellm_model_table": MappingProxyType({"model_aliases": True})}
+)
+
+
+def _decode_model_aliases(value: object) -> object:
+    """``LiteLLM_ModelTable.model_aliases`` is typed ``str | dict``; writers hand Prisma ``json.dumps(...)``, so take both."""
+    if not isinstance(value, str):
+        return value
+    try:
+        return _MODEL_ALIASES_ADAPTER.validate_json(value)
+    except ValidationError:
+        return None
+
+
+class TeamModelAliasTable(BaseModel):
+    model_config = ConfigDict(protected_namespaces=())
+
+    model_aliases: Annotated[Mapping[str, str] | None, BeforeValidator(_decode_model_aliases)] = None
+
+
+class _TeamJsonColumns(BaseModel):
+    """The two loosely typed columns on ``LiteLLM_TeamTable``, re-read with the shape the badge needs."""
+
+    metadata: Mapping[str, object] | None = None
+    litellm_model_table: TeamModelAliasTable | None = None
+
+
+class TeamGrants(TypedDict, total=False):
+    """Keyword arguments for ``UserAPIKeyAuth``. Empty when the caller has no team, so the model's own defaults apply."""
+
+    team_alias: ReadOnly[str | None]
+    team_tpm_limit: ReadOnly[int | None]
+    team_rpm_limit: ReadOnly[int | None]
+    team_max_budget: ReadOnly[float | None]
+    team_soft_budget: ReadOnly[float | None]
+    team_spend: ReadOnly[float | None]
+    team_models: ReadOnly[Sequence[str]]
+    team_blocked: ReadOnly[bool]
+    team_metadata: ReadOnly[Mapping[str, object] | None]
+    team_model_aliases: ReadOnly[Mapping[str, str] | None]
+    team_object_permission_id: ReadOnly[str | None]
+    team_object_permission: ReadOnly[LiteLLM_ObjectPermissionTable | None]
+    team_member: ReadOnly[Member | None]
+    team_member_spend: ReadOnly[float | None]
+    team_member_tpm_limit: ReadOnly[int | None]
+    team_member_rpm_limit: ReadOnly[int | None]
+
+
+def _json_columns(team_object: LiteLLM_TeamTable) -> _TeamJsonColumns:
+    try:
+        return _TeamJsonColumns.model_validate(team_object.model_dump(include=_JSON_COLUMNS))
+    except ValidationError:
+        return _TeamJsonColumns()
+
+
+def team_model_aliases(team_object: LiteLLM_TeamTable | None) -> Mapping[str, str] | None:
+    if team_object is None:
+        return None
+    alias_table: Final = _json_columns(team_object).litellm_model_table
+    return alias_table.model_aliases if alias_table is not None else None
+
+
+def team_grants(
+    team_object: LiteLLM_TeamTable | None,
+    team_membership: LiteLLM_TeamMembership | None,
+    user_id: str | None,
+) -> TeamGrants:
+    if team_object is None:
+        return TeamGrants()
+    json_columns: Final = _json_columns(team_object)
+    return TeamGrants(
+        team_alias=team_object.team_alias,
+        team_tpm_limit=team_object.tpm_limit,
+        team_rpm_limit=team_object.rpm_limit,
+        team_max_budget=team_object.max_budget,
+        team_soft_budget=team_object.soft_budget,
+        team_spend=team_object.spend,
+        team_models=tuple(team_object.models),
+        team_blocked=team_object.blocked,
+        team_metadata=json_columns.metadata,
+        team_model_aliases=(
+            json_columns.litellm_model_table.model_aliases if json_columns.litellm_model_table is not None else None
+        ),
+        team_object_permission_id=team_object.object_permission_id,
+        team_object_permission=team_object.object_permission,
+        team_member=next(
+            (m for m in team_object.members_with_roles if user_id is not None and m.user_id == user_id),
+            None,
+        ),
+        team_member_spend=team_membership.spend if team_membership is not None else None,
+        team_member_tpm_limit=(
+            team_membership.safe_get_team_member_tpm_limit() if team_membership is not None else None
+        ),
+        team_member_rpm_limit=(
+            team_membership.safe_get_team_member_rpm_limit() if team_membership is not None else None
+        ),
+    )

litellm/proxy/auth/user_api_key_auth.py

@@ -82,6 +82,7 @@ from litellm.proxy.auth.oauth2_proxy_hook import handle_oauth2_proxy_request
 from litellm.proxy.auth.resolvers import CredentialRef, Principal
 from litellm.proxy.auth.resolvers.store import IdentityStore
 from litellm.proxy.auth.route_checks import RouteChecks
+from litellm.proxy.auth.team_grants import team_grants
 from litellm.proxy.auth.trusted_proxy_utils import get_trusted_proxy_cidrs
 from litellm.proxy.common_utils.cache_coordinator import EventDrivenCacheCoordinator
 from litellm.proxy.common_utils.http_parsing_utils import (
@@ -1476,24 +1477,16 @@ async def _user_api_key_auth_builder(
                             user_id=user_id,
                             user_email=user_email,
                             team_id=team_id,
-                            team_alias=(team_object.team_alias if team_object is not None else None),
-                            team_tpm_limit=(team_object.tpm_limit if team_object is not None else None),
-                            team_rpm_limit=(team_object.rpm_limit if team_object is not None else None),
-                            team_models=(team_object.models if team_object is not None else []),
-                            team_metadata=(team_object.metadata if team_object is not None else None),
                             org_id=org_id,
                             end_user_id=end_user_id,
                             parent_otel_span=parent_otel_span,
                             jwt_claims=jwt_claims,
+                            **team_grants(team_object=team_object, team_membership=team_membership, user_id=user_id),
                         )
 
                     valid_token = UserAPIKeyAuth(
                         api_key=None,
                         team_id=team_id,
-                        team_alias=(team_object.team_alias if team_object is not None else None),
-                        team_tpm_limit=(team_object.tpm_limit if team_object is not None else None),
-                        team_rpm_limit=(team_object.rpm_limit if team_object is not None else None),
-                        team_models=(team_object.models if team_object is not None else []),
                         user_role=(
                             LitellmUserRoles(user_object.user_role)
                             if user_object is not None and user_object.user_role is not None
@@ -1507,17 +1500,8 @@ async def _user_api_key_auth_builder(
                         user_tpm_limit=(user_object.tpm_limit if user_object is not None else None),
                         user_rpm_limit=(user_object.rpm_limit if user_object is not None else None),
                         user_model_max_budget=(user_object.model_max_budget if user_object is not None else None),
-                        team_member_rpm_limit=(
-                            team_membership.safe_get_team_member_rpm_limit() if team_membership is not None else None
-                        ),
-                        team_member_tpm_limit=(
-                            team_membership.safe_get_team_member_tpm_limit() if team_membership is not None else None
-                        ),
-                        team_metadata=(team_object.metadata if team_object is not None else None),
                         jwt_claims=jwt_claims,
-                    )
-                    valid_token.team_object_permission = (
-                        team_object.object_permission if team_object is not None else None
+                        **team_grants(team_object=team_object, team_membership=team_membership, user_id=user_id),
                     )
 
                     # AUTO_REGISTER deferred from _resolve_jwt_to_virtual_key.

litellm/proxy/common_utils/callback_utils.py

@@ -1,10 +1,12 @@
 import copy
+import json
 import os
 from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass
+from itertools import accumulate
 from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Optional, TypeAlias
 
-from typing_extensions import assert_never
+from typing_extensions import ReadOnly, TypedDict, assert_never
 
 import litellm
 from litellm import get_secret
@@ -12,6 +14,7 @@ from litellm._logging import verbose_proxy_logger
 from litellm.constants import (
     CLIENT_OUTPUT_CEILING_METADATA_KEY,
     CONSUMED_REQUEST_TAGS_METADATA_KEY,
+    MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH,
     PRE_CALL_EXECUTED_GUARDRAILS_KEY,
     ROUTING_REQUEST_TAGS_METADATA_KEY,
     SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
@@ -28,6 +31,7 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import (
     encrypt_value_helper,
 )
 from litellm.proxy.types_utils.utils import get_instance_fn
+from litellm.types.guardrails import GuardrailEventHooks
 from litellm.types.utils import (
     StandardLoggingGuardrailInformation,
     StandardLoggingPayload,
@@ -52,6 +56,15 @@ reset_color_code: Final = "\033[0m"
 TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY: Final = "_pillar_response_headers_trusted"
 
 GUARDRAIL_SCAN_IDS_METADATA_KEY: Final = "guardrail_scan_ids"
+GUARDRAIL_SCAN_METADATA_METADATA_KEY: Final = "guardrail_scan_metadata"
+
+
+class GuardrailScanMetadata(TypedDict):
+    guardrail: ReadOnly[str | None]
+    stage: ReadOnly[str]
+    provider: ReadOnly[str]
+    scan_id: ReadOnly[str]
+
 
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
@@ -450,6 +463,16 @@ def get_remaining_tokens_and_requests_from_request_data(data: dict) -> dict[str,
     return headers
 
 
+def _serialize_scan_metadata_header(entries: Iterable[object], *, max_length: int) -> str | None:
+    """Compact JSON list of scan metadata entries, dropping trailing entries so the header fits in max_length."""
+    encoded: Final = tuple(json.dumps(entry, separators=(",", ":")) for entry in entries)
+    lengths: Final = tuple(accumulate(len(item) + 1 for item in encoded))
+    kept: Final = sum(1 for length in lengths if length + 1 <= max_length)
+    if kept == 0:
+        return None
+    return f"[{','.join(encoded[:kept])}]"
+
+
 def get_logging_caching_headers(request_data: dict) -> dict | None:
     _metadata: Final[dict] = {}
     metadata_bucket: Final = request_data.get("metadata")
@@ -468,6 +491,15 @@ def get_logging_caching_headers(request_data: dict) -> dict | None:
     if scan_ids:
         headers["x-litellm-guardrail-scan-id"] = ",".join(scan_ids)
 
+    scan_metadata: Final = _metadata.get(GUARDRAIL_SCAN_METADATA_METADATA_KEY)
+    scan_metadata_header: Final = (
+        _serialize_scan_metadata_header(scan_metadata, max_length=MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH)
+        if isinstance(scan_metadata, (list, tuple))
+        else None
+    )
+    if scan_metadata_header:
+        headers["x-litellm-guardrail-scan-metadata"] = scan_metadata_header
+
     if "applied_policies" in _metadata:
         headers["x-litellm-applied-policies"] = ",".join(_metadata["applied_policies"])
 
@@ -501,6 +533,7 @@ LITELLM_PROXY_INTERNAL_METADATA_KEYS: Final = frozenset(
         "applied_policies",
         "applied_guardrails",
         GUARDRAIL_SCAN_IDS_METADATA_KEY,
+        GUARDRAIL_SCAN_METADATA_METADATA_KEY,
         "policy_sources",
         "guardrails",
         "guardrail_config",
@@ -565,21 +598,40 @@ def add_guardrail_to_applied_guardrails_header(request_data: dict, guardrail_nam
         _metadata["applied_guardrails"] = [guardrail_name]
 
 
-def add_guardrail_scan_id(request_data: dict, scan_id: str | None) -> None:
+def add_guardrail_scan_id(
+    request_data: dict[str, object],
+    scan_id: str | None,
+    *,
+    guardrail_name: str | None,
+    provider: str,
+    stage: GuardrailEventHooks,
+) -> None:
     """
-    Record a provider scan id so it can be surfaced to the caller.
+    Record a provider scan id, keyed to the guardrail execution that produced it, so it can be surfaced to the caller.
 
     Guardrails only return scan details to the client when they block, so allowed requests carry no
-    audit trail. Ids recorded here become the x-litellm-guardrail-scan-id response header.
+    audit trail. Ids recorded here become the x-litellm-guardrail-scan-id response header, and the
+    (guardrail, stage, provider, scan_id) entries become the x-litellm-guardrail-scan-metadata header.
     """
     if not scan_id:
         return
     _, _metadata = get_or_create_metadata_bucket(request_data)
     existing: Final = _metadata.get(GUARDRAIL_SCAN_IDS_METADATA_KEY)
-    scan_ids: Final = tuple(existing) if isinstance(existing, (list, tuple)) else ()
+    scan_ids: Final[tuple[object, ...]] = tuple(existing) if isinstance(existing, (list, tuple)) else ()
     if scan_id not in scan_ids:
         _metadata[GUARDRAIL_SCAN_IDS_METADATA_KEY] = (*scan_ids, scan_id)
 
+    entry: Final[GuardrailScanMetadata] = {
+        "guardrail": guardrail_name,
+        "stage": stage.value,
+        "provider": provider,
+        "scan_id": scan_id,
+    }
+    existing_entries: Final = _metadata.get(GUARDRAIL_SCAN_METADATA_METADATA_KEY)
+    entries: Final[tuple[object, ...]] = tuple(existing_entries) if isinstance(existing_entries, (list, tuple)) else ()
+    if entry not in entries:
+        _metadata[GUARDRAIL_SCAN_METADATA_METADATA_KEY] = (*entries, entry)
+
 
 def add_policy_to_applied_policies_header(request_data: dict, policy_name: str | None):
     """

litellm/proxy/guardrails/guardrail_hooks/openai/moderations.py

@@ -17,7 +17,8 @@ from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
     httpxSpecialProvider,
 )
-from litellm.types.guardrails import GuardrailEventHooks
+from litellm.proxy.common_utils.callback_utils import add_guardrail_scan_id
+from litellm.types.guardrails import GuardrailEventHooks, SupportedGuardrailIntegrations
 from litellm.types.utils import (
     GenericGuardrailAPIInputs,
     GuardrailStatus,
@@ -218,6 +219,13 @@ class OpenAIModerationGuardrail(OpenAIGuardrailBase, CustomGuardrail):
             metadata: Final = request_data.get("metadata") or {}
             request_data["metadata"] = metadata
             metadata["_openai_moderation_response"] = moderation_response.model_dump()
+            add_guardrail_scan_id(
+                request_data=request_data,
+                scan_id=moderation_response.id,
+                guardrail_name=self.guardrail_name,
+                provider=SupportedGuardrailIntegrations.OPENAI_MODERATION.value,
+                stage=GuardrailEventHooks.post_call if input_type == "response" else GuardrailEventHooks.pre_call,
+            )
 
         # Check if content is flagged and raise exception if needed
         self._check_moderation_result(moderation_response)

litellm/proxy/guardrails/guardrail_hooks/panw_prisma_airs/panw_prisma_airs.py

@@ -721,10 +721,18 @@ class PanwPrismaAirsHandler(CustomGuardrail):
             }
         }
 
-    def _record_scan_id(self, request_data: dict[str, object], scan_result: Mapping[str, object]) -> None:
+    def _record_scan_id(
+        self, request_data: dict[str, object], scan_result: Mapping[str, object], stage: GuardrailEventHooks
+    ) -> None:
         """Surface the AIRS scan id on the response, so allowed calls are auditable too."""
         scan_id: Final = scan_result.get("scan_id")
-        add_guardrail_scan_id(request_data=request_data, scan_id=str(scan_id) if scan_id else None)
+        add_guardrail_scan_id(
+            request_data=request_data,
+            scan_id=str(scan_id) if scan_id else None,
+            guardrail_name=self.guardrail_name,
+            provider=self._PROVIDER_NAME,
+            stage=stage,
+        )
 
     def _handle_api_error_with_logging(
         self,
@@ -948,7 +956,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
             event_type=GuardrailEventHooks.post_call,
         )
         add_guardrail_to_applied_guardrails_header(request_data=request_data, guardrail_name=self.guardrail_name)
-        self._record_scan_id(request_data, scan_result)
+        self._record_scan_id(request_data, scan_result, GuardrailEventHooks.post_call)
 
     def _check_and_mark_scanned(self, data: dict, scan_type: str) -> bool:
         """
@@ -1078,7 +1086,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
                 duration=(end_time - start_time).total_seconds(),
                 event_type=GuardrailEventHooks.pre_call,
             )
-            self._record_scan_id(data, scan_result)
+            self._record_scan_id(data, scan_result, GuardrailEventHooks.pre_call)
 
             action: Final = scan_result.get("action", "block")
             category: Final = scan_result.get("category", "unknown")
@@ -1199,7 +1207,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
                 duration=(end_time - start_time).total_seconds(),
                 event_type=GuardrailEventHooks.post_call,
             )
-            self._record_scan_id(data, scan_result)
+            self._record_scan_id(data, scan_result, GuardrailEventHooks.post_call)
 
             action: Final = scan_result.get("action", "block")
             category: Final = scan_result.get("category", "unknown")
@@ -1401,7 +1409,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
                     duration=(end_time - start_time).total_seconds(),
                     event_type=GuardrailEventHooks.post_call,
                 )
-                self._record_scan_id(request_data, scan_result)
+                self._record_scan_id(request_data, scan_result, GuardrailEventHooks.post_call)
 
                 # Add guardrail to applied guardrails header for observability
                 add_guardrail_to_applied_guardrails_header(
@@ -1475,7 +1483,11 @@ class PanwPrismaAirsHandler(CustomGuardrail):
                 )
                 continue
 
-            self._record_scan_id(request_data, scan_result)
+            self._record_scan_id(
+                request_data,
+                scan_result,
+                GuardrailEventHooks.post_call if is_response else GuardrailEventHooks.pre_call,
+            )
 
             action = scan_result.get("action", "block")
             masked_args = self._masked_tool_call_arguments(
@@ -1829,7 +1841,11 @@ class PanwPrismaAirsHandler(CustomGuardrail):
                 new_texts.append(text)
                 continue
 
-            self._record_scan_id(request_data, scan_result)
+            self._record_scan_id(
+                request_data,
+                scan_result,
+                GuardrailEventHooks.post_call if is_response else GuardrailEventHooks.pre_call,
+            )
 
             action = scan_result.get("action", "block")
             masked_text = self._get_masked_text(scan_result, is_response=is_response)
@@ -1901,7 +1917,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
                 )
                 # If we reach here, fallback_on_error="allow"
             else:
-                self._record_scan_id(request_data, mcp_scan_result)
+                self._record_scan_id(request_data, mcp_scan_result, GuardrailEventHooks.pre_call)
                 action = mcp_scan_result.get("action", "block")
                 masked_text = self._get_masked_text(mcp_scan_result, is_response=False)
                 if action == "allow":

litellm/proxy/litellm_pre_call_utils.py

@@ -235,6 +235,7 @@ _UNTRUSTED_ROOT_CONTROL_FIELDS: Final = (
     "applied_policies",
     "policy_sources",
     "guardrail_scan_ids",
+    "guardrail_scan_metadata",
     "routing_decision",
     GATEWAY_INJECTED_CACHE_METADATA_KEY,
     "pillar_response_headers",
@@ -291,6 +292,7 @@ _UNTRUSTED_METADATA_CONTROL_FIELDS: Final = (
     "applied_policies",
     "policy_sources",
     "guardrail_scan_ids",
+    "guardrail_scan_metadata",
     "routing_decision",
     GATEWAY_INJECTED_CACHE_METADATA_KEY,
     SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,

litellm/proxy/management_endpoints/team_endpoints.py

@@ -5601,7 +5601,7 @@ async def team_model_add(
     updated_team: Final = await _team_db(prisma_client).update(
         where={"team_id": data.team_id},
         data={"updated_at": datetime.now(timezone.utc)},
-        include={"object_permission": True},
+        include={"litellm_model_table": True, "object_permission": True},
     )
     if updated_team is None:
         raise HTTPException(
@@ -5688,7 +5688,7 @@ async def team_model_delete(
     updated_team: Final = await _team_db(prisma_client).update(
         where={"team_id": data.team_id},
         data={"models": updated_models},
-        include={"object_permission": True},
+        include={"litellm_model_table": True, "object_permission": True},
     )
     if updated_team is None:
         raise HTTPException(

litellm/proxy/management_endpoints/ui_sso.py

@@ -22,7 +22,6 @@ from html import escape
 from types import MappingProxyType
 from typing import (
     TYPE_CHECKING,
-    Annotated,
     Any,
     Final,
     Literal,
@@ -42,7 +41,7 @@ if TYPE_CHECKING:
 import jwt
 from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response, status
 from fastapi.responses import RedirectResponse
-from pydantic import BaseModel, BeforeValidator, ConfigDict, TypeAdapter, ValidationError
+from pydantic import BaseModel, TypeAdapter, ValidationError
 
 import litellm
 from litellm._logging import verbose_proxy_logger
@@ -95,6 +94,7 @@ from litellm.proxy.auth.auth_utils import (
 )
 from litellm.proxy.auth.handle_jwt import JWTHandler
 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
+from litellm.proxy.auth.team_grants import TeamModelAliasTable
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.common_utils.admin_ui_utils import (
     admin_ui_disabled,
@@ -209,31 +209,14 @@ def _team_detail_db(repo: TeamRepository) -> "TableActions[_TeamDetailRow]":
     return repo.table
 
 
-_MODEL_ALIASES_ADAPTER: Final = TypeAdapter(dict[str, str])
 _SSO_TOKEN_CLAIMS_ADAPTER: Final = TypeAdapter(Mapping[str, object])
 
 
-def _decode_model_aliases(value: object) -> object:
-    """``/team/new`` stores team model aliases as a JSON-encoded string in the Json column."""
-    if not isinstance(value, str):
-        return value
-    try:
-        return _MODEL_ALIASES_ADAPTER.validate_json(value)
-    except ValidationError:
-        return None
-
-
-class _TeamModelAliasTable(BaseModel):
-    model_config = ConfigDict(protected_namespaces=())
-
-    model_aliases: Annotated[Mapping[str, str] | None, BeforeValidator(_decode_model_aliases)] = None
-
-
 class _TeamRowGrants(BaseModel):
     team_id: str
     team_alias: str | None = None
     models: tuple[str, ...] = ()
-    litellm_model_table: _TeamModelAliasTable | None = None
+    litellm_model_table: TeamModelAliasTable | None = None
 
 
 class CliSsoTeamDetail(BaseModel):

litellm/proxy/proxy_server.py

@@ -138,11 +138,7 @@ from litellm.types.utils import (
     TextCompletionResponse,
     TokenCountResponse,
 )
-from litellm.utils import (
-    _invalidate_model_cost_lowercase_map,
-    load_credentials_from_list,
-    reapply_runtime_model_cost_registrations,
-)
+from litellm.utils import load_credentials_from_list
 
 if TYPE_CHECKING:
     from aiohttp import ClientSession
@@ -4436,20 +4432,9 @@ def resolve_classifier_plugin(
 
 
 def _swap_in_model_cost_map(new_model_cost_map: dict) -> int:
-    """Adopt a freshly fetched cost map into this process's litellm state, return the model count"""
-    litellm.model_cost = new_model_cost_map
-    # Invalidate case-insensitive lookup map since model_cost was replaced
-    _invalidate_model_cost_lowercase_map()
-    # Repopulate provider model sets (e.g. litellm.anthropic_models) so that
-    # wildcard patterns like "anthropic/*" include any newly added models.
-    litellm.add_known_models(model_cost_map=new_model_cost_map)
-    # Counted before the re-apply below, which writes into this same dict, so the
-    # number reported describes the fetched price data alone.
-    fetched_model_count: Final = len(new_model_cost_map) if new_model_cost_map else 0
-    # The swap discards everything registered at runtime (deployment model_info,
-    # register_model overrides), so put it back on top of the fresh catalog.
-    reapply_runtime_model_cost_registrations()
-    return fetched_model_count
+    from litellm.litellm_core_utils.get_model_cost_map import adopt_model_cost_map
+
+    return adopt_model_cost_map(new_model_cost_map)
 
 
 def should_load_db_object(object_type: str | SupportedDBObjectType) -> bool:

litellm/utils.py

@@ -3079,7 +3079,7 @@ def register_model(
         # Convert stringified numbers to appropriate numeric types
         loaded_model_cost = model_cost
     elif isinstance(model_cost, str):
-        loaded_model_cost = litellm.get_model_cost_map(url=model_cost)
+        loaded_model_cost = litellm.get_model_cost_map(url=model_cost, max_attempts=1)
 
     if persist_across_reloads:
         _registrations: Final[Mapping[str, Mapping[str, object]]] = loaded_model_cost

tests/test_litellm/integrations/test_custom_guardrail.py

@@ -1842,12 +1842,14 @@ class _ApplyStyleGuardrail(CustomGuardrail):
         self.block = block
         self.apply_called = False
         self.seen_texts = None
+        self.seen_request_data = None
 
     async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
         from fastapi import HTTPException
 
         self.apply_called = True
         self.seen_texts = inputs.get("texts")
+        self.seen_request_data = request_data
         if self.block:
             raise HTTPException(status_code=400, detail={"error": "Violated moderation policy"})
         return inputs
@@ -2646,6 +2648,91 @@ class TestCustomGuardrailPostCallSuccessDeploymentHook:
     which starved every later callback in litellm.callbacks (notably the lazily-appended
     VectorStorePreCallHook that attaches provider_specific_fields["search_results"])."""
 
+    @pytest.mark.asyncio
+    async def test_apply_guardrail_retains_request_identity(self) -> None:
+        from litellm.types.guardrails import GuardrailEventHooks
+        from litellm.types.utils import Choices, Message, ModelResponse
+
+        guardrail: Final = _ApplyStyleGuardrail(block=False)
+        guardrail.event_hook = GuardrailEventHooks.post_call
+        request_data: Final = {"guardrails": ["apply-style-guardrail"]}
+        response: Final = ModelResponse(choices=[Choices(message=Message(content="review me"))])
+
+        await guardrail.async_post_call_success_deployment_hook(
+            request_data=request_data, response=response, call_type=CallTypes.acompletion
+        )
+
+        assert guardrail.seen_request_data is request_data
+        assert guardrail.seen_texts == ["review me"]
+        assert "guardrail_to_apply" not in request_data
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("call_type", (None, CallTypes.acompletion))
+    async def test_apply_guardrail_masks_response_and_records_metadata(self, call_type: CallTypes | None) -> None:
+        from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
+            ContentFilterGuardrail,
+        )
+        from litellm.types.guardrails import BlockedWord, ContentFilterAction, GuardrailEventHooks
+        from litellm.types.utils import Choices, Message, ModelResponse
+
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="response-filter",
+            event_hook=GuardrailEventHooks.post_call,
+            blocked_words=[BlockedWord(keyword="secret", action=ContentFilterAction.MASK)],
+        )
+        request_data: Final = {"guardrails": ["response-filter"]}
+        response: Final = ModelResponse(choices=[Choices(message=Message(content="a secret"))])
+
+        result: Final = await guardrail.async_post_call_success_deployment_hook(
+            request_data=request_data, response=response, call_type=call_type
+        )
+
+        assert isinstance(result, ModelResponse)
+        assert result.choices[0].message.content == f"a {guardrail.keyword_redaction_tag}"
+        entries: Final = _guardrail_entries(request_data)
+        assert len(entries) == 1
+        assert entries[0]["guardrail_name"] == "response-filter"
+        assert entries[0]["guardrail_mode"] == "post_call"
+        assert "guardrail_to_apply" not in request_data
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("error_type", (None, RuntimeError, asyncio.CancelledError))
+    async def test_dispatch_cleans_up_request_on_every_exit(self, error_type: type[BaseException] | None) -> None:
+        from contextlib import nullcontext
+
+        from litellm.integrations.custom_logger import CustomLogger
+        from litellm.types.utils import LLMResponseTypes, ModelResponse
+
+        error: Final = error_type("dispatch interrupted") if error_type is not None else None
+
+        class Dispatch(CustomLogger):
+            request_data: dict[str, object] | None = None
+
+            async def async_post_call_success_hook(
+                self, data: dict[str, object], user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes
+            ) -> LLMResponseTypes:
+                self.request_data = data
+                if error is not None:
+                    raise error
+                return response
+
+        dispatch: Final = Dispatch()
+
+        class Guardrail(_ApplyStyleGuardrail):
+            def _deployment_hook_target(self) -> CustomLogger:
+                return dispatch
+
+        guardrail: Final = Guardrail(block=False)
+        guardrail.event_hook = GuardrailEventHooks.post_call
+        request_data: Final = {"guardrails": ["apply-style-guardrail"]}
+        with pytest.raises(error_type) if error_type is not None else nullcontext():
+            await guardrail.async_post_call_success_deployment_hook(
+                request_data=request_data, response=ModelResponse(), call_type=CallTypes.acompletion
+            )
+
+        assert dispatch.request_data is request_data
+        assert "guardrail_to_apply" not in request_data
+
     @pytest.mark.asyncio
     async def test_returns_none_when_request_has_no_guardrails(self):
         from litellm.types.utils import ModelResponse
@@ -2740,4 +2827,5 @@ class TestCustomGuardrailPostCallSuccessDeploymentHook:
 
         assert result is response
         assert response.choices[0].message.content == "filtered response"
-        assert request_data == {"guardrails": ["test-guardrail"]}
+        assert "guardrail_to_apply" not in request_data
+        assert len(_guardrail_entries(request_data)) == 1

tests/test_litellm/litellm_core_utils/test_get_model_cost_map.py

@@ -7,6 +7,7 @@ count actual model entries, not reserved meta keys) and the extraction of the
 import json
 import os
 import sys
+import threading
 
 import pytest
 
@@ -27,9 +28,7 @@ from litellm.litellm_core_utils.get_model_cost_map import (
 
 
 def _load_root_cost_map() -> dict:
-    path = os.path.join(
-        os.path.dirname(__file__), "../../../model_prices_and_context_window.json"
-    )
+    path = os.path.join(os.path.dirname(__file__), "../../../model_prices_and_context_window.json")
     with open(path) as f:
         return json.load(f)
 
@@ -45,9 +44,7 @@ def test_git_blob_id_is_what_git_hash_object_prints():
 
 
 def _make_models(n: int) -> dict:
-    return {
-        f"model-{i}": {"litellm_provider": "openai", "mode": "chat"} for i in range(n)
-    }
+    return {f"model-{i}": {"litellm_provider": "openai", "mode": "chat"} for i in range(n)}
 
 
 def test_count_model_entries_excludes_reserved_keys():
@@ -130,9 +127,7 @@ def test_finalize_pops_key_and_installs_rules():
 def test_finalize_with_no_block_clears_rules():
     previous = list(get_fallback_generalization_rules())
     try:
-        set_fallback_generalizations(
-            [{"name": "stale", "pattern": r"^x", "model_info": {"a": 1}}]
-        )
+        set_fallback_generalizations([{"name": "stale", "pattern": r"^x", "model_info": {"a": 1}}])
         _finalize_model_cost_map(_make_models(2))
         assert match_capability_generalizations("x-1") is None
     finally:
@@ -318,9 +313,7 @@ def test_get_model_cost_map_stamps_loaded_at():
 
     from litellm.litellm_core_utils import get_model_cost_map as module
 
-    client, _calls = _mock_client(
-        [httpx.Response(200, content=_real_map_bytes())], client_cls=httpx.Client
-    )
+    client, _calls = _mock_client([httpx.Response(200, content=_real_map_bytes())], client_cls=httpx.Client)
 
     before = datetime.now(timezone.utc)
     module.get_model_cost_map(url="https://example.invalid/cost_map.json", client=client)
@@ -329,6 +322,7 @@ def test_get_model_cost_map_stamps_loaded_at():
     assert loaded_at is not None
     assert before <= loaded_at <= datetime.now(timezone.utc)
 
+
 # ---------------------------------------------------------------------------
 # refetch_model_cost_map: retry/backoff behavior for runtime reloads
 # ---------------------------------------------------------------------------
@@ -395,9 +389,7 @@ async def test_refetch_retries_429_honoring_retry_after():
         ]
     )
     sleeper = _SleepRecorder()
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=sleeper, rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloaded)
     assert len(result.model_cost_map) > 100
     assert calls["count"] == 3
@@ -409,9 +401,7 @@ async def test_refetch_gives_up_after_max_attempts_with_exponential_backoff():
     """All 429 without Retry-After: exponential backoff waits, then a failure value."""
     client, calls = _mock_client([httpx.Response(429)])
     sleeper = _SleepRecorder()
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=sleeper, rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloadUnavailable)
     assert "429" in result.reason
     assert "after 3 attempts" in result.reason
@@ -431,9 +421,7 @@ async def test_refetch_caps_retry_after_wait():
         ]
     )
     sleeper = _SleepRecorder()
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=sleeper, rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloaded)
     assert sleeper.waits == [30.0]
 
@@ -448,9 +436,7 @@ async def test_refetch_retries_transport_errors():
         ]
     )
     sleeper = _SleepRecorder()
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=sleeper, rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloaded)
     assert calls["count"] == 2
     assert len(sleeper.waits) == 1
@@ -461,9 +447,7 @@ async def test_refetch_non_retryable_status_fails_immediately():
     """A 404 is permanent: one attempt, no sleeps, failure value."""
     client, calls = _mock_client([httpx.Response(404)])
     sleeper = _SleepRecorder()
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=sleeper, rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloadUnavailable)
     assert "404" in result.reason
     assert calls["count"] == 1
@@ -474,9 +458,7 @@ async def test_refetch_non_retryable_status_fails_immediately():
 async def test_refetch_invalid_json_fails_immediately():
     client, calls = _mock_client([httpx.Response(200, content=b"not json")])
     sleeper = _SleepRecorder()
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=sleeper, rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloadUnavailable)
     assert "invalid JSON" in result.reason
     assert calls["count"] == 1
@@ -488,9 +470,7 @@ async def test_refetch_shrunk_map_fails_integrity_not_swapped_in():
     """A drastically shrunk upstream file is rejected instead of being adopted."""
     tiny = json.dumps(_make_models(60)).encode()
     client, _calls = _mock_client([httpx.Response(200, content=tiny)])
-    result = await refetch_model_cost_map(
-        url=_URL, sleep=_SleepRecorder(), rng=random.Random(0), client=client
-    )
+    result = await refetch_model_cost_map(url=_URL, sleep=_SleepRecorder(), rng=random.Random(0), client=client)
     assert isinstance(result, ModelCostMapReloadUnavailable)
     assert "integrity validation" in result.reason
 
@@ -587,69 +567,125 @@ from litellm.litellm_core_utils.get_model_cost_map import (
 class _SyncSleepRecorder:
     """Injected in place of time.sleep so the boot path's waits are asserted without delay."""
 
-    def __init__(self):
+    def __init__(self, block=False):
         self.waits = []
+        self.block = block
+        self.started = threading.Event()
+        self.release = threading.Event()
 
     def __call__(self, seconds: float) -> None:
+        if self.block:
+            self.started.set()
+            self.release.wait(timeout=10)
         self.waits.append(seconds)
 
 
-def test_boot_load_retries_transient_failures_instead_of_falling_back():
-    """A refused connection then a 503 at pod boot used to pin the process to the bundled
-    backup for its lifetime; both are transient and must be retried before giving up."""
+def _retry_threads():
+    return [thread for thread in threading.enumerate() if thread.name == "litellm-model-cost-map-retry"]
+
+
+def test_boot_load_success_does_not_start_background_retry():
+    client, calls = _mock_client([httpx.Response(200, content=_real_map_bytes())], client_cls=httpx.Client)
+    sleeper = _SyncSleepRecorder()
+
+    cost_map = get_model_cost_map(
+        url=_URL,
+        sleep=sleeper,
+        rng=random.Random(0),
+        client=client,
+    )
+    assert calls["count"] == 1
+    assert sleeper.waits == []
+    assert _retry_threads() == []
+    assert cost_map.keys() >= _load_root_cost_map().keys() - {"sample_spec", FALLBACK_GENERALIZATIONS_KEY}
+    assert get_model_cost_map_source_info()["source"] == "remote"
+
+
+def test_boot_load_transient_failure_returns_local_then_background_retry_adopts_remote(monkeypatch):
+    import litellm
+    from litellm import utils as litellm_utils
+    from litellm.litellm_core_utils import get_model_cost_map as module
+
+    original_model_cost = litellm.model_cost
+    monkeypatch.setattr(litellm, "model_cost", dict(original_model_cost))
+    for name, provider_models in tuple(vars(litellm).items()):
+        if name.endswith("_models") and isinstance(provider_models, set):
+            monkeypatch.setattr(litellm, name, set(provider_models))
+    monkeypatch.setattr(litellm, "models_by_provider", dict(litellm.models_by_provider))
+    monkeypatch.setattr(
+        litellm_utils,
+        "_runtime_registered_model_cost",
+        dict(litellm_utils._runtime_registered_model_cost),
+    )
+    source_info = module._cost_map_source_info
+    for name in ("source", "url", "is_env_forced", "fallback_reason", "loaded_at", "source_revision", "etag"):
+        monkeypatch.setattr(source_info, name, getattr(source_info, name))
+
+    remote_map = _load_root_cost_map()
+    remote_map["claude-remote-only-test"] = {"litellm_provider": "anthropic", "mode": "chat"}
     client, calls = _mock_client(
         [
             httpx.ConnectError("connection refused"),
-            httpx.Response(503),
-            httpx.Response(200, content=_real_map_bytes()),
+            httpx.Response(200, content=json.dumps(remote_map).encode()),
         ],
         client_cls=httpx.Client,
     )
-    sleeper = _SyncSleepRecorder()
+    sleeper = _SyncSleepRecorder(block=True)
+    litellm.register_model({"my-runtime-model": {"litellm_provider": "custom", "max_input_tokens": 4321}})
 
-    cost_map = get_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
-
-    assert calls["count"] == 3
-    assert len(sleeper.waits) == 2
-    assert 2.0 <= sleeper.waits[0] < 3.0
-    assert 4.0 <= sleeper.waits[1] < 5.0
-    source = get_model_cost_map_source_info()
-    assert source["source"] == "remote"
-    assert source["fallback_reason"] is None
-    assert cost_map.keys() >= _load_root_cost_map().keys() - {"sample_spec", FALLBACK_GENERALIZATIONS_KEY}
-
-
-def test_boot_load_honors_retry_after_then_falls_back_after_max_attempts():
-    """An outage longer than the retry budget still ends on the bundled backup, and the
-    recorded fallback reason says how many attempts were spent so operators can tell."""
-    client, calls = _mock_client(
-        [httpx.Response(429, headers={"Retry-After": "7"})], client_cls=httpx.Client
+    cost_map = get_model_cost_map(
+        url=_URL,
+        max_attempts=3,
+        sleep=sleeper,
+        rng=random.Random(0),
+        client=client,
     )
-    sleeper = _SyncSleepRecorder()
 
-    cost_map = get_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
-
-    assert calls["count"] == 3
-    assert sleeper.waits == [7.0, 7.0]
-    source = get_model_cost_map_source_info()
-    assert source["source"] == "local"
-    assert "after 3 attempts" in source["fallback_reason"]
-    assert len(cost_map) > 100
+    assert calls["count"] == 1
+    assert sleeper.waits == []
+    assert sleeper.started.wait(timeout=10)
+    threads = _retry_threads()
+    try:
+        assert len(threads) == 1
+        assert "claude-remote-only-test" not in cost_map
+        assert cost_map.keys() == _finalize_model_cost_map(GetModelCostMap.load_local_model_cost_map()).keys()
+        source = get_model_cost_map_source_info()
+        assert source["source"] == "local"
+        assert source["fallback_reason"].startswith("Remote fetch failed:")
+        sleeper.release.set()
+        for thread in threads:
+            thread.join(timeout=10)
+        assert all(not thread.is_alive() for thread in threads)
+        assert sleeper.waits and 2.0 <= sleeper.waits[0] < 3.0
+        assert calls["count"] == 2
+        assert "claude-remote-only-test" in litellm.model_cost
+        assert "claude-remote-only-test" in litellm.anthropic_models
+        assert "my-runtime-model" in litellm.model_cost
+        source = get_model_cost_map_source_info()
+        assert source["source"] == "remote"
+        assert source["fallback_reason"] is None
+    finally:
+        sleeper.release.set()
+        for thread in _retry_threads():
+            thread.join(timeout=10)
 
 
-def test_boot_load_does_not_retry_permanent_failures():
-    """A 404 or a malformed URL cannot heal by waiting: one attempt, no sleeps, backup."""
+def test_boot_load_does_not_retry_non_retryable_failure():
     client, calls = _mock_client([httpx.Response(404)], client_cls=httpx.Client)
     sleeper = _SyncSleepRecorder()
 
-    get_model_cost_map(url=_URL, sleep=sleeper, rng=random.Random(0), client=client)
+    get_model_cost_map(
+        url=_URL,
+        sleep=sleeper,
+        rng=random.Random(0),
+        client=client,
+    )
     assert calls["count"] == 1
     assert sleeper.waits == []
-    assert get_model_cost_map_source_info()["source"] == "local"
-
-    get_model_cost_map(url="not a url", sleep=sleeper, rng=random.Random(0))
-    assert sleeper.waits == []
-    assert get_model_cost_map_source_info()["source"] == "local"
+    assert _retry_threads() == []
+    source = get_model_cost_map_source_info()
+    assert source["source"] == "local"
+    assert source["fallback_reason"] is not None
 
 
 def test_boot_load_respects_local_env_override(monkeypatch):
@@ -702,7 +738,9 @@ def test_boot_load_that_fails_the_integrity_check_reports_the_backup_not_the_rej
     )
     get_model_cost_map(url=_URL, sleep=_SyncSleepRecorder(), rng=random.Random(0), client=remote)
     shrunk_body = b'{"gpt-5.4-mini": {"mode": "chat", "input_cost_per_token": 1e-06, "output_cost_per_token": 2e-06}}'
-    shrunk, _ = _mock_client([httpx.Response(200, headers={"ETag": 'W/"shrunk"'}, content=shrunk_body)], client_cls=httpx.Client)
+    shrunk, _ = _mock_client(
+        [httpx.Response(200, headers={"ETag": 'W/"shrunk"'}, content=shrunk_body)], client_cls=httpx.Client
+    )
 
     get_model_cost_map(url=_URL, sleep=_SyncSleepRecorder(), rng=random.Random(0), client=shrunk)
 

tests/test_litellm/proxy/auth/test_auth_checks.py

@@ -2374,6 +2374,44 @@ def _mock_prisma_for_team_lookup(find_unique):
     return mock_prisma_client
 
 
+_TEAM_ALIAS_TABLE_ROW = {"id": 1, "model_aliases": '{"fast": "gpt-4o"}', "created_by": "admin", "updated_by": "admin"}
+
+
+def _prisma_team_row(include):
+    """Mimics Prisma: the `litellm_model_table` relation rides on the row only when the query `include`s it."""
+    columns = {"team_id": "team-aliases", "team_alias": "aliases", "models": ["gpt-4o"]}
+    row = (
+        {**columns, "litellm_model_table": _TEAM_ALIAS_TABLE_ROW}
+        if (include or {}).get("litellm_model_table")
+        else columns
+    )
+    return SimpleNamespace(dict=lambda: row, model_dump=lambda: row)
+
+
+@pytest.mark.asyncio
+async def test_get_team_object_loads_model_aliases_relation():
+    """LIT-5858: the auth path read teams without `include`ing `litellm_model_table`, so every JWT
+    team came back with `model_aliases=None` and alias requests 403'd."""
+    from litellm.proxy.auth.auth_checks import get_team_object
+    from litellm.proxy.auth.team_grants import team_model_aliases
+
+    async def find_unique(where, include=None):
+        return _prisma_team_row(include)
+
+    mock_cache = MagicMock()
+    mock_cache.async_get_cache = AsyncMock(return_value=None)
+    mock_cache.async_set_cache = AsyncMock()
+
+    team = await get_team_object(
+        team_id="team-aliases",
+        prisma_client=_mock_prisma_for_team_lookup(AsyncMock(side_effect=find_unique)),
+        user_api_key_cache=mock_cache,
+        check_db_only=True,
+    )
+
+    assert team_model_aliases(team) == {"fast": "gpt-4o"}
+
+
 @pytest.mark.asyncio
 async def test_get_team_object_distinguishes_absent_team_from_unreadable_row():
     """A deleted team and a database that would not answer both surface as a 404,
@@ -6195,6 +6233,32 @@ async def test_get_team_object_by_alias_db_fetch_returns_cached_obj():
     assert result.models == ["gpt-4"]
 
 
+@pytest.mark.asyncio
+async def test_get_team_object_by_alias_loads_model_aliases_relation():
+    """LIT-5858: same regression as `test_get_team_object_loads_model_aliases_relation`, for the
+    `team_alias_jwt_field` lookup."""
+    from litellm.proxy.auth.auth_checks import get_team_object_by_alias
+    from litellm.proxy.auth.team_grants import team_model_aliases
+
+    async def find_many(where, include=None):
+        return [_prisma_team_row(include)]
+
+    mock_prisma_client = MagicMock()
+    mock_prisma_client.db.litellm_teamtable.find_many = AsyncMock(side_effect=find_many)
+
+    mock_cache = MagicMock()
+    mock_cache.async_get_cache = AsyncMock(return_value=None)
+    mock_cache.async_set_cache = AsyncMock()
+
+    team = await get_team_object_by_alias(
+        team_alias="aliases",
+        prisma_client=mock_prisma_client,
+        user_api_key_cache=mock_cache,
+    )
+
+    assert team_model_aliases(team) == {"fast": "gpt-4o"}
+
+
 @pytest.mark.asyncio
 async def test_get_org_object_by_alias_db_fetch_returns_validated_org():
     from litellm.proxy._types import LiteLLM_OrganizationTable

tests/test_litellm/proxy/auth/test_handle_jwt.py

@@ -13,6 +13,7 @@ from litellm.proxy._types import (
     DEFAULT_JWKS_STALE_TTL,
     JWTLiteLLMRoleMap,
     LiteLLM_JWTAuth,
+    LiteLLM_ModelTable,
     LiteLLM_TeamMembership,
     LiteLLM_TeamTable,
     LiteLLM_UserTable,
@@ -1255,6 +1256,57 @@ async def test_find_team_with_model_access_model_group(monkeypatch):
     assert team_obj.team_id == "team-1"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "model_aliases",
+    ['{"fast": "gpt-4o"}', {"fast": "gpt-4o"}],
+    ids=["json-string", "dict"],
+)
+async def test_find_team_with_model_access_resolves_team_model_alias(monkeypatch, model_aliases):
+    """LIT-5858: a JWT team that grants `gpt-4o` under the alias `fast` must resolve a request
+    for `fast`. The JWT path used to pass `team_model_aliases=None`, so every alias request 403'd."""
+    import sys
+    import types
+
+    from litellm.caching import DualCache
+    from litellm.proxy.utils import ProxyLogging
+    from litellm.router import Router
+
+    router = Router(model_list=[{"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o"}}])
+    proxy_server_module = types.ModuleType("proxy_server")
+    proxy_server_module.llm_router = router
+    monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
+
+    team = LiteLLM_TeamTable(
+        team_id="team-aliases",
+        models=["gpt-4o"],
+        litellm_model_table=LiteLLM_ModelTable(model_aliases=model_aliases, created_by="admin", updated_by="admin"),
+    )
+
+    async def mock_get_team_object(*args, **kwargs):
+        return team
+
+    monkeypatch.setattr("litellm.proxy.auth.handle_jwt.get_team_object", mock_get_team_object)
+
+    jwt_handler = JWTHandler()
+    jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth()
+    user_api_key_cache = DualCache()
+
+    team_id, team_obj = await JWTAuthManager.find_team_with_model_access(
+        team_ids={"team-aliases"},
+        requested_model="fast",
+        route="/chat/completions",
+        jwt_handler=jwt_handler,
+        prisma_client=None,
+        user_api_key_cache=user_api_key_cache,
+        parent_otel_span=None,
+        proxy_logging_obj=ProxyLogging(user_api_key_cache=user_api_key_cache),
+    )
+
+    assert team_id == "team-aliases"
+    assert team_obj is team
+
+
 @pytest.mark.asyncio
 async def test_find_team_with_model_access_v1_messages_default_routes(monkeypatch):
     """Regression for #31189: a single-team JWT that grants the requested model

tests/test_litellm/proxy/auth/test_team_grants.py

@@ -0,0 +1,129 @@
+import pytest
+
+from litellm.proxy._types import (
+    LiteLLM_BudgetTable,
+    LiteLLM_ObjectPermissionTable,
+    LiteLLM_TeamMembership,
+    LiteLLM_TeamTable,
+    LiteLLM_VerificationTokenView,
+    Member,
+    UserAPIKeyAuth,
+)
+from litellm.models.team import LiteLLM_ModelTable
+from litellm.proxy.auth.team_grants import team_grants, team_model_aliases
+
+TEAM_ID = "team-grants"
+USER_ID = "user-in-team"
+ALIASES = {"fast": "gpt-4o-mini", "smart": "gpt-4o"}
+
+
+def _alias_table(model_aliases) -> LiteLLM_ModelTable:
+    return LiteLLM_ModelTable(model_aliases=model_aliases, created_by="admin", updated_by="admin")
+
+
+def _full_team(model_aliases=ALIASES) -> LiteLLM_TeamTable:
+    return LiteLLM_TeamTable(
+        team_id=TEAM_ID,
+        team_alias="grants-team",
+        tpm_limit=1000,
+        rpm_limit=10,
+        max_budget=50.0,
+        soft_budget=25.0,
+        spend=12.5,
+        models=["gpt-4o", "gpt-4o-mini"],
+        blocked=True,
+        metadata={"tier": "gold"},
+        litellm_model_table=_alias_table(model_aliases),
+        object_permission_id="op-1",
+        object_permission=LiteLLM_ObjectPermissionTable(object_permission_id="op-1", mcp_servers=["mcp-a"]),
+        members_with_roles=[
+            Member(user_id="someone-else", role="user"),
+            Member(user_id=USER_ID, role="admin"),
+        ],
+    )
+
+
+def _membership() -> LiteLLM_TeamMembership:
+    return LiteLLM_TeamMembership(
+        user_id=USER_ID,
+        team_id=TEAM_ID,
+        spend=3.25,
+        litellm_budget_table=LiteLLM_BudgetTable(tpm_limit=500, rpm_limit=5),
+    )
+
+
+def test_team_grants_cover_every_team_field_the_key_path_gets():
+    """Class guard for LIT-5858 and its siblings: every ``team_*`` column the combined-view SQL hands the
+    virtual-key path must come out of the projection too, with the team's actual value, so adding a column
+    to ``LiteLLM_VerificationTokenView`` without teaching ``team_grants`` fails here instead of in prod."""
+    team = _full_team()
+    grants = team_grants(team_object=team, team_membership=_membership(), user_id=USER_ID)
+    token = UserAPIKeyAuth(team_id=TEAM_ID, **grants)
+
+    view_team_fields = {name for name in LiteLLM_VerificationTokenView.model_fields if name.startswith("team_")}
+    assert view_team_fields - {"team_id"} <= set(grants)
+    assert all(grants[name] is not None for name in view_team_fields - {"team_id"})
+
+    assert token.team_alias == "grants-team"
+    assert token.team_tpm_limit == 1000
+    assert token.team_rpm_limit == 10
+    assert token.team_max_budget == 50.0
+    assert token.team_soft_budget == 25.0
+    assert token.team_spend == 12.5
+    assert token.team_models == ["gpt-4o", "gpt-4o-mini"]
+    assert token.team_blocked is True
+    assert token.team_metadata == {"tier": "gold"}
+    assert token.team_model_aliases == ALIASES
+    assert token.team_object_permission_id == "op-1"
+    assert token.team_object_permission is not None
+    assert token.team_object_permission.mcp_servers == ["mcp-a"]
+    assert token.team_member == Member(user_id=USER_ID, role="admin")
+    assert token.team_member_spend == 3.25
+    assert token.team_member_tpm_limit == 500
+    assert token.team_member_rpm_limit == 5
+
+
+def test_team_grants_without_team_leave_token_defaults():
+    token = UserAPIKeyAuth(**team_grants(team_object=None, team_membership=None, user_id=USER_ID))
+    assert token == UserAPIKeyAuth()
+
+
+@pytest.mark.parametrize(
+    "stored_aliases",
+    [ALIASES, '{"fast": "gpt-4o-mini", "smart": "gpt-4o"}'],
+    ids=["json-object", "json-string-as-written-by-team-new"],
+)
+def test_team_model_aliases_decode_both_storage_shapes(stored_aliases):
+    team = _full_team(model_aliases=stored_aliases)
+    assert team_model_aliases(team) == ALIASES
+    assert team_grants(team_object=team, team_membership=None, user_id=None)["team_model_aliases"] == ALIASES
+
+
+@pytest.mark.parametrize("stored_aliases", [None, "not json", '["a", "b"]', {"fast": 3}], ids=str)
+def test_team_model_aliases_treat_unusable_column_as_no_aliases(stored_aliases):
+    team = _full_team(model_aliases=stored_aliases)
+    assert team_model_aliases(team) is None
+    assert team_grants(team_object=team, team_membership=None, user_id=None)["team_model_aliases"] is None
+
+
+def test_team_model_aliases_none_without_relation_loaded():
+    team = _full_team()
+    team.litellm_model_table = None
+    assert team_model_aliases(team) is None
+    assert team_model_aliases(None) is None
+
+
+def test_team_member_is_the_callers_row_only():
+    team = _full_team()
+    assert team_grants(team_object=team, team_membership=None, user_id="someone-else")["team_member"] == Member(
+        user_id="someone-else", role="user"
+    )
+    assert team_grants(team_object=team, team_membership=None, user_id="stranger")["team_member"] is None
+    assert team_grants(team_object=team, team_membership=None, user_id=None)["team_member"] is None
+
+
+def test_membership_limits_absent_without_membership_row():
+    grants = team_grants(team_object=_full_team(), team_membership=None, user_id=USER_ID)
+    assert grants["team_member_spend"] is None
+    assert grants["team_member_tpm_limit"] is None
+    assert grants["team_member_rpm_limit"] is None

tests/test_litellm/proxy/auth/test_user_api_key_auth.py

@@ -7133,3 +7133,119 @@ def test_user_api_key_auth_opens_a_datadog_span_for_accepted_and_rejected_keys(t
     assert report["outcomes"] == ["accepted", "rejected"]
     auth_span = "litellm.proxy.auth.user_api_key_auth.user_api_key_auth"
     assert [span for span in report["spans"] if span == auth_span] == [auth_span, auth_span]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("is_proxy_admin", [False, True], ids=["standard-return", "proxy-admin-return"])
+async def test_jwt_builder_returns_every_team_grant_the_key_path_gets(is_proxy_admin):
+    """LIT-5858: the team-based JWT path hand-built ``UserAPIKeyAuth`` from a short list of team fields, so the
+    team's model aliases (and on the admin return, its object permission) never reached the token and alias
+    requests 403'd. Both returns now go through ``team_grants``; pin the fields that used to be dropped."""
+    import litellm.proxy.proxy_server as _proxy_server_mod
+    from fastapi import Request
+    from starlette.datastructures import URL
+
+    from litellm.models.team import LiteLLM_ModelTable
+    from litellm.proxy._types import (
+        LiteLLM_ObjectPermissionTable,
+        LiteLLM_TeamMembership,
+        LiteLLM_TeamTable,
+        Member,
+    )
+
+    class _AcceptEveryJwt(JWTHandler):
+        def is_jwt(self, token: str) -> bool:
+            return True
+
+    jwt_handler = _AcceptEveryJwt()
+    jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth()
+
+    team = LiteLLM_TeamTable(
+        team_id="team-jwt-aliases",
+        team_alias="jwt-aliases",
+        models=["gpt-4o"],
+        max_budget=40.0,
+        spend=4.0,
+        blocked=False,
+        metadata={"tier": "gold"},
+        litellm_model_table=LiteLLM_ModelTable(
+            model_aliases='{"fast": "gpt-4o"}', created_by="admin", updated_by="admin"
+        ),
+        object_permission_id="op-jwt",
+        object_permission=LiteLLM_ObjectPermissionTable(object_permission_id="op-jwt", mcp_servers=["mcp-a"]),
+        members_with_roles=[Member(user_id="jwt-user", role="admin")],
+    )
+    membership = LiteLLM_TeamMembership(user_id="jwt-user", team_id="team-jwt-aliases", spend=1.5)
+    builder_result = {
+        "is_proxy_admin": is_proxy_admin,
+        "team_object": team,
+        "user_object": None,
+        "end_user_object": None,
+        "org_object": None,
+        "token": "jwt",
+        "team_id": "team-jwt-aliases",
+        "user_id": "jwt-user",
+        "user_email": "jwt-user@example.com",
+        "end_user_id": None,
+        "org_id": None,
+        "team_membership": membership,
+        "jwt_claims": {"sub": "jwt-user"},
+    }
+
+    mock_proxy_logging_obj = MagicMock()
+    mock_proxy_logging_obj.internal_usage_cache = MagicMock()
+    mock_proxy_logging_obj.internal_usage_cache.dual_cache = AsyncMock()
+    mock_proxy_logging_obj.post_call_failure_hook = AsyncMock(return_value=None)
+    attrs = {
+        "prisma_client": MagicMock(),
+        "user_api_key_cache": DualCache(),
+        "proxy_logging_obj": mock_proxy_logging_obj,
+        "master_key": "sk-master-key",
+        "general_settings": {"enable_jwt_auth": True},
+        "llm_model_list": [],
+        "llm_router": None,
+        "open_telemetry_logger": None,
+        "model_max_budget_limiter": MagicMock(),
+        "user_custom_auth": None,
+        "jwt_handler": jwt_handler,
+        "premium_user": True,
+        "litellm_proxy_admin_name": "admin",
+    }
+    originals = {a: getattr(_proxy_server_mod, a, None) for a in attrs}
+    try:
+        for k, v in attrs.items():
+            setattr(_proxy_server_mod, k, v)
+        request = Request(scope={"type": "http", "headers": [], "method": "POST"})
+        request._url = URL(url="/chat/completions")
+        with patch(  # test-quality-ok: auth_builder is the claim-resolution seam; the regression is how its result is projected onto the token
+            "litellm.proxy.auth.user_api_key_auth.JWTAuthManager.auth_builder",
+            new_callable=AsyncMock,
+            return_value=builder_result,
+        ):
+            token = await _user_api_key_auth_builder(
+                request=request,
+                api_key="Bearer header.payload.signature",
+                azure_api_key_header="",
+                anthropic_api_key_header=None,
+                google_ai_studio_api_key_header=None,
+                azure_apim_header=None,
+                request_data={},
+            )
+    finally:
+        for k, v in originals.items():
+            setattr(_proxy_server_mod, k, v)
+
+    assert token.team_id == "team-jwt-aliases"
+    assert token.user_role == (LitellmUserRoles.PROXY_ADMIN if is_proxy_admin else LitellmUserRoles.INTERNAL_USER)
+    assert token.team_model_aliases == {"fast": "gpt-4o"}
+    assert token.team_object_permission is not None
+    assert token.team_object_permission.mcp_servers == ["mcp-a"]
+    assert token.team_object_permission_id == "op-jwt"
+    assert token.team_alias == "jwt-aliases"
+    assert token.team_models == ["gpt-4o"]
+    assert token.team_max_budget == 40.0
+    assert token.team_spend == 4.0
+    assert token.team_metadata == {"tier": "gold"}
+    assert token.team_member == Member(user_id="jwt-user", role="admin")
+    assert token.team_member_spend == 1.5
+    assert token.jwt_claims == {"sub": "jwt-user"}

tests/test_litellm/proxy/common_utils/test_callback_utils.py

@@ -1,30 +1,33 @@
 import copy
+import json
 import sys
 from types import ModuleType, SimpleNamespace
+from typing import Final
+from unittest.mock import patch
 
 import pytest
 
-
+import litellm
+from litellm.caching.caching import DualCache
+from litellm.constants import MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.common_utils.callback_utils import (
+    _serialize_scan_metadata_header,
     add_guardrail_scan_id,
     add_policy_to_applied_policies_header,
     decrypt_callback_vars,
     encrypt_callback_vars,
     get_logging_caching_headers,
-    initialize_callbacks_on_proxy,
     get_remaining_tokens_and_requests_from_request_data,
+    initialize_callbacks_on_proxy,
     normalize_callback_names,
+    process_callback,
     sanitize_openai_provider_metadata,
     strip_callback_config,
 )
-import litellm
-from litellm.caching.caching import DualCache
-from litellm.integrations.custom_logger import CustomLogger
-from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.utils import ProxyLogging
-
-from unittest.mock import patch
-from litellm.proxy.common_utils.callback_utils import process_callback
+from litellm.types.guardrails import GuardrailEventHooks
 
 
 def test_get_remaining_tokens_and_requests_from_request_data():
@@ -189,20 +192,109 @@ def test_get_logging_caching_headers_merges_metadata_and_litellm_metadata():
     assert headers["x-litellm-policy-sources"] == "global-baseline=team_default"
 
 
+def _record(
+    request_data: dict[str, object],
+    scan_id: str | None,
+    guardrail_name: str = "airs",
+    provider: str = "panw_prisma_airs",
+    stage: GuardrailEventHooks = GuardrailEventHooks.pre_call,
+) -> None:
+    add_guardrail_scan_id(
+        request_data=request_data, scan_id=scan_id, guardrail_name=guardrail_name, provider=provider, stage=stage
+    )
+
+
 def test_add_guardrail_scan_id_dedupes_and_becomes_response_header():
     request_data = {"litellm_metadata": {}}
 
-    add_guardrail_scan_id(request_data=request_data, scan_id="scan-1")
-    add_guardrail_scan_id(request_data=request_data, scan_id="scan-1")
-    add_guardrail_scan_id(request_data=request_data, scan_id="scan-2")
-    add_guardrail_scan_id(request_data=request_data, scan_id=None)
+    _record(request_data, "scan-1")
+    _record(request_data, "scan-1")
+    _record(request_data, "scan-2")
+    _record(request_data, None)
 
     assert request_data["litellm_metadata"]["guardrail_scan_ids"] == ("scan-1", "scan-2")
     assert get_logging_caching_headers(request_data)["x-litellm-guardrail-scan-id"] == "scan-1,scan-2"
 
 
-def test_get_logging_caching_headers_omits_scan_id_header_without_scans():
-    assert "x-litellm-guardrail-scan-id" not in get_logging_caching_headers({"litellm_metadata": {}})
+def test_scan_metadata_header_maps_each_id_to_its_guardrail_stage_and_provider():
+    request_data: Final[dict[str, object]] = {"litellm_metadata": {}}
+
+    _record(
+        request_data, "scan-1", guardrail_name="airs", provider="panw_prisma_airs", stage=GuardrailEventHooks.pre_call
+    )
+    _record(
+        request_data, "mod-1", guardrail_name="mod", provider="openai_moderation", stage=GuardrailEventHooks.pre_call
+    )
+    _record(
+        request_data, "scan-2", guardrail_name="airs", provider="panw_prisma_airs", stage=GuardrailEventHooks.post_call
+    )
+    _record(
+        request_data, "scan-2", guardrail_name="airs", provider="panw_prisma_airs", stage=GuardrailEventHooks.post_call
+    )
+    _record(request_data, None, guardrail_name="mod", provider="openai_moderation", stage=GuardrailEventHooks.post_call)
+
+    headers: Final = get_logging_caching_headers(request_data)
+    assert headers is not None
+    assert headers["x-litellm-guardrail-scan-id"] == "scan-1,mod-1,scan-2"
+    assert json.loads(headers["x-litellm-guardrail-scan-metadata"]) == [
+        {"guardrail": "airs", "stage": "pre_call", "provider": "panw_prisma_airs", "scan_id": "scan-1"},
+        {"guardrail": "mod", "stage": "pre_call", "provider": "openai_moderation", "scan_id": "mod-1"},
+        {"guardrail": "airs", "stage": "post_call", "provider": "panw_prisma_airs", "scan_id": "scan-2"},
+    ]
+
+
+def test_scan_metadata_keeps_same_id_reused_across_stages():
+    request_data: Final[dict[str, object]] = {"metadata": {}}
+
+    _record(request_data, "scan-1", stage=GuardrailEventHooks.pre_call)
+    _record(request_data, "scan-1", stage=GuardrailEventHooks.post_call)
+
+    headers: Final = get_logging_caching_headers(request_data)
+    assert headers is not None
+    assert headers["x-litellm-guardrail-scan-id"] == "scan-1"
+    assert [entry["stage"] for entry in json.loads(headers["x-litellm-guardrail-scan-metadata"])] == [
+        "pre_call",
+        "post_call",
+    ]
+
+
+def test_scan_metadata_header_drops_trailing_entries_to_stay_within_length_limit():
+    request_data: Final[dict[str, object]] = {"litellm_metadata": {}}
+    scan_ids: Final = tuple(f"0f9c4b7e-3d2a-4c1b-9e8f-{index:012d}" for index in range(40))
+    for scan_id in scan_ids:
+        _record(request_data, scan_id, stage=GuardrailEventHooks.post_call)
+
+    headers: Final = get_logging_caching_headers(request_data)
+    assert headers is not None
+    assert headers["x-litellm-guardrail-scan-id"] == ",".join(scan_ids)
+    header: Final = headers["x-litellm-guardrail-scan-metadata"]
+    assert len(header) <= MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH
+    kept: Final = json.loads(header)
+    assert 1 < len(kept) < len(scan_ids)
+    assert [entry["scan_id"] for entry in kept] == list(scan_ids[: len(kept)])
+
+
+def test_serialize_scan_metadata_header_keeps_exactly_the_entries_that_fit():
+    entries: Final = ({"scan_id": "a"}, {"scan_id": "b"}, {"scan_id": "c"})
+    two_entries: Final = '[{"scan_id":"a"},{"scan_id":"b"}]'
+
+    assert _serialize_scan_metadata_header(entries, max_length=len(two_entries)) == two_entries
+    assert _serialize_scan_metadata_header(entries, max_length=len(two_entries) - 1) == '[{"scan_id":"a"}]'
+    assert _serialize_scan_metadata_header(entries, max_length=len(two_entries) + 1) == two_entries
+    assert _serialize_scan_metadata_header(entries, max_length=1000) == json.dumps(entries, separators=(",", ":"))
+    assert _serialize_scan_metadata_header(entries, max_length=5) is None
+    assert _serialize_scan_metadata_header((), max_length=1000) is None
+
+
+def test_scan_metadata_is_an_internal_metadata_key():
+    assert sanitize_openai_provider_metadata({"guardrail_scan_metadata": "x", "keep": "y"}) == {"keep": "y"}
+
+
+def test_get_logging_caching_headers_omits_scan_headers_without_scans():
+    headers: Final = get_logging_caching_headers({"litellm_metadata": {}})
+    assert headers is not None
+    assert "x-litellm-guardrail-scan-id" not in headers
+    assert "x-litellm-guardrail-scan-metadata" not in headers
 
 
 def test_initialize_callbacks_on_proxy_instantiates_compression_interception(

tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_moderations.py

@@ -3,14 +3,19 @@
 Test OpenAI Moderation Guardrail
 """
 
+import json
 import os
+from typing import Final
 
 
 from unittest.mock import MagicMock, patch
 
+import httpx
 import pytest
 
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.common_utils.callback_utils import get_logging_caching_headers
 from litellm.proxy.guardrails.guardrail_hooks.openai.moderations import (
     OpenAIModerationGuardrail,
 )
@@ -989,3 +994,30 @@ async def test_openai_moderation_initialize_guardrail_forwards_streaming_flags()
             assert guardrail.streaming_sampling_rate == 2
     finally:
         litellm.logging_callback_manager._reset_all_callbacks()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("input_type", "stage"), [("request", "pre_call"), ("response", "post_call")])
+async def test_openai_moderation_records_moderation_id_as_scan_metadata(input_type: str, stage: str):
+    """Each moderation call's id is exposed with the guardrail name, stage and provider that produced it."""
+    payload: Final = {
+        "id": f"modr-{stage}",
+        "model": "omni-moderation-latest",
+        "results": [{"flagged": False, "categories": {}, "category_scores": {}, "category_applied_input_types": {}}],
+    }
+    http_client: Final = AsyncHTTPHandler()
+    http_client.client = httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200, json=payload)))
+
+    with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
+        guardrail: Final = OpenAIModerationGuardrail(guardrail_name="openai-mod")
+    guardrail.async_handler = http_client
+    request_data: Final[dict[str, object]] = {"metadata": {}}
+
+    await guardrail.apply_guardrail(inputs={"texts": ["hello"]}, request_data=request_data, input_type=input_type)
+
+    headers: Final = get_logging_caching_headers(request_data)
+    assert headers is not None
+    assert headers["x-litellm-guardrail-scan-id"] == f"modr-{stage}"
+    assert json.loads(headers["x-litellm-guardrail-scan-metadata"]) == [
+        {"guardrail": "openai-mod", "stage": stage, "provider": "openai_moderation", "scan_id": f"modr-{stage}"}
+    ]

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_panw_prisma_airs.py

@@ -12,6 +12,7 @@ This test file follows LiteLLM's testing patterns and covers:
 import copy
 import json
 from datetime import datetime
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -5785,7 +5786,14 @@ class TestPanwAirsScanIdExposure:
 
         headers = get_logging_caching_headers(data)
         assert headers["x-litellm-guardrail-scan-id"] == "scan-abc-123"
-        assert "x-litellm-guardrail-scan-metadata" not in headers
+        assert json.loads(headers["x-litellm-guardrail-scan-metadata"]) == [
+            {
+                "guardrail": handler.guardrail_name,
+                "stage": "pre_call",
+                "provider": "panw_prisma_airs",
+                "scan_id": "scan-abc-123",
+            }
+        ]
 
     @pytest.mark.asyncio
     async def test_request_and_response_scan_ids_are_both_exposed(self, user_api_key_dict):
@@ -5809,6 +5817,26 @@ class TestPanwAirsScanIdExposure:
 
         headers = get_logging_caching_headers(data)
         assert headers["x-litellm-guardrail-scan-id"] == "scan-abc-123,scan-response-456"
+        assert [(e["stage"], e["scan_id"]) for e in json.loads(headers["x-litellm-guardrail-scan-metadata"])] == [
+            ("pre_call", "scan-abc-123"),
+            ("post_call", "scan-response-456"),
+        ]
+
+    @pytest.mark.asyncio
+    async def test_apply_guardrail_response_scan_is_tagged_post_call(self):
+        from litellm.proxy.common_utils.callback_utils import get_logging_caching_headers
+
+        handler: Final = self._handler(self.ALLOW_SCAN_RESULT)
+        request_data: Final[dict[str, object]] = {"litellm_call_id": "test-call-id", "model": "gpt-4", "metadata": {}}
+
+        await handler.apply_guardrail(
+            inputs={"texts": ["Hello world"]}, request_data=request_data, input_type="response"
+        )
+
+        headers: Final = get_logging_caching_headers(request_data)
+        assert headers is not None
+        entries: Final = json.loads(headers["x-litellm-guardrail-scan-metadata"])
+        assert [(e["stage"], e["provider"]) for e in entries] == [("post_call", "panw_prisma_airs")]
 
     @pytest.mark.asyncio
     async def test_repeated_scan_id_is_not_duplicated(self, user_api_key_dict):
@@ -5850,6 +5878,8 @@ class TestPanwAirsScanIdExposure:
 
         assert "guardrail_scan_ids" in _UNTRUSTED_METADATA_CONTROL_FIELDS
         assert "guardrail_scan_ids" in _UNTRUSTED_ROOT_CONTROL_FIELDS
+        assert "guardrail_scan_metadata" in _UNTRUSTED_METADATA_CONTROL_FIELDS
+        assert "guardrail_scan_metadata" in _UNTRUSTED_ROOT_CONTROL_FIELDS
 class TestPanwAirsBlockedErrorDetailPassthrough:
     """Regression tests for the full AIRS scan response on blocks.
 

tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py

@@ -2205,6 +2205,50 @@ async def test_team_model_add_delete_refresh_team_cache(endpoint_name):
         assert update_call_kwargs.get("include", {}).get("object_permission") is True
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("endpoint_name", ["team_model_add", "team_model_delete"])
+async def test_team_model_add_delete_keep_model_aliases_in_team_cache(endpoint_name, monkeypatch):
+    """LIT-5858: Prisma only returns `litellm_model_table` when the `update` asks for it, so the refreshed
+    cache entry lost the team's model aliases and JWT alias requests 403'd until the next DB read."""
+    from litellm.proxy._types import TeamModelAddRequest, TeamModelDeleteRequest
+    from litellm.proxy.auth.team_grants import team_model_aliases
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+    from litellm.proxy.management_endpoints.team_endpoints import team_model_add, team_model_delete
+
+    columns = {"team_id": "team-1234", "models": ["gpt-4o", "openai/*"]}
+    alias_table = {"id": 1, "model_aliases": '{"fast": "gpt-4o"}', "created_by": "admin", "updated_by": "admin"}
+
+    async def update(where, data, include=None):
+        row = {**columns, "litellm_model_table": alias_table} if (include or {}).get("litellm_model_table") else columns
+        return SimpleNamespace(team_id="team-1234", model_dump=lambda: row)
+
+    prisma_client = MagicMock()
+    prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=SimpleNamespace(model_dump=lambda: columns))
+    prisma_client.db.litellm_teamtable.update = AsyncMock(side_effect=update)
+    prisma_client.db.execute_raw = AsyncMock(return_value=None)
+    cache = UserApiKeyCache()
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma_client)
+    monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", cache)
+    monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", None)
+
+    admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin")
+    if endpoint_name == "team_model_add":
+        await team_model_add(
+            data=TeamModelAddRequest(team_id="team-1234", models=["team-byok-1"]),
+            http_request=MagicMock(),
+            user_api_key_dict=admin,
+        )
+    else:
+        await team_model_delete(
+            data=TeamModelDeleteRequest(team_id="team-1234", models=["openai/*"]),
+            http_request=MagicMock(),
+            user_api_key_dict=admin,
+        )
+
+    cached_team = await cache.async_get_cache(key="team_id:team-1234", model_type=LiteLLM_TeamTableCachedObj)
+    assert team_model_aliases(cached_team) == {"fast": "gpt-4o"}
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "endpoint_name",

tests/test_litellm/test_utils.py

@@ -2,10 +2,12 @@ import asyncio
 import json
 import logging
 import os
+import threading
 from datetime import datetime, timedelta, timezone
 from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import pytest
 import respx
 from jsonschema import validate
@@ -2382,6 +2384,28 @@ def test_register_model_with_scientific_notation():
     _invalidate_model_cost_lowercase_map()
 
 
+@respx.mock
+def test_register_model_url_fetch_uses_single_attempt(monkeypatch):
+    monkeypatch.delenv("LITELLM_LOCAL_MODEL_COST_MAP", raising=False)
+    monkeypatch.setattr(litellm, "model_cost", dict(litellm.model_cost))
+    before = dict(litellm.model_cost)
+    threads_before = {thread.name for thread in threading.enumerate()}
+    route = respx.get("https://example.invalid/custom_pricing.json").mock(
+        return_value=httpx.Response(503)
+    )
+
+    litellm.register_model(model_cost="https://example.invalid/custom_pricing.json")
+
+    threads_after = {thread.name for thread in threading.enumerate()}
+    assert route.call_count == 1
+    assert not (threads_after - threads_before) & {"litellm-model-cost-map-retry"}
+    assert not any(
+        thread.name == "litellm-model-cost-map-retry" and thread.is_alive()
+        for thread in threading.enumerate()
+    )
+    assert litellm.model_cost.keys() >= before.keys()
+
+
 def test_register_model_openrouter_without_slash():
     """
     Test that register_model handles openrouter models without '/' in the name.