Merge pull request #26900 from BerriAI/litellm_model-deprecation-alerts-55bc

feat(proxy): proactive model deprecation alerts and `/model/deprecations` endpoint

backend/routes/allowlist.py

@@ -35,6 +35,7 @@ BACKEND_PATH_PREFIXES: tuple[str, ...] = (
     # Models & routing config
     "/model/",
     "/v1/model/info",
+    "/v1/model/deprecations",
     "/v2/model/",
     "/model_group",
     "/model_access_group/",

litellm/constants.py

@@ -1487,6 +1487,7 @@ WEEKLY_SPEND_REPORT_JOB_ID: Final = "weekly_spend_report_job"
 MONTHLY_SPEND_REPORT_JOB_ID: Final = "monthly_spend_report_job"
 PROMETHEUS_FALLBACK_STATS_JOB_ID: Final = "prometheus_fallback_stats_job"
 SLACK_DAILY_REPORT_LOCK_ID: Final = "slack_daily_report"
+SLACK_MODEL_DEPRECATION_LOCK_ID: Final = "slack_model_deprecation_warning"
 SPEND_LOG_RUN_LOOPS: Final = int(os.getenv("SPEND_LOG_RUN_LOOPS", 500))
 SPEND_LOG_CLEANUP_BATCH_SIZE: Final = int(os.getenv("SPEND_LOG_CLEANUP_BATCH_SIZE", 1000))
 SPEND_LOG_CLEANUP_MAX_CONSECUTIVE_BATCH_FAILURES = int(os.getenv("SPEND_LOG_CLEANUP_MAX_CONSECUTIVE_BATCH_FAILURES", 3))

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -5,6 +5,7 @@ import datetime
 import os
 import random
 import time
+from collections.abc import Callable
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any, Final, Literal
 
@@ -17,7 +18,11 @@ import litellm.litellm_core_utils.litellm_logging
 import litellm.types
 from litellm._logging import verbose_logger, verbose_proxy_logger
 from litellm.caching.caching import DualCache
-from litellm.constants import HOURS_IN_A_DAY, SLACK_DAILY_REPORT_LOCK_ID
+from litellm.constants import (
+    HOURS_IN_A_DAY,
+    SLACK_DAILY_REPORT_LOCK_ID,
+    SLACK_MODEL_DEPRECATION_LOCK_ID,
+)
 from litellm.integrations.custom_batch_logger import CustomBatchLogger
 from litellm.integrations.SlackAlerting.budget_alert_types import get_budget_alert_type
 from litellm.integrations.SlackAlerting.hanging_request_check import (
@@ -45,6 +50,10 @@ from litellm.repositories.table_repositories import InvitationLinkRepository
 from litellm.repositories.team_repository import TeamRepository
 from litellm.repositories.user_repository import UserRepository
 from litellm.types.integrations.slack_alerting import *
+from litellm.types.proxy.model_deprecation import (
+    DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+    DEPRECATION_IDLE_POLL_SECONDS,
+)
 
 from ..email_templates.templates import *
 from .batching_handler import send_to_webhook, squash_payloads
@@ -59,6 +68,12 @@ else:
     Router = Any
 
 
+def _proxy_llm_router() -> Router | None:
+    from litellm.proxy.proxy_server import llm_router
+
+    return llm_router
+
+
 class SlackAlerting(CustomBatchLogger):
     """
     Class for sending Slack Alerts
@@ -1044,6 +1059,99 @@ Model Info:
     async def model_removed_alert(self, model_name: str):
         pass
 
+    def _deprecation_alerts_enabled(self) -> bool:
+        return self.alerting is not None and AlertType.model_deprecation_warnings in self.alert_types
+
+    async def send_model_deprecation_alert(
+        self,
+        llm_router: Router | None = None,
+        pod_lock_manager: "PodLockManager | None" = None,
+    ) -> bool:
+        """Alert on the router's deprecated and imminent models, True when one was sent
+
+        The daily lock is claimed only once there is something to say, so an empty pass never blocks a
+        later real one, and a sent alert is stamped in the shared cache for a day so sibling pods stop asking
+        """
+        if not self._deprecation_alerts_enabled():
+            return False
+
+        from litellm.proxy.common_utils.model_deprecation import (
+            collect_model_deprecations,
+            format_deprecation_alert_message,
+        )
+
+        snapshot: Final = collect_model_deprecations(llm_router=llm_router)
+        message: Final = format_deprecation_alert_message(snapshot)
+        if message is None:
+            return False
+        if not await self._claimed_deprecation_alert_window(pod_lock_manager):
+            return False
+
+        level: Final[Literal["Low", "Medium", "High"]] = "High" if snapshot.deprecated else "Medium"
+
+        await self.send_alert(
+            message=message,
+            level=level,
+            alert_type=AlertType.model_deprecation_warnings,
+            alerting_metadata={  # mutable-ok: send_alert takes a dict payload
+                "deprecated_count": len(snapshot.deprecated),
+                "imminent_count": len(snapshot.imminent),
+                "upcoming_count": len(snapshot.upcoming),
+            },
+        )
+        await self.internal_usage_cache.async_set_cache(
+            key=SlackAlertingCacheKeys.deprecation_alert_sent_key.value,
+            value=time.time(),
+            ttl=DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+        )
+        return True
+
+    async def _claimed_deprecation_alert_window(self, pod_lock_manager: "PodLockManager | None") -> bool:
+        """Without a redis backed lock there is no fleet to coordinate, so a lone pod always alerts"""
+        if pod_lock_manager is None:
+            return True
+        return (
+            await pod_lock_manager.acquire_lock(
+                cronjob_id=SLACK_MODEL_DEPRECATION_LOCK_ID,
+                ttl=DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+                allow_reentrant=False,
+            )
+        ) is not False
+
+    async def _deprecation_alert_sent_within_a_day(self) -> bool:
+        return (
+            await self.internal_usage_cache.async_get_cache(key=SlackAlertingCacheKeys.deprecation_alert_sent_key.value)
+        ) is not None
+
+    async def _run_deprecation_alert_pass(
+        self, llm_router: Router | None, pod_lock_manager: "PodLockManager | None"
+    ) -> bool:
+        if llm_router is None or not self._deprecation_alerts_enabled():
+            return False
+        if await self._deprecation_alert_sent_within_a_day():
+            return False
+        return await self.send_model_deprecation_alert(llm_router=llm_router, pod_lock_manager=pod_lock_manager)
+
+    async def run_scheduled_deprecation_check(
+        self,
+        get_llm_router: Callable[[], Router | None] = _proxy_llm_router,
+        pod_lock_manager: "PodLockManager | None" = None,
+    ) -> None:
+        """Poll every pass for a loaded router, the alert being on, and no alert in the last day, then alert
+
+        A pass that could not alert (no router yet, alert type off, a sibling pod holds the daily lock, or a
+        redis blip at claim time) is retried on the next poll instead of costing a day, while a pass that
+        raised (a missing webhook, say) backs off a full day so a misconfiguration logs once, not every poll
+        """
+        while True:
+            try:
+                await self._run_deprecation_alert_pass(get_llm_router(), pod_lock_manager)
+            except Exception as e:  # noqa: BLE001  # a failed alert must not kill the loop
+                verbose_proxy_logger.exception("Error in model deprecation alert loop: %s", e)
+                await asyncio.sleep(DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS)
+                continue
+            await asyncio.sleep(DEPRECATION_IDLE_POLL_SECONDS)
+
     async def send_webhook_alert(self, webhook_event: WebhookEvent) -> bool:
         """
         Sends structured alert to webhook, if set.

litellm/proxy/common_utils/model_deprecation.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+from itertools import groupby
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final
+
+import litellm
+from litellm._logging import verbose_logger
+from litellm.types.proxy.model_deprecation import (
+    DEFAULT_DEPRECATION_WARN_DAYS,
+    DeprecationStatus,
+    ModelDeprecationInfo,
+    ModelDeprecationResponse,
+)
+
+if TYPE_CHECKING:
+    from litellm.router import Router
+
+_NO_MODEL_METADATA: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+@dataclass(frozen=True, slots=True)
+class _ResolvedDeprecation:
+    deprecation_date: date
+    litellm_model: str | None
+    litellm_provider: str | None
+
+
+def _parse_deprecation_date(raw_value: object) -> date | None:
+    if isinstance(raw_value, datetime):
+        return raw_value.date()
+    if isinstance(raw_value, date):
+        return raw_value
+    if not isinstance(raw_value, str):
+        return None
+    try:
+        return date.fromisoformat(raw_value.strip())
+    except ValueError:
+        return None
+
+
+def _cost_map_lookup(model_key: object) -> _ResolvedDeprecation | None:
+    if not isinstance(model_key, str) or not model_key:
+        return None
+    entry: Final = litellm.model_cost.get(model_key)
+    if not isinstance(entry, Mapping):
+        return None
+    parsed: Final = _parse_deprecation_date(entry.get("deprecation_date"))
+    if parsed is None:
+        return None
+    provider: Final = entry.get("litellm_provider")
+    return _ResolvedDeprecation(
+        deprecation_date=parsed,
+        litellm_model=model_key,
+        litellm_provider=provider if isinstance(provider, str) else None,
+    )
+
+
+def _mapping_field(deployment: Mapping[str, object], key: str) -> Mapping[str, object]:
+    value: Final = deployment.get(key)
+    return value if isinstance(value, Mapping) else _NO_MODEL_METADATA
+
+
+def _resolve_deployment_deprecation(
+    deployment: Mapping[str, object],
+) -> _ResolvedDeprecation | None:
+    """Resolve a deployment's deprecation date, preferring its explicit override"""
+    model_info: Final = _mapping_field(deployment, "model_info")
+    raw_model: Final = _mapping_field(deployment, "litellm_params").get("model")
+
+    override: Final = _parse_deprecation_date(model_info.get("deprecation_date"))
+    if override is not None:
+        provider: Final = model_info.get("litellm_provider")
+        return _ResolvedDeprecation(
+            deprecation_date=override,
+            litellm_model=raw_model if isinstance(raw_model, str) else None,
+            litellm_provider=provider if isinstance(provider, str) else None,
+        )
+
+    unprefixed: Final = raw_model.split("/", 1)[1] if isinstance(raw_model, str) and "/" in raw_model else None
+    return next(
+        (
+            resolved
+            for resolved in (
+                _cost_map_lookup(model_info.get("base_model")),
+                _cost_map_lookup(raw_model),
+                _cost_map_lookup(unprefixed),
+            )
+            if resolved is not None
+        ),
+        None,
+    )
+
+
+def _classify(days_until: int, warn_within_days: int) -> DeprecationStatus:
+    if days_until < 0:
+        return "deprecated"
+    if days_until <= warn_within_days:
+        return "imminent"
+    return "upcoming"
+
+
+def _build_info(deployment: Mapping[str, object], today: date, warn_within_days: int) -> ModelDeprecationInfo | None:
+    model_name: Final = deployment.get("model_name")
+    if not isinstance(model_name, str) or not model_name:
+        return None
+
+    resolved: Final = _resolve_deployment_deprecation(deployment)
+    if resolved is None:
+        return None
+
+    days_until: Final = (resolved.deprecation_date - today).days
+    return ModelDeprecationInfo(
+        model_name=model_name,
+        litellm_model=resolved.litellm_model,
+        deprecation_date=resolved.deprecation_date,
+        days_until_deprecation=days_until,
+        status=_classify(days_until, warn_within_days),
+        litellm_provider=resolved.litellm_provider,
+    )
+
+
+def _dedupe(
+    models: Sequence[ModelDeprecationInfo],
+) -> tuple[ModelDeprecationInfo, ...]:
+    """Report a model group carrying the same date on several deployments once"""
+    ordered: Final = sorted(models, key=lambda model: (model.model_name, model.deprecation_date))
+    return tuple(
+        next(group) for _, group in groupby(ordered, key=lambda model: (model.model_name, model.deprecation_date))
+    )
+
+
+def _bucket(models: Sequence[ModelDeprecationInfo], status: DeprecationStatus) -> tuple[ModelDeprecationInfo, ...]:
+    return tuple(
+        sorted(
+            (model for model in models if model.status == status),
+            key=lambda model: model.deprecation_date,
+        )
+    )
+
+
+def collect_model_deprecations(
+    llm_router: Router | None,
+    warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS,
+    today: date | None = None,
+) -> ModelDeprecationResponse:
+    """Bucket every deployment carrying a deprecation date by how urgent it is"""
+    snapshot_time: Final = datetime.now(timezone.utc)
+    effective_today: Final = today or snapshot_time.date()
+    deployments: Final = (llm_router.get_model_list() or ()) if llm_router is not None else ()
+
+    deduped: Final = _dedupe(
+        tuple(
+            info
+            for info in (_build_info(deployment, effective_today, warn_within_days) for deployment in deployments)
+            if info is not None
+        )
+    )
+
+    verbose_logger.debug(
+        "model_deprecation: %d/%d deployments carry a deprecation date",
+        len(deduped),
+        len(deployments),
+    )
+
+    return ModelDeprecationResponse(
+        deprecated=_bucket(deduped, "deprecated"),
+        imminent=_bucket(deduped, "imminent"),
+        upcoming=_bucket(deduped, "upcoming"),
+        warn_within_days=warn_within_days,
+        checked_at=snapshot_time,
+    )
+
+
+def _escape_slack_mrkdwn(value: str) -> str:
+    """Neutralize Slack control characters so a model name cannot forge a mention or link"""
+    return value.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def _format_entry(info: ModelDeprecationInfo) -> str:
+    suffix: Final = (
+        f"already deprecated {abs(info.days_until_deprecation)}d ago"
+        if info.days_until_deprecation < 0
+        else f"in {info.days_until_deprecation}d"
+    )
+    return (
+        f"• `{_escape_slack_mrkdwn(info.model_name)}` "
+        f"(provider: {_escape_slack_mrkdwn(info.litellm_provider) if info.litellm_provider else 'unknown'}, "
+        f"deprecates {info.deprecation_date.isoformat()}, {suffix})"
+    )
+
+
+def format_deprecation_alert_message(
+    snapshot: ModelDeprecationResponse,
+) -> str | None:
+    """Render the alert for the deprecated and imminent buckets, None when both are empty
+
+    Upcoming models are left out of the alert to keep it actionable.
+    """
+    if not snapshot.deprecated and not snapshot.imminent:
+        return None
+
+    deprecated_section: Final = (
+        ("\n*Already deprecated:*", *(_format_entry(i) for i in snapshot.deprecated)) if snapshot.deprecated else ()
+    )
+    imminent_section: Final = (
+        (
+            f"\n*Deprecating within {snapshot.warn_within_days} days:*",
+            *(_format_entry(i) for i in snapshot.imminent),
+        )
+        if snapshot.imminent
+        else ()
+    )
+
+    return "\n".join(
+        (
+            "*⚠️ Model Deprecation Warning*",
+            *deprecated_section,
+            *imminent_section,
+            "\nPlan migrations to a supported model. See "
+            "https://docs.litellm.ai/docs/proxy/model_management for guidance.",
+        )
+    )

litellm/proxy/proxy_server.py

@@ -328,6 +328,7 @@ from litellm.proxy.common_utils.load_config_utils import (
     get_config_file_contents_from_gcs,
     get_file_contents_from_s3,
 )
+from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations
 from litellm.proxy.common_utils.model_listing_utils import TeamModelNameTranslator
 from litellm.proxy.common_utils.openai_endpoint_utils import (
     remove_sensitive_info_from_deployment,
@@ -650,6 +651,10 @@ from litellm.types.proxy.management_endpoints.ui_sso import (
     DefaultTeamSSOParams,
     LiteLLM_UpperboundKeyGenerateParams,
 )
+from litellm.types.proxy.model_deprecation import (
+    DEFAULT_DEPRECATION_WARN_DAYS,
+    ModelDeprecationResponse,
+)
 from litellm.types.realtime import RealtimeQueryParams
 from litellm.types.router import (
     DeploymentTypedDict,
@@ -13927,6 +13932,48 @@ async def model_info_v1(
     return {"data": all_models}
 
 
+@router.get(
+    "/model/deprecations",
+    tags=("model management",),
+    dependencies=(Depends(user_api_key_auth),),
+    response_model=ModelDeprecationResponse,
+)
+@router.get(
+    "/v1/model/deprecations",
+    tags=("model management",),
+    dependencies=(Depends(user_api_key_auth),),
+    response_model=ModelDeprecationResponse,
+)
+async def model_deprecations(
+    warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS,
+) -> ModelDeprecationResponse:
+    """List models with known deprecation/sunset dates, bucketed by urgency.
+
+    Reads `deprecation_date` metadata from `model_prices_and_context_window.json`
+    (and any per-deployment `model_info.deprecation_date` overrides) for the
+    models configured on this proxy.
+
+    Parameters:
+        warn_within_days: Window (in days) used to bucket "imminent" models,
+            30 by default.
+
+    Returns:
+        A payload with three lists of `ModelDeprecationInfo` entries:
+
+        - `deprecated`: deprecation date is in the past, so these requests may
+          fail at any time.
+        - `imminent`: deprecation date is within `warn_within_days` from today.
+        - `upcoming`: deprecation date is further out.
+
+    Example:
+    ```shell
+    curl -X GET 'http://localhost:4000/model/deprecations' \\
+        -H 'Authorization: Bearer sk-1234'
+    ```
+    """
+    return collect_model_deprecations(llm_router=llm_router, warn_within_days=warn_within_days)
+
+
 def _get_model_group_info(
     llm_router: Router, all_models_str: list[str], model_group: str | None
 ) -> list[ModelGroupInfoProxy]:

litellm/proxy/utils.py

@@ -475,6 +475,7 @@ class ProxyLogging:
         # Guard flags to prevent duplicate background tasks
         self.daily_report_started: bool = False
         self.hanging_requests_check_started: bool = False
+        self.deprecation_check_started: bool = False
 
     def startup_event(
         self,
@@ -517,6 +518,25 @@ class ProxyLogging:
             )  # RUN HANGING REQUEST CHECK (if user wants to alert on hanging requests)
             self.hanging_requests_check_started = True
 
+        self._ensure_deprecation_check_scheduled()
+
+    def _ensure_deprecation_check_scheduled(self) -> None:
+        """Alerting can be configured at startup or by a later config reload, so schedule from either path"""
+        if self.alerting is None or self.deprecation_check_started:
+            return
+
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return
+
+        asyncio.create_task(
+            self.slack_alerting_instance.run_scheduled_deprecation_check(
+                pod_lock_manager=self.db_spend_update_writer.pod_lock_manager
+            )
+        )
+        self.deprecation_check_started = True
+
     def update_values(
         self,
         alerting: list | None = None,
@@ -544,6 +564,7 @@ class ProxyLogging:
             updated_slack_alerting = True
 
         if updated_slack_alerting is True:
+            self._ensure_deprecation_check_scheduled()
             self.slack_alerting_instance.update_values(
                 alerting=self.alerting,
                 alerting_threshold=self.alerting_threshold,

litellm/types/integrations/slack_alerting.py

@@ -121,6 +121,7 @@ class SlackAlertingCacheKeys(Enum):
     failed_requests_key = "failed_requests_daily_metrics"
     latency_key = "latency_daily_metrics"
     report_sent_key = "daily_metrics_report_sent"
+    deprecation_alert_sent_key = "model_deprecation_alert_sent"
 
 
 class AlertType(str, Enum):
@@ -147,6 +148,7 @@ class AlertType(str, Enum):
     # Deployment alerts
     cooldown_deployment = "cooldown_deployment"
     new_model_added = "new_model_added"
+    model_deprecation_warnings = "model_deprecation_warnings"
 
     # Outage alerts
     outage_alerts = "outage_alerts"
@@ -187,6 +189,7 @@ DEFAULT_ALERT_TYPES: Final[list[AlertType]] = [
     # Deployment alerts
     AlertType.cooldown_deployment,
     AlertType.new_model_added,
+    AlertType.model_deprecation_warnings,
     # Outage alerts
     AlertType.outage_alerts,
     AlertType.region_outage_alerts,

litellm/types/proxy/model_deprecation.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Final, Literal
+
+from pydantic import BaseModel, Field
+
+DEFAULT_DEPRECATION_WARN_DAYS: Final = 30
+
+DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS: Final = 24 * 60 * 60
+
+DEPRECATION_IDLE_POLL_SECONDS: Final = 30
+
+DeprecationStatus = Literal["upcoming", "imminent", "deprecated"]
+
+
+class ModelDeprecationInfo(BaseModel):
+    model_name: str = Field(description="The public name of the model on the proxy (model_group).")
+    litellm_model: str | None = Field(
+        default=None,
+        description="The underlying litellm model string the deprecation date is sourced from.",
+    )
+    deprecation_date: date = Field(description="The date (UTC) when the model becomes deprecated.")
+    days_until_deprecation: int = Field(
+        description=("Days remaining until the deprecation date. Negative if the model is already deprecated."),
+    )
+    status: DeprecationStatus = Field(
+        description=(
+            "'deprecated' if the date has passed, 'imminent' if it falls within warn_within_days, 'upcoming' otherwise."
+        ),
+    )
+    litellm_provider: str | None = Field(default=None, description="The provider this model belongs to.")
+
+
+class ModelDeprecationResponse(BaseModel):
+    deprecated: list[ModelDeprecationInfo] = Field(
+        default_factory=list,
+        description="Models whose deprecation date has already passed.",
+    )
+    imminent: list[ModelDeprecationInfo] = Field(
+        default_factory=list,
+        description=(
+            "Models whose deprecation date is within warn_within_days from "
+            "today and require immediate migration planning."
+        ),
+    )
+    upcoming: list[ModelDeprecationInfo] = Field(
+        default_factory=list,
+        description="Models with a future deprecation date outside the warn window.",
+    )
+    warn_within_days: int = Field(description="The window (in days) used to bucket 'imminent' models.")
+    checked_at: datetime = Field(description="UTC timestamp when the deprecation snapshot was generated.")

tests/test_litellm/integrations/SlackAlerting/test_model_deprecation_alert.py

@@ -0,0 +1,395 @@
+"""Tests for the Slack alerting model deprecation hook."""
+
+import asyncio
+import os
+import sys
+from itertools import chain, repeat
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+import litellm
+from litellm.constants import SLACK_MODEL_DEPRECATION_LOCK_ID
+from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
+from litellm.proxy._types import AlertType
+from litellm.types.integrations.slack_alerting import SlackAlertingCacheKeys
+from litellm.types.proxy.model_deprecation import (
+    DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+    DEPRECATION_IDLE_POLL_SECONDS,
+)
+
+DEAD_MODEL_COST = {
+    "dead-model": {"deprecation_date": "2020-01-01", "litellm_provider": "openai"}
+}
+DEAD_ALIAS_DEPLOYMENT = {
+    "model_name": "dead-alias",
+    "litellm_params": {"model": "dead-model"},
+    "model_info": {"id": "1"},
+}
+
+
+def _make_router(deployments):
+    router = MagicMock()
+    router.get_model_list.return_value = deployments
+    return router
+
+
+@pytest.mark.asyncio
+async def test_should_skip_when_alert_type_disabled():
+    alerting = SlackAlerting(
+        alerting=["slack"],
+        alert_types=[AlertType.llm_exceptions],
+    )
+    sent = await alerting.send_model_deprecation_alert(llm_router=MagicMock())
+    assert sent is False
+
+
+@pytest.mark.asyncio
+async def test_should_skip_when_no_alerting_configured():
+    alerting = SlackAlerting(
+        alerting=None,
+        alert_types=[AlertType.model_deprecation_warnings],
+    )
+    sent = await alerting.send_model_deprecation_alert(llm_router=MagicMock())
+    assert sent is False
+
+
+@pytest.mark.asyncio
+async def test_should_skip_when_no_deprecations_found(monkeypatch):
+    monkeypatch.setattr(litellm, "model_cost", {})
+    alerting = SlackAlerting(
+        alerting=["slack"],
+        alert_types=[AlertType.model_deprecation_warnings],
+    )
+    router = _make_router(
+        [
+            {
+                "model_name": "fresh",
+                "litellm_params": {"model": "openai/gpt-4o"},
+                "model_info": {"id": "x"},
+            }
+        ]
+    )
+    sent = await alerting.send_model_deprecation_alert(llm_router=router)
+    assert sent is False
+
+
+@pytest.mark.asyncio
+async def test_should_dispatch_high_severity_when_deprecated(monkeypatch):
+    monkeypatch.setattr(
+        litellm,
+        "model_cost",
+        {
+            "dead-model": {
+                "deprecation_date": "2020-01-01",
+                "litellm_provider": "openai",
+            }
+        },
+    )
+    alerting = SlackAlerting(
+        alerting=["slack"],
+        alert_types=[AlertType.model_deprecation_warnings],
+    )
+    router = _make_router(
+        [
+            {
+                "model_name": "dead-alias",
+                "litellm_params": {"model": "dead-model"},
+                "model_info": {"id": "1"},
+            }
+        ]
+    )
+
+    with patch.object(
+        alerting, "send_alert", new_callable=AsyncMock
+    ) as mock_send_alert:
+        sent = await alerting.send_model_deprecation_alert(llm_router=router)
+
+    assert sent is True
+    mock_send_alert.assert_awaited_once()
+    call_kwargs = mock_send_alert.await_args.kwargs
+    assert call_kwargs["alert_type"] == AlertType.model_deprecation_warnings
+    assert call_kwargs["level"] == "High"
+    assert call_kwargs["alerting_metadata"]["deprecated_count"] == 1
+    assert call_kwargs["alerting_metadata"]["imminent_count"] == 0
+    assert "dead-alias" in call_kwargs["message"]
+    assert isinstance(
+        await alerting.internal_usage_cache.async_get_cache(
+            key=SlackAlertingCacheKeys.deprecation_alert_sent_key.value
+        ),
+        float,
+    )
+
+
+@pytest.mark.asyncio
+async def test_should_alert_once_the_alert_type_and_router_arrive_after_startup(
+    monkeypatch,
+):
+    """The loop starts before config reload, so a disabled pass must not cost a day of alerts"""
+    monkeypatch.setattr(
+        litellm,
+        "model_cost",
+        {"dead-model": {"deprecation_date": "2020-01-01", "litellm_provider": "openai"}},
+    )
+    alerting = SlackAlerting(alerting=["slack"], alert_types=[AlertType.llm_exceptions])
+    router = _make_router(
+        [
+            {
+                "model_name": "dead-alias",
+                "litellm_params": {"model": "dead-model"},
+                "model_info": {"id": "1"},
+            }
+        ]
+    )
+
+    slept: list[float] = []
+
+    async def stop_after_third_pass(seconds):
+        slept.append(seconds)
+        if alerting.alert_types == [AlertType.llm_exceptions]:
+            alerting.update_values(
+                alert_types=[AlertType.model_deprecation_warnings]
+            )  # simulates a config reload enabling the alert
+        if len(slept) == 3:
+            raise asyncio.CancelledError
+
+    with (
+        patch.object(alerting, "send_alert", new_callable=AsyncMock) as mock_send_alert,
+        patch(
+            "litellm.integrations.SlackAlerting.slack_alerting.asyncio.sleep",
+            side_effect=stop_after_third_pass,
+        ),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        await alerting.run_scheduled_deprecation_check(get_llm_router=lambda: router)
+
+    assert slept == [DEPRECATION_IDLE_POLL_SECONDS] * 3
+    mock_send_alert.assert_awaited_once()
+    assert "dead-alias" in mock_send_alert.await_args.kwargs["message"]
+
+
+@pytest.mark.asyncio
+async def test_should_wait_for_the_router_instead_of_sleeping_a_full_day(monkeypatch):
+    """Config load can start the loop before the router exists, which must not cost a day of alerts"""
+    monkeypatch.setattr(
+        litellm,
+        "model_cost",
+        {"dead-model": {"deprecation_date": "2020-01-01", "litellm_provider": "openai"}},
+    )
+    alerting = SlackAlerting(
+        alerting=["slack"], alert_types=[AlertType.model_deprecation_warnings]
+    )
+    router = _make_router(
+        [
+            {
+                "model_name": "dead-alias",
+                "litellm_params": {"model": "dead-model"},
+                "model_info": {"id": "1"},
+            }
+        ]
+    )
+    router_absent_passes = 100
+    routers = chain(repeat(None, router_absent_passes), repeat(router))
+    slept: list[float] = []
+
+    async def record_sleep(seconds):
+        slept.append(seconds)
+        if len(slept) > router_absent_passes:
+            raise asyncio.CancelledError
+
+    with (
+        patch.object(alerting, "send_alert", new_callable=AsyncMock) as mock_send_alert,
+        patch(
+            "litellm.integrations.SlackAlerting.slack_alerting.asyncio.sleep",
+            side_effect=record_sleep,
+        ),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        await alerting.run_scheduled_deprecation_check(
+            get_llm_router=lambda: next(routers)
+        )
+
+    assert slept == [DEPRECATION_IDLE_POLL_SECONDS] * (router_absent_passes + 1)
+    mock_send_alert.assert_awaited_once()
+    assert "dead-alias" in mock_send_alert.await_args.kwargs["message"]
+
+
+@pytest.mark.parametrize(
+    "lock_acquired, expect_alert",
+    [(True, True), (None, True), (False, False)],
+    ids=["lock won", "no redis lock", "another pod holds the lock"],
+)
+@pytest.mark.asyncio
+async def test_should_alert_only_from_the_pod_holding_the_daily_lock(
+    monkeypatch, lock_acquired, expect_alert
+):
+    """Every pod runs the loop, so a fleet must not send one identical alert per replica"""
+    monkeypatch.setattr(
+        litellm,
+        "model_cost",
+        {"dead-model": {"deprecation_date": "2020-01-01", "litellm_provider": "openai"}},
+    )
+    alerting = SlackAlerting(
+        alerting=["slack"], alert_types=[AlertType.model_deprecation_warnings]
+    )
+    router = _make_router(
+        [
+            {
+                "model_name": "dead-alias",
+                "litellm_params": {"model": "dead-model"},
+                "model_info": {"id": "1"},
+            }
+        ]
+    )
+    pod_lock_manager = MagicMock()
+    pod_lock_manager.acquire_lock = AsyncMock(return_value=lock_acquired)
+
+    with (
+        patch.object(alerting, "send_alert", new_callable=AsyncMock) as mock_send_alert,
+        patch(
+            "litellm.integrations.SlackAlerting.slack_alerting.asyncio.sleep",
+            side_effect=asyncio.CancelledError,
+        ),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        await alerting.run_scheduled_deprecation_check(
+            get_llm_router=lambda: router, pod_lock_manager=pod_lock_manager
+        )
+
+    assert mock_send_alert.await_count == int(expect_alert)
+    assert pod_lock_manager.acquire_lock.await_args.kwargs == {
+        "cronjob_id": SLACK_MODEL_DEPRECATION_LOCK_ID,
+        "ttl": DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+        "allow_reentrant": False,
+    }
+
+
+@pytest.mark.asyncio
+async def test_should_retry_on_the_next_poll_when_the_lock_claim_fails(monkeypatch):
+    """A redis blip at claim time returns False like a held lock, and must not cost every pod a day of alerts"""
+    monkeypatch.setattr(litellm, "model_cost", DEAD_MODEL_COST)
+    alerting = SlackAlerting(
+        alerting=["slack"], alert_types=[AlertType.model_deprecation_warnings]
+    )
+    router = _make_router([DEAD_ALIAS_DEPLOYMENT])
+    pod_lock_manager = MagicMock()
+    pod_lock_manager.acquire_lock = AsyncMock(side_effect=[False, True])
+    slept: list[float] = []
+
+    async def stop_after_second_pass(seconds):
+        slept.append(seconds)
+        if len(slept) == 2:
+            raise asyncio.CancelledError
+
+    with (
+        patch.object(alerting, "send_alert", new_callable=AsyncMock) as mock_send_alert,
+        patch(
+            "litellm.integrations.SlackAlerting.slack_alerting.asyncio.sleep",
+            side_effect=stop_after_second_pass,
+        ),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        await alerting.run_scheduled_deprecation_check(
+            get_llm_router=lambda: router, pod_lock_manager=pod_lock_manager
+        )
+
+    assert slept == [DEPRECATION_IDLE_POLL_SECONDS] * 2
+    assert pod_lock_manager.acquire_lock.await_count == 2
+    mock_send_alert.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_should_not_claim_the_lock_when_there_is_nothing_to_report(monkeypatch):
+    """An empty pass must not hold the daily lock, or a sunset added later waits out the whole window"""
+    monkeypatch.setattr(litellm, "model_cost", {})
+    alerting = SlackAlerting(
+        alerting=["slack"], alert_types=[AlertType.model_deprecation_warnings]
+    )
+    router = _make_router(
+        [
+            {
+                "model_name": "fresh",
+                "litellm_params": {"model": "openai/gpt-4o"},
+                "model_info": {"id": "x"},
+            }
+        ]
+    )
+    pod_lock_manager = MagicMock()
+    pod_lock_manager.acquire_lock = AsyncMock(return_value=True)
+
+    with patch.object(alerting, "send_alert", new_callable=AsyncMock) as mock_send_alert:
+        sent = await alerting.send_model_deprecation_alert(
+            llm_router=router, pod_lock_manager=pod_lock_manager
+        )
+
+    assert sent is False
+    pod_lock_manager.acquire_lock.assert_not_awaited()
+    mock_send_alert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_should_not_alert_or_claim_the_lock_within_a_day_of_a_sent_alert(monkeypatch):
+    """The shared sent stamp keeps sibling pods and restarts from re-alerting or re-asking redis for a day"""
+    monkeypatch.setattr(litellm, "model_cost", DEAD_MODEL_COST)
+    alerting = SlackAlerting(
+        alerting=["slack"], alert_types=[AlertType.model_deprecation_warnings]
+    )
+    await alerting.internal_usage_cache.async_set_cache(
+        key=SlackAlertingCacheKeys.deprecation_alert_sent_key.value,
+        value=1.0,
+        ttl=DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+    )
+    router = _make_router([DEAD_ALIAS_DEPLOYMENT])
+    pod_lock_manager = MagicMock()
+    pod_lock_manager.acquire_lock = AsyncMock(return_value=True)
+
+    with (
+        patch.object(alerting, "send_alert", new_callable=AsyncMock) as mock_send_alert,
+        patch(
+            "litellm.integrations.SlackAlerting.slack_alerting.asyncio.sleep",
+            side_effect=asyncio.CancelledError,
+        ),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        await alerting.run_scheduled_deprecation_check(
+            get_llm_router=lambda: router, pod_lock_manager=pod_lock_manager
+        )
+
+    pod_lock_manager.acquire_lock.assert_not_awaited()
+    mock_send_alert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_should_back_off_a_full_day_after_a_pass_raises(monkeypatch):
+    """A misconfigured webhook raises on every send, which must log once a day rather than every poll"""
+    monkeypatch.setattr(litellm, "model_cost", DEAD_MODEL_COST)
+    alerting = SlackAlerting(
+        alerting=["slack"], alert_types=[AlertType.model_deprecation_warnings]
+    )
+    router = _make_router([DEAD_ALIAS_DEPLOYMENT])
+    slept: list[float] = []
+
+    async def stop_after_second_pass(seconds):
+        slept.append(seconds)
+        if len(slept) == 2:
+            raise asyncio.CancelledError
+
+    with (
+        patch.object(
+            alerting,
+            "send_alert",
+            new_callable=AsyncMock,
+            side_effect=ValueError("Missing SLACK_WEBHOOK_URL from environment"),
+        ) as mock_send_alert,
+        patch(
+            "litellm.integrations.SlackAlerting.slack_alerting.asyncio.sleep",
+            side_effect=stop_after_second_pass,
+        ),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        await alerting.run_scheduled_deprecation_check(get_llm_router=lambda: router)
+
+    assert slept == [DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS] * 2
+    assert mock_send_alert.await_count == 2

tests/test_litellm/proxy/common_utils/test_model_deprecation.py

@@ -0,0 +1,362 @@
+"""Tests for the model deprecation helper module.
+
+These tests focus on the helper itself — not on the proxy endpoint or
+Slack integration — so they can run without the full proxy stack.
+"""
+
+import os
+import sys
+from datetime import date, datetime, timezone
+from unittest.mock import MagicMock
+
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+import litellm
+from litellm.proxy.common_utils.model_deprecation import (
+    _classify,
+    _parse_deprecation_date,
+    collect_model_deprecations,
+    format_deprecation_alert_message,
+)
+
+
+def _make_router(deployments):
+    router = MagicMock()
+    router.get_model_list.return_value = deployments
+    return router
+
+
+class TestParseDeprecationDate:
+    def test_should_parse_iso_string(self):
+        assert _parse_deprecation_date("2026-12-31") == date(2026, 12, 31)
+
+    def test_should_pass_through_date_object(self):
+        d = date(2026, 1, 1)
+        assert _parse_deprecation_date(d) == d
+
+    def test_should_return_none_for_documentation_sentinel(self):
+        # The JSON map ships a sentinel string under the "sample_spec" key.
+        assert (
+            _parse_deprecation_date(
+                "date when the model becomes deprecated in the format YYYY-MM-DD"
+            )
+            is None
+        )
+
+    def test_should_return_none_for_none(self):
+        assert _parse_deprecation_date(None) is None
+
+    def test_should_return_none_for_unsupported_type(self):
+        assert _parse_deprecation_date(12345) is None
+
+    def test_should_narrow_datetime_to_date(self):
+        assert _parse_deprecation_date(
+            datetime(2026, 12, 31, 23, 59, tzinfo=timezone.utc)
+        ) == date(2026, 12, 31)
+
+
+class TestClassify:
+    def test_should_classify_past_dates_as_deprecated(self):
+        assert _classify(-1, warn_within_days=30) == "deprecated"
+        assert _classify(-365, warn_within_days=30) == "deprecated"
+
+    def test_should_classify_inside_window_as_imminent(self):
+        assert _classify(0, warn_within_days=30) == "imminent"
+        assert _classify(15, warn_within_days=30) == "imminent"
+        assert _classify(30, warn_within_days=30) == "imminent"
+
+    def test_should_classify_outside_window_as_upcoming(self):
+        assert _classify(31, warn_within_days=30) == "upcoming"
+        assert _classify(365, warn_within_days=30) == "upcoming"
+
+
+class TestCollectModelDeprecations:
+    def test_should_return_empty_response_when_router_is_none(self):
+        snapshot = collect_model_deprecations(llm_router=None)
+        assert snapshot.deprecated == []
+        assert snapshot.imminent == []
+        assert snapshot.upcoming == []
+
+    def test_should_skip_models_without_deprecation_metadata(self, monkeypatch):
+        monkeypatch.setattr(litellm, "model_cost", {})
+        router = _make_router(
+            [
+                {
+                    "model_name": "gpt-4o",
+                    "litellm_params": {"model": "openai/gpt-4o"},
+                    "model_info": {"id": "abc"},
+                }
+            ]
+        )
+        snapshot = collect_model_deprecations(llm_router=router)
+        assert snapshot.deprecated == []
+        assert snapshot.imminent == []
+        assert snapshot.upcoming == []
+
+    def test_should_classify_into_three_buckets(self, monkeypatch):
+        today = date(2026, 6, 1)
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {
+                "deprecated-model": {
+                    "deprecation_date": "2026-01-01",
+                    "litellm_provider": "openai",
+                },
+                "imminent-model": {
+                    "deprecation_date": "2026-06-15",
+                    "litellm_provider": "openai",
+                },
+                "upcoming-model": {
+                    "deprecation_date": "2027-01-01",
+                    "litellm_provider": "openai",
+                },
+            },
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "deprecated-alias",
+                    "litellm_params": {"model": "openai/deprecated-model"},
+                    "model_info": {"id": "1"},
+                },
+                {
+                    "model_name": "imminent-alias",
+                    "litellm_params": {"model": "imminent-model"},
+                    "model_info": {"id": "2"},
+                },
+                {
+                    "model_name": "upcoming-alias",
+                    "litellm_params": {"model": "openai/upcoming-model"},
+                    "model_info": {"id": "3"},
+                },
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=today
+        )
+
+        assert [m.model_name for m in snapshot.deprecated] == ["deprecated-alias"]
+        assert [m.model_name for m in snapshot.imminent] == ["imminent-alias"]
+        assert [m.model_name for m in snapshot.upcoming] == ["upcoming-alias"]
+
+        assert snapshot.deprecated[0].days_until_deprecation < 0
+        assert snapshot.imminent[0].days_until_deprecation == 14
+        assert snapshot.upcoming[0].days_until_deprecation > 30
+
+    def test_should_prefer_explicit_deployment_override(self, monkeypatch):
+        today = date(2026, 6, 1)
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {"some-model": {"deprecation_date": "2030-01-01"}},
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "my-alias",
+                    "litellm_params": {"model": "some-model"},
+                    "model_info": {
+                        "id": "x",
+                        "deprecation_date": "2026-06-10",
+                    },
+                }
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=today
+        )
+
+        assert len(snapshot.imminent) == 1
+        assert snapshot.imminent[0].deprecation_date == date(2026, 6, 10)
+
+    def test_should_dedupe_duplicate_deployments_in_same_group(self, monkeypatch):
+        today = date(2026, 6, 1)
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {"shared-model": {"deprecation_date": "2026-06-10"}},
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "alias",
+                    "litellm_params": {"model": "shared-model"},
+                    "model_info": {"id": "1"},
+                },
+                {
+                    "model_name": "alias",
+                    "litellm_params": {"model": "shared-model"},
+                    "model_info": {"id": "2"},
+                },
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=today
+        )
+
+        assert len(snapshot.imminent) == 1
+
+    def test_should_resolve_via_unprefixed_model_name(self, monkeypatch):
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {"gpt-4o": {"deprecation_date": "2026-06-10"}},
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "alias",
+                    "litellm_params": {"model": "openai/gpt-4o"},
+                    "model_info": {"id": "1"},
+                }
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=date(2026, 6, 1)
+        )
+
+        assert [m.litellm_model for m in snapshot.imminent] == ["gpt-4o"]
+
+    def test_should_keep_both_dates_when_group_has_conflicting_dates(self, monkeypatch):
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {"shared-model": {"deprecation_date": "2026-06-10"}},
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "alias",
+                    "litellm_params": {"model": "shared-model"},
+                    "model_info": {"id": "1"},
+                },
+                {
+                    "model_name": "alias",
+                    "litellm_params": {"model": "shared-model"},
+                    "model_info": {"id": "2", "deprecation_date": "2027-01-01"},
+                },
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=date(2026, 6, 1)
+        )
+
+        assert len(snapshot.imminent) == 1
+        assert len(snapshot.upcoming) == 1
+
+    def test_should_resolve_via_base_model(self, monkeypatch):
+        today = date(2026, 6, 1)
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {"base-thing": {"deprecation_date": "2026-06-10"}},
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "alias",
+                    "litellm_params": {"model": "azure/some-deployment-name"},
+                    "model_info": {"id": "1", "base_model": "base-thing"},
+                }
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=today
+        )
+
+        assert len(snapshot.imminent) == 1
+        assert snapshot.imminent[0].litellm_model == "base-thing"
+
+
+class TestFormatDeprecationAlertMessage:
+    def test_should_return_none_when_nothing_to_alert(self):
+        snapshot = collect_model_deprecations(llm_router=None)
+        assert format_deprecation_alert_message(snapshot) is None
+
+    def test_should_render_imminent_and_deprecated_sections(self, monkeypatch):
+        today = date(2026, 6, 1)
+        monkeypatch.setattr(
+            litellm,
+            "model_cost",
+            {
+                "dead-model": {
+                    "deprecation_date": "2026-01-01",
+                    "litellm_provider": "openai",
+                },
+                "soon-model": {
+                    "deprecation_date": "2026-06-15",
+                    "litellm_provider": "anthropic",
+                },
+                "later-model": {
+                    "deprecation_date": "2027-01-01",
+                    "litellm_provider": "anthropic",
+                },
+            },
+        )
+        router = _make_router(
+            [
+                {
+                    "model_name": "dead",
+                    "litellm_params": {"model": "dead-model"},
+                    "model_info": {"id": "1"},
+                },
+                {
+                    "model_name": "soon",
+                    "litellm_params": {"model": "soon-model"},
+                    "model_info": {"id": "2"},
+                },
+                {
+                    "model_name": "later",
+                    "litellm_params": {"model": "later-model"},
+                    "model_info": {"id": "3"},
+                },
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=today
+        )
+        message = format_deprecation_alert_message(snapshot)
+
+        assert message is not None
+        assert "Already deprecated" in message
+        assert "Deprecating within 30 days" in message
+        assert "`dead`" in message
+        assert "`soon`" in message
+        # Upcoming models must NOT be in the alert (avoid alert fatigue).
+        assert "`later`" not in message
+
+    def test_should_neutralize_slack_markup_from_model_metadata(self):
+        today = date(2026, 6, 1)
+        router = _make_router(
+            [
+                {
+                    "model_name": "<!channel> pwned",
+                    "litellm_params": {"model": "openai/whatever"},
+                    "model_info": {
+                        "id": "1",
+                        "deprecation_date": "2026-06-10",
+                        "litellm_provider": "<https://evil.example|openai> & co",
+                    },
+                }
+            ]
+        )
+
+        snapshot = collect_model_deprecations(
+            llm_router=router, warn_within_days=30, today=today
+        )
+        message = format_deprecation_alert_message(snapshot)
+
+        assert message is not None
+        assert "<!channel>" not in message
+        assert "<https://evil.example|openai>" not in message
+        assert "&lt;!channel&gt; pwned" in message
+        assert "&lt;https://evil.example|openai&gt; &amp; co" in message

tests/test_litellm/proxy/test_model_deprecations_endpoint.py

@@ -0,0 +1,77 @@
+import os
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+import litellm
+from litellm.proxy import proxy_server
+from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.proxy.proxy_server import app
+
+client = TestClient(app)
+
+
+@pytest.fixture
+def authenticated_client(monkeypatch):
+    app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
+        user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1234"
+    )
+    monkeypatch.setattr(
+        litellm,
+        "model_cost",
+        {
+            "sunset-model": {
+                "deprecation_date": "2020-01-01",
+                "litellm_provider": "openai",
+            },
+            "future-model": {
+                "deprecation_date": "2099-01-01",
+                "litellm_provider": "openai",
+            },
+        },
+    )
+    router = MagicMock()
+    router.get_model_list.return_value = [
+        {
+            "model_name": "sunset-alias",
+            "litellm_params": {"model": "sunset-model"},
+            "model_info": {"id": "1"},
+        },
+        {
+            "model_name": "future-alias",
+            "litellm_params": {"model": "future-model"},
+            "model_info": {"id": "2"},
+        },
+    ]
+    monkeypatch.setattr(proxy_server, "llm_router", router)
+    yield client
+    app.dependency_overrides.pop(user_api_key_auth, None)
+
+
+def test_should_bucket_configured_models_by_urgency(authenticated_client):
+    response = authenticated_client.get("/model/deprecations")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert [m["model_name"] for m in payload["deprecated"]] == ["sunset-alias"]
+    assert [m["model_name"] for m in payload["upcoming"]] == ["future-alias"]
+    assert payload["imminent"] == []
+    assert payload["warn_within_days"] == 30
+    assert payload["deprecated"][0]["days_until_deprecation"] < 0
+
+
+def test_should_rebucket_with_warn_within_days_override(authenticated_client):
+    response = authenticated_client.get(
+        "/v1/model/deprecations", params={"warn_within_days": 40000}
+    )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert [m["model_name"] for m in payload["imminent"]] == ["future-alias"]
+    assert payload["upcoming"] == []
+    assert payload["warn_within_days"] == 40000

tests/test_litellm/proxy/utils/proxy_logging/test_lifecycle.py

@@ -130,6 +130,42 @@ def test_startup_event_initializes_slack_and_callbacks(proxy_logging):
     }
 
 
+@pytest.mark.asyncio
+async def test_startup_event_schedules_deprecation_check_before_its_alert_type_is_on(proxy_logging):
+    """Alerting config can enable the deprecation alert after startup, so the loop must already be running"""
+    proxy_logging.alerting = ["slack"]
+    proxy_logging.slack_alerting_instance = MagicMock()
+    proxy_logging.slack_alerting_instance.alert_types = []
+    proxy_logging.slack_alerting_instance.run_scheduled_deprecation_check = AsyncMock()
+    proxy_logging._init_litellm_callbacks = MagicMock()
+
+    proxy_logging.startup_event(llm_router=None, redis_usage_cache=None)
+
+    assert proxy_logging.deprecation_check_started is True
+    proxy_logging.slack_alerting_instance.run_scheduled_deprecation_check.assert_called_once_with(
+        pod_lock_manager=proxy_logging.db_spend_update_writer.pod_lock_manager
+    )
+
+
+@pytest.mark.asyncio
+async def test_update_values_schedules_deprecation_check_when_alerting_arrives_later(proxy_logging):
+    """A proxy that boots without alerting still needs the loop once a config reload turns it on"""
+    proxy_logging.slack_alerting_instance = MagicMock()
+    proxy_logging.slack_alerting_instance.alert_types = []
+    proxy_logging.slack_alerting_instance.run_scheduled_deprecation_check = AsyncMock()
+    proxy_logging._init_litellm_callbacks = MagicMock()
+
+    proxy_logging.startup_event(llm_router=None, redis_usage_cache=None)
+    assert proxy_logging.deprecation_check_started is False
+
+    proxy_logging.update_values(alerting=["slack"])
+
+    assert proxy_logging.deprecation_check_started is True
+    proxy_logging.slack_alerting_instance.run_scheduled_deprecation_check.assert_called_once_with(
+        pod_lock_manager=proxy_logging.db_spend_update_writer.pod_lock_manager
+    )
+
+
 def test_startup_event_propagates_init_callbacks_failure_raises(proxy_logging):
     proxy_logging.slack_alerting_instance = MagicMock()
     proxy_logging.slack_alerting_instance.alert_types = []

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

@@ -296,6 +296,7 @@ const Settings: React.FC<SettingsPageProps> = ({ accessToken, userRole, userID,
     daily_reports: "Weekly/Monthly Spend Reports",
     outage_alerts: "Outage Alerts",
     region_outage_alerts: "Region Outage Alerts",
+    model_deprecation_warnings: "Model Deprecation Warnings",
   };
 
   useEffect(() => {

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -8036,6 +8036,48 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/model/deprecations": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /**
+         * Model Deprecations
+         * @description List models with known deprecation/sunset dates, bucketed by urgency.
+         *
+         *     Reads `deprecation_date` metadata from `model_prices_and_context_window.json`
+         *     (and any per-deployment `model_info.deprecation_date` overrides) for the
+         *     models configured on this proxy.
+         *
+         *     Parameters:
+         *         warn_within_days: Window (in days) used to bucket "imminent" models,
+         *             30 by default.
+         *
+         *     Returns:
+         *         A payload with three lists of `ModelDeprecationInfo` entries:
+         *
+         *         - `deprecated`: deprecation date is in the past, so these requests may
+         *           fail at any time.
+         *         - `imminent`: deprecation date is within `warn_within_days` from today.
+         *         - `upcoming`: deprecation date is further out.
+         *
+         *     Example:
+         *     ```shell
+         *     curl -X GET 'http://localhost:4000/model/deprecations' \
+         *         -H 'Authorization: Bearer sk-1234'
+         *     ```
+         */
+        get: operations["model_deprecations_model_deprecations_get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/model/info": {
         parameters: {
             query?: never;
@@ -17674,6 +17716,48 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/v1/model/deprecations": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /**
+         * Model Deprecations
+         * @description List models with known deprecation/sunset dates, bucketed by urgency.
+         *
+         *     Reads `deprecation_date` metadata from `model_prices_and_context_window.json`
+         *     (and any per-deployment `model_info.deprecation_date` overrides) for the
+         *     models configured on this proxy.
+         *
+         *     Parameters:
+         *         warn_within_days: Window (in days) used to bucket "imminent" models,
+         *             30 by default.
+         *
+         *     Returns:
+         *         A payload with three lists of `ModelDeprecationInfo` entries:
+         *
+         *         - `deprecated`: deprecation date is in the past, so these requests may
+         *           fail at any time.
+         *         - `imminent`: deprecation date is within `warn_within_days` from today.
+         *         - `upcoming`: deprecation date is further out.
+         *
+         *     Example:
+         *     ```shell
+         *     curl -X GET 'http://localhost:4000/model/deprecations' \
+         *         -H 'Authorization: Bearer sk-1234'
+         *     ```
+         */
+        get: operations["model_deprecations_v1_model_deprecations_get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/v1/model/info": {
         parameters: {
             query?: never;
@@ -21540,7 +21624,7 @@ export interface components {
          * @description Enum for alert types and management event types
          * @enum {string}
          */
-        AlertType: "llm_exceptions" | "llm_too_slow" | "llm_requests_hanging" | "budget_alerts" | "spend_reports" | "failed_tracking_spend" | "db_exceptions" | "daily_reports" | "cooldown_deployment" | "new_model_added" | "outage_alerts" | "region_outage_alerts" | "fallback_reports" | "new_virtual_key_created" | "virtual_key_updated" | "virtual_key_deleted" | "new_team_created" | "team_updated" | "team_deleted" | "new_internal_user_created" | "internal_user_updated" | "internal_user_deleted";
+        AlertType: "llm_exceptions" | "llm_too_slow" | "llm_requests_hanging" | "budget_alerts" | "spend_reports" | "failed_tracking_spend" | "db_exceptions" | "daily_reports" | "cooldown_deployment" | "new_model_added" | "model_deprecation_warnings" | "outage_alerts" | "region_outage_alerts" | "fallback_reports" | "new_virtual_key_created" | "virtual_key_updated" | "virtual_key_deleted" | "new_team_created" | "team_updated" | "team_deleted" | "new_internal_user_created" | "internal_user_updated" | "internal_user_deleted";
         /** AllowedVectorStoreIndexItem */
         AllowedVectorStoreIndexItem: {
             /** Index Name */
@@ -29123,6 +29207,70 @@ export interface components {
                 [key: string]: string | string[];
             };
         };
+        /** ModelDeprecationInfo */
+        ModelDeprecationInfo: {
+            /**
+             * Days Until Deprecation
+             * @description Days remaining until the deprecation date. Negative if the model is already deprecated.
+             */
+            days_until_deprecation: number;
+            /**
+             * Deprecation Date
+             * Format: date
+             * @description The date (UTC) when the model becomes deprecated.
+             */
+            deprecation_date: string;
+            /**
+             * Litellm Model
+             * @description The underlying litellm model string the deprecation date is sourced from.
+             */
+            litellm_model?: string | null;
+            /**
+             * Litellm Provider
+             * @description The provider this model belongs to.
+             */
+            litellm_provider?: string | null;
+            /**
+             * Model Name
+             * @description The public name of the model on the proxy (model_group).
+             */
+            model_name: string;
+            /**
+             * Status
+             * @description 'deprecated' if the date has passed, 'imminent' if it falls within warn_within_days, 'upcoming' otherwise.
+             * @enum {string}
+             */
+            status: "upcoming" | "imminent" | "deprecated";
+        };
+        /** ModelDeprecationResponse */
+        ModelDeprecationResponse: {
+            /**
+             * Checked At
+             * Format: date-time
+             * @description UTC timestamp when the deprecation snapshot was generated.
+             */
+            checked_at: string;
+            /**
+             * Deprecated
+             * @description Models whose deprecation date has already passed.
+             */
+            deprecated?: components["schemas"]["ModelDeprecationInfo"][];
+            /**
+             * Imminent
+             * @description Models whose deprecation date is within warn_within_days from today and require immediate migration planning.
+             */
+            imminent?: components["schemas"]["ModelDeprecationInfo"][];
+            /**
+             * Upcoming
+             * @description Models with a future deprecation date outside the warn window.
+             */
+            upcoming?: components["schemas"]["ModelDeprecationInfo"][];
+            /**
+             * Warn Within Days
+             * @description The window (in days) used to bucket 'imminent' models.
+             */
+            warn_within_days: number;
+        };
         /** ModelGroupInfoProxy */
         ModelGroupInfoProxy: {
             /** Configurable Clientside Auth Params */
@@ -46893,6 +47041,37 @@ export interface operations {
             };
         };
     };
+    model_deprecations_model_deprecations_get: {
+        parameters: {
+            query?: {
+                warn_within_days?: number;
+            };
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ModelDeprecationResponse"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     model_info_v1_model_info_get: {
         parameters: {
             query?: {
@@ -58730,6 +58909,37 @@ export interface operations {
             };
         };
     };
+    model_deprecations_v1_model_deprecations_get: {
+        parameters: {
+            query?: {
+                warn_within_days?: number;
+            };
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ModelDeprecationResponse"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     model_info_v1_v1_model_info_get: {
         parameters: {
             query?: {