feat: add scoped daily usage alert API and storage

litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_add_daily_usage_alerts/migration.sql

@@ -0,0 +1,24 @@
+CREATE TABLE "LiteLLM_UsageAlertRule" (
+ "id" TEXT PRIMARY KEY, "scope_type" TEXT NOT NULL, "scope_id" TEXT,
+ "enabled" BOOLEAN NOT NULL DEFAULT true, "record" JSONB NOT NULL,
+ "deleted" BOOLEAN NOT NULL DEFAULT false
+);
+CREATE INDEX "LiteLLM_UsageAlertRule_enabled_deleted_id_idx" ON "LiteLLM_UsageAlertRule" ("enabled", "deleted", "id");
+CREATE INDEX "LiteLLM_UsageAlertRule_scope_type_scope_id_deleted_idx" ON "LiteLLM_UsageAlertRule" ("scope_type", "scope_id", "deleted");
+CREATE TABLE "LiteLLM_UsageAlertIncident" (
+ "id" TEXT PRIMARY KEY, "rule_id" TEXT NOT NULL, "scope_key" TEXT NOT NULL,
+ "scope_type" TEXT NOT NULL, "scope_id" TEXT, "metric" TEXT NOT NULL,
+ "status" TEXT NOT NULL, "day" TEXT NOT NULL, "evaluation_day" TEXT NOT NULL,
+ "breached" BOOLEAN NOT NULL, "record" JSONB NOT NULL,
+ "notified_channels" TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[],
+ "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX "LiteLLM_UsageAlertIncident_rule_id_scope_key_evaluation_day_idx" ON "LiteLLM_UsageAlertIncident" ("rule_id", "scope_key", "evaluation_day");
+CREATE INDEX "LiteLLM_UsageAlertIncident_scope_type_scope_id_day_idx" ON "LiteLLM_UsageAlertIncident" ("scope_type", "scope_id", "day");
+CREATE INDEX "LiteLLM_UsageAlertIncident_status_day_idx" ON "LiteLLM_UsageAlertIncident" ("status", "day");
+CREATE INDEX "LiteLLM_UsageAlertIncident_updated_at_idx" ON "LiteLLM_UsageAlertIncident" ("updated_at");
+CREATE TABLE "LiteLLM_UsageAlertCheckpoint" (
+ "id" TEXT PRIMARY KEY, "completed" BOOLEAN NOT NULL DEFAULT false,
+ "owner" TEXT, "expires_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX "LiteLLM_UsageAlertCheckpoint_expires_at_idx" ON "LiteLLM_UsageAlertCheckpoint" ("expires_at");

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -2109,3 +2109,43 @@ model LiteLLM_SubscriptionAccountTable {
 
   @@unique([custom_llm_provider, account_id], map: "LiteLLM_SubscriptionAccountTable_provider_account_key")
 }
+
+// Daily alerts consume existing aggregate tables; no request-path writes.
+model LiteLLM_UsageAlertRule {
+  id String @id
+  scope_type String
+  scope_id String?
+  enabled Boolean @default(true)
+  record Json
+  deleted Boolean @default(false)
+  @@index([enabled, deleted, id])
+  @@index([scope_type, scope_id, deleted])
+}
+
+model LiteLLM_UsageAlertIncident {
+  id String @id
+  rule_id String
+  scope_key String
+  scope_type String
+  scope_id String?
+  metric String
+  status String
+  day String
+  evaluation_day String
+  breached Boolean
+  record Json
+  notified_channels String[] @default([])
+  updated_at DateTime @default(now())
+  @@index([rule_id, scope_key, evaluation_day])
+  @@index([scope_type, scope_id, day])
+  @@index([status, day])
+  @@index([updated_at])
+}
+
+model LiteLLM_UsageAlertCheckpoint {
+  id String @id
+  completed Boolean @default(false)
+  owner String?
+  expires_at DateTime @default(now())
+  @@index([expires_at])
+}

litellm/proxy/_types.py

@@ -646,6 +646,10 @@ class LiteLLMRoutes(enum.Enum):
 
     management_routes = (
         [
+            "/usage/alert-rules",
+            "/usage/alert-rules/{rule_id}",
+            "/usage/alerts",
+            "/usage/alerts/{alert_id}",
             # user
             "/user/new",
             "/management/v1/users/bulk",
@@ -855,7 +859,15 @@ class LiteLLMRoutes(enum.Enum):
         ]
     )
 
+    usage_alert_routes = [
+        "/usage/alert-rules",
+        "/usage/alert-rules/{rule_id}",
+        "/usage/alerts",
+        "/usage/alerts/{alert_id}",
+    ]
+
     self_managed_routes = [
+        *usage_alert_routes,
         "/lens",
         "/lens/{path:path}",
         "/telemetry/ui_events",

litellm/proxy/alerting_endpoints/usage_alerts.py

@@ -0,0 +1,175 @@
+"""Public management API for daily usage alert rules and incidents."""
+
+from datetime import date
+from typing import Annotated, Final
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.proxy.deps.database import DatabaseDep
+from litellm.repositories.daily_activity_buckets import api_key_from_bucket
+from litellm.repositories.usage_alerts import UsageAlertsRepository
+from litellm.types.proxy.auth.user_roles import LitellmUserRoles
+from litellm.types.proxy.usage_alerts import (
+    UsageAlert,
+    UsageAlertList,
+    UsageAlertMetric,
+    UsageAlertRule,
+    UsageAlertRuleInput,
+    UsageAlertRuleList,
+    UsageAlertScope,
+    UsageAlertStatus,
+    UsageAlertUpdate,
+)
+
+router: Final = APIRouter(prefix="/usage", tags=["usage alerts"])
+
+
+def usage_alert_repository(database: DatabaseDep) -> UsageAlertsRepository:
+    if database is None:
+        raise HTTPException(status_code=503, detail="Usage alerts require a database connection")
+    return UsageAlertsRepository.from_prisma(database.prisma)
+
+
+User = Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)]
+Repository = Annotated[UsageAlertsRepository, Depends(usage_alert_repository)]
+Limit = Annotated[int, Query(ge=1, le=100)]
+Offset = Annotated[int, Query(ge=0, le=100000)]
+ScopeId = Annotated[str | None, Query(min_length=1, max_length=1024)]
+
+
+async def require_scope(
+    user: UserAPIKeyAuth,
+    repository: UsageAlertsRepository,
+    scope_type: UsageAlertScope | None,
+    scope_id: str | None,
+    *,
+    write: bool,
+) -> None:
+    if user.key_type == "llm_api" or (write and user.key_type == "read_only"):
+        raise HTTPException(status_code=403, detail="A management-capable credential is required")
+    if user.user_role == LitellmUserRoles.PROXY_ADMIN:
+        return
+    if user.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY:
+        if not write:
+            return
+        raise HTTPException(status_code=403, detail="Read-only users cannot manage usage alerts")
+    if user.user_role not in (
+        LitellmUserRoles.INTERNAL_USER,
+        LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
+        LitellmUserRoles.ORG_ADMIN,
+        LitellmUserRoles.TEAM,
+    ):
+        raise HTTPException(status_code=403, detail="Usage alerts require a management role")
+    if write and user.user_role == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY:
+        raise HTTPException(status_code=403, detail="Read-only users cannot manage usage alerts")
+    if scope_id is None:
+        raise HTTPException(status_code=403, detail="Specify a scope you can access")
+    if scope_type == "user" and user.user_id is not None and scope_id == user.user_id:
+        return
+    if scope_type == "team":
+        role: Final = await repository.team_role(scope_id, user.user_id) if user.user_id else None
+        if role == "admin" or (
+            not write and (role == "user" or (user.user_role == LitellmUserRoles.TEAM and user.team_id == scope_id))
+        ):
+            return
+    if scope_type == "key":
+        ownership: Final = await repository.key_ownership(api_key_from_bucket(scope_id))
+        if ownership is not None:
+            owner_id, team_id = ownership
+            if user.user_id is not None and owner_id == user.user_id:
+                return
+            if user.user_id and team_id and await repository.team_role(team_id, user.user_id) == "admin":
+                return
+    raise HTTPException(status_code=403, detail="You cannot access this alert scope")
+
+
+@router.get("/alert-rules", response_model=UsageAlertRuleList)
+async def list_usage_alert_rules(
+    user: User,
+    repository: Repository,
+    scope_type: UsageAlertScope | None = None,
+    scope_id: ScopeId = None,
+    limit: Limit = 50,
+    offset: Offset = 0,
+) -> UsageAlertRuleList:
+    await require_scope(user, repository, scope_type, scope_id, write=False)
+    return await repository.list_rules(scope_type, scope_id, limit, offset)
+
+
+@router.post("/alert-rules", response_model=UsageAlertRule, status_code=201)
+async def create_usage_alert_rule(data: UsageAlertRuleInput, user: User, repository: Repository) -> UsageAlertRule:
+    await require_scope(user, repository, data.scope_type, data.scope_id, write=True)
+    return await repository.save_rule(data, user.user_id)
+
+
+@router.get("/alert-rules/{rule_id}", response_model=UsageAlertRule)
+async def get_usage_alert_rule(rule_id: str, user: User, repository: Repository) -> UsageAlertRule:
+    rule: Final = await repository.get_rule(rule_id)
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Alert rule not found")
+    await require_scope(user, repository, rule.scope_type, rule.scope_id, write=False)
+    return rule
+
+
+@router.put("/alert-rules/{rule_id}", response_model=UsageAlertRule)
+async def update_usage_alert_rule(
+    rule_id: str, data: UsageAlertRuleInput, user: User, repository: Repository
+) -> UsageAlertRule:
+    rule: Final = await repository.get_rule(rule_id)
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Alert rule not found")
+    await require_scope(user, repository, rule.scope_type, rule.scope_id, write=True)
+    await require_scope(user, repository, data.scope_type, data.scope_id, write=True)
+    return await repository.save_rule(data, user.user_id, existing=rule)
+
+
+@router.delete("/alert-rules/{rule_id}", status_code=204)
+async def delete_usage_alert_rule(rule_id: str, user: User, repository: Repository) -> Response:
+    rule: Final = await repository.get_rule(rule_id)
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Alert rule not found")
+    await require_scope(user, repository, rule.scope_type, rule.scope_id, write=True)
+    await repository.delete_rule(rule_id)
+    return Response(status_code=204)
+
+
+@router.get("/alerts", response_model=UsageAlertList)
+async def list_usage_alerts(
+    user: User,
+    repository: Repository,
+    scope_type: UsageAlertScope | None = None,
+    scope_id: ScopeId = None,
+    metric: UsageAlertMetric | None = None,
+    status: UsageAlertStatus | None = None,
+    start_date: date | None = None,
+    end_date: date | None = None,
+    limit: Limit = 50,
+    offset: Offset = 0,
+) -> UsageAlertList:
+    await require_scope(user, repository, scope_type, scope_id, write=False)
+    if start_date is not None and end_date is not None and start_date > end_date:
+        raise HTTPException(status_code=422, detail="start_date must not be after end_date")
+    return await repository.list_alerts(scope_type, scope_id, metric, status, start_date, end_date, limit, offset)
+
+
+@router.get("/alerts/{alert_id}", response_model=UsageAlert)
+async def get_usage_alert(alert_id: str, user: User, repository: Repository) -> UsageAlert:
+    alert: Final = await repository.get_alert(alert_id)
+    if alert is None:
+        raise HTTPException(status_code=404, detail="Usage alert not found")
+    await require_scope(user, repository, alert.scope_type, alert.scope_id, write=False)
+    return alert
+
+
+@router.patch("/alerts/{alert_id}", response_model=UsageAlert)
+async def update_usage_alert(alert_id: str, data: UsageAlertUpdate, user: User, repository: Repository) -> UsageAlert:
+    alert: Final = await repository.get_alert(alert_id)
+    if alert is None:
+        raise HTTPException(status_code=404, detail="Usage alert not found")
+    await require_scope(user, repository, alert.scope_type, alert.scope_id, write=True)
+    updated: Final = await repository.set_status(alert_id, data.status)
+    if updated is None:
+        raise HTTPException(status_code=404, detail="Usage alert not found")
+    return updated

litellm/proxy/proxy_server.py

@@ -427,6 +427,7 @@ from litellm.proxy.agent_endpoints.auth.agent_access_groups import CeilingResolv
 from litellm.proxy.alerting_endpoints.model_deprecation_alerts import (
     router as model_deprecation_alerts_router,
 )
+from litellm.proxy.alerting_endpoints.usage_alerts import router as usage_alerts_router
 from litellm.proxy.analytics_endpoints.analytics_endpoints import (
     router as analytics_router,
 )
@@ -22438,6 +22439,7 @@ app.include_router(router_settings_router)
 app.include_router(fallback_management_router)
 app.include_router(cache_settings_router)
 app.include_router(model_deprecation_alerts_router)
+app.include_router(usage_alerts_router)
 app.include_router(coordination_redis_settings_router)
 app.include_router(user_agent_analytics_router)
 app.include_router(gateway_request_router)

litellm/proxy/schema.prisma

@@ -2109,3 +2109,43 @@ model LiteLLM_SubscriptionAccountTable {
 
   @@unique([custom_llm_provider, account_id], map: "LiteLLM_SubscriptionAccountTable_provider_account_key")
 }
+
+// Daily alerts consume existing aggregate tables; no request-path writes.
+model LiteLLM_UsageAlertRule {
+  id String @id
+  scope_type String
+  scope_id String?
+  enabled Boolean @default(true)
+  record Json
+  deleted Boolean @default(false)
+  @@index([enabled, deleted, id])
+  @@index([scope_type, scope_id, deleted])
+}
+
+model LiteLLM_UsageAlertIncident {
+  id String @id
+  rule_id String
+  scope_key String
+  scope_type String
+  scope_id String?
+  metric String
+  status String
+  day String
+  evaluation_day String
+  breached Boolean
+  record Json
+  notified_channels String[] @default([])
+  updated_at DateTime @default(now())
+  @@index([rule_id, scope_key, evaluation_day])
+  @@index([scope_type, scope_id, day])
+  @@index([status, day])
+  @@index([updated_at])
+}
+
+model LiteLLM_UsageAlertCheckpoint {
+  id String @id
+  completed Boolean @default(false)
+  owner String?
+  expires_at DateTime @default(now())
+  @@index([expires_at])
+}

litellm/repositories/usage_alerts.py

@@ -0,0 +1,460 @@
+"""Bounded, writer-consistent storage for daily alert rules and episodes."""
+
+import json
+from datetime import date, datetime, timedelta, timezone
+from typing import TYPE_CHECKING, Final, Protocol, cast
+from urllib.parse import urlencode
+from uuid import NAMESPACE_URL, uuid4, uuid5
+
+from pydantic import BaseModel, TypeAdapter
+
+from litellm.types.proxy.usage_alerts import (
+    UsageAlert,
+    UsageAlertChannel,
+    UsageAlertEvaluationStatus,
+    UsageAlertList,
+    UsageAlertMetric,
+    UsageAlertNotification,
+    UsageAlertObservation,
+    UsageAlertRule,
+    UsageAlertRuleInput,
+    UsageAlertRuleList,
+    UsageAlertScope,
+    UsageAlertStatus,
+)
+
+if TYPE_CHECKING:
+    from litellm.proxy.utils import PrismaClient
+
+
+class AlertDatabase(Protocol):
+    async def query_raw(self, query: str, *args: object) -> object: ...
+
+
+class _Record(BaseModel):
+    record: dict[str, object]
+
+
+class _Count(BaseModel):
+    count: int
+
+
+class _Episode(_Record):
+    evaluation_day: date
+    breached: bool
+
+
+class _Pending(_Record):
+    rule: dict[str, object]
+    notified_channels: tuple[UsageAlertChannel, ...]
+
+
+class _Ownership(BaseModel):
+    user_id: str | None
+    team_id: str | None
+
+
+def activity_path(
+    scope: UsageAlertScope, entity_id: str | None, day: date, alert_id: str, baseline_days: int = 7
+) -> str:
+    routes: Final = {
+        "key": ("/api-keys", "key"),
+        "team": ("/teams", "team"),
+        "user": ("/users", "user"),
+        "model": ("/models-and-endpoints", "activity_model"),
+        "deployment": ("/usage", ""),
+    }
+    route, parameter = routes[scope]
+    if scope == "deployment":
+        return (
+            route
+            + "?"
+            + urlencode(
+                {
+                    "range": "custom",
+                    "from": (day - timedelta(days=baseline_days)).isoformat(),
+                    "to": day.isoformat(),
+                    "alert": alert_id,
+                }
+            )
+        )
+    query: Final = {
+        "activity_from": (day - timedelta(days=baseline_days)).isoformat(),
+        "activity_to": day.isoformat(),
+        "alert": alert_id,
+    }
+    if parameter and entity_id and not entity_id.startswith("sk-"):
+        query[parameter] = entity_id
+        query["detail_tab"] = "activity"
+    return route + "?" + urlencode(query)
+
+
+class UsageAlertsRepository:
+    def __init__(self, database: AlertDatabase):
+        self.database: Final = database
+
+    @classmethod
+    def from_prisma(cls, prisma: "PrismaClient") -> "UsageAlertsRepository":
+        return cls(cast(AlertDatabase, prisma.writer_db))
+
+    async def get_rule(self, rule_id: str) -> UsageAlertRule | None:
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE id=$1 AND NOT deleted', rule_id
+            )
+        )
+        return UsageAlertRule.model_validate(rows[0].record) if rows else None
+
+    async def save_rule(
+        self, data: UsageAlertRuleInput, created_by: str | None, existing: UsageAlertRule | None = None
+    ) -> UsageAlertRule:
+        now: Final = datetime.now(timezone.utc)
+        rule: Final = UsageAlertRule(
+            **data.model_dump(),
+            id=existing.id if existing else str(uuid4()),
+            created_by=existing.created_by if existing else created_by,
+            created_at=existing.created_at if existing else now,
+            updated_at=now,
+        )
+        rows: Final = await self.database.query_raw(
+            """INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb)
+            ON CONFLICT (id) DO UPDATE SET scope_type=EXCLUDED.scope_type,scope_id=EXCLUDED.scope_id,enabled=EXCLUDED.enabled,record=EXCLUDED.record
+            WHERE NOT "LiteLLM_UsageAlertRule".deleted RETURNING record""",
+            rule.id,
+            rule.scope_type,
+            rule.scope_id,
+            rule.enabled,
+            rule.model_dump_json(),
+        )
+        return UsageAlertRule.model_validate(TypeAdapter(tuple[_Record, ...]).validate_python(rows)[0].record)
+
+    async def delete_rule(self, rule_id: str) -> None:
+        await self.database.query_raw(
+            'UPDATE "LiteLLM_UsageAlertRule" SET deleted=true,enabled=false WHERE id=$1 RETURNING id', rule_id
+        )
+
+    async def list_rules(
+        self, scope_type: UsageAlertScope | None = None, scope_id: str | None = None, limit: int = 50, offset: int = 0
+    ) -> UsageAlertRuleList:
+        args: Final = (scope_type, scope_id)
+        where: Final = "NOT deleted AND ($1::text IS NULL OR scope_type=$1) AND ($2::text IS NULL OR scope_id=$2)"
+        count: Final = TypeAdapter(tuple[_Count, ...]).validate_python(
+            await self.database.query_raw(
+                f'SELECT count(*)::int AS count FROM "LiteLLM_UsageAlertRule" WHERE {where}', *args
+            )
+        )
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                f'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE {where} ORDER BY id LIMIT $3 OFFSET $4',
+                *args,
+                limit,
+                offset,
+            )
+        )
+        return UsageAlertRuleList(
+            data=tuple(UsageAlertRule.model_validate(row.record) for row in rows),
+            total=count[0].count,
+            limit=limit,
+            offset=offset,
+        )
+
+    async def list_enabled_rules(self, limit: int = 100, offset: int = 0) -> tuple[UsageAlertRule, ...]:
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE enabled AND NOT deleted ORDER BY id LIMIT $1 OFFSET $2',
+                limit,
+                offset,
+            )
+        )
+        return tuple(UsageAlertRule.model_validate(row.record) for row in rows)
+
+    async def get_alert(self, alert_id: str) -> UsageAlert | None:
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw('SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE id=$1', alert_id)
+        )
+        return UsageAlert.model_validate(rows[0].record) if rows else None
+
+    async def list_alerts(
+        self,
+        scope_type: UsageAlertScope | None = None,
+        scope_id: str | None = None,
+        metric: UsageAlertMetric | None = None,
+        status: UsageAlertStatus | None = None,
+        start_date: date | None = None,
+        end_date: date | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> UsageAlertList:
+        args: Final = (
+            scope_type,
+            scope_id,
+            metric,
+            status,
+            start_date.isoformat() if start_date else None,
+            end_date.isoformat() if end_date else None,
+        )
+        where: Final = "($1::text IS NULL OR scope_type=$1) AND ($2::text IS NULL OR scope_id=$2) AND ($3::text IS NULL OR metric=$3) AND ($4::text IS NULL OR status=$4) AND ($5::text IS NULL OR day >= $5) AND ($6::text IS NULL OR day <= $6)"
+        count: Final = TypeAdapter(tuple[_Count, ...]).validate_python(
+            await self.database.query_raw(
+                f'SELECT count(*)::int AS count FROM "LiteLLM_UsageAlertIncident" WHERE {where}', *args
+            )
+        )
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                f'SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE {where} ORDER BY day DESC,id LIMIT $7 OFFSET $8',
+                *args,
+                limit,
+                offset,
+            )
+        )
+        return UsageAlertList(
+            data=tuple(UsageAlert.model_validate(row.record) for row in rows),
+            total=count[0].count,
+            limit=limit,
+            offset=offset,
+        )
+
+    async def set_status(self, alert_id: str, status: UsageAlertStatus) -> UsageAlert | None:
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                """UPDATE "LiteLLM_UsageAlertIncident" SET status=$2, record=jsonb_set(record,'{status}',to_jsonb($2::text)),updated_at=CURRENT_TIMESTAMP WHERE id=$1 RETURNING record""",
+                alert_id,
+                status,
+            )
+        )
+        return UsageAlert.model_validate(rows[0].record) if rows else None
+
+    async def key_ownership(self, key_id: str) -> tuple[str | None, str | None] | None:
+        rows: Final = TypeAdapter(tuple[_Ownership, ...]).validate_python(
+            await self.database.query_raw(
+                'SELECT user_id,team_id FROM "LiteLLM_VerificationToken" WHERE token=$1', key_id
+            )
+        )
+        return (rows[0].user_id, rows[0].team_id) if rows else None
+
+    async def claim_job(self, owner: str, lease_seconds: int = 900) -> bool:
+        rows: Final = TypeAdapter(tuple[_Count, ...]).validate_python(
+            await self.database.query_raw(
+                """INSERT INTO "LiteLLM_UsageAlertCheckpoint" (id,owner,expires_at) VALUES ('worker',$1,CURRENT_TIMESTAMP+($2::int * INTERVAL '1 second'))
+            ON CONFLICT(id) DO UPDATE SET owner=EXCLUDED.owner,expires_at=EXCLUDED.expires_at WHERE "LiteLLM_UsageAlertCheckpoint".expires_at<CURRENT_TIMESTAMP OR "LiteLLM_UsageAlertCheckpoint".owner=$1 RETURNING 1 AS count""",
+                owner,
+                lease_seconds,
+            )
+        )
+        return bool(rows)
+
+    async def release_job(self, owner: str) -> None:
+        await self.database.query_raw(
+            "DELETE FROM \"LiteLLM_UsageAlertCheckpoint\" WHERE id='worker' AND owner=$1 RETURNING id", owner
+        )
+
+    @staticmethod
+    def _checkpoint(rule: UsageAlertRule, day: date) -> str:
+        return f"{rule.id}:{rule.updated_at.isoformat()}:{day.isoformat()}"
+
+    async def is_day_complete(self, rule: UsageAlertRule, day: date) -> bool:
+        rows: Final = TypeAdapter(tuple[_Count, ...]).validate_python(
+            await self.database.query_raw(
+                'SELECT 1 AS count FROM "LiteLLM_UsageAlertCheckpoint" WHERE id=$1 AND completed',
+                self._checkpoint(rule, day),
+            )
+        )
+        return bool(rows)
+
+    async def finish_day(self, rule: UsageAlertRule, day: date) -> None:
+        await self.database.query_raw(
+            'INSERT INTO "LiteLLM_UsageAlertCheckpoint" (id,completed) VALUES ($1,true) ON CONFLICT(id) DO NOTHING RETURNING id',
+            self._checkpoint(rule, day),
+        )
+
+    async def ensure_default_rules(self, rules: tuple[UsageAlertRuleInput, ...]) -> None:
+        for rule in rules:
+            rule_id: Final = str(
+                uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.scope_type}:{rule.metric}:{rule.direction}")
+            )
+            now: Final = datetime.now(timezone.utc)
+            record: Final = UsageAlertRule(
+                **rule.model_dump(), id=rule_id, created_at=now, updated_at=now, created_by=None
+            )
+            await self.database.query_raw(
+                'INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb) ON CONFLICT(id) DO NOTHING RETURNING id',
+                rule_id,
+                rule.scope_type,
+                rule.scope_id,
+                rule.enabled,
+                record.model_dump_json(),
+            )
+
+    async def list_breached_entity_ids(self, rule_id: str, limit: int = 500, offset: int = 0) -> tuple[str | None, ...]:
+        rows: Final = TypeAdapter(tuple[dict[str, str | None], ...]).validate_python(
+            await self.database.query_raw(
+                'SELECT DISTINCT scope_id FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND breached ORDER BY scope_id NULLS FIRST LIMIT $2 OFFSET $3',
+                rule_id,
+                limit,
+                offset,
+            )
+        )
+        return tuple(row["scope_id"] for row in rows)
+
+    async def record_evaluation_status(
+        self, rule: UsageAlertRule, day: date, status: UsageAlertEvaluationStatus
+    ) -> None:
+        await self.database.query_raw(
+            """UPDATE "LiteLLM_UsageAlertRule" SET record=record || jsonb_build_object('last_evaluated_day',$2::text,'evaluation_status',$3::text) WHERE id=$1 AND record->>'updated_at'=$4 AND NOT deleted RETURNING id""",
+            rule.id,
+            day.isoformat(),
+            status,
+            rule.model_dump(mode="json")["updated_at"],
+        )
+
+    async def record_observation(
+        self,
+        rule: UsageAlertRule,
+        entity_id: str | None,
+        day: date,
+        observed: float,
+        baseline: float,
+        threshold: float,
+        breached: bool,
+    ) -> UsageAlert | None:
+        observation: Final = UsageAlertObservation(
+            entity_id=entity_id, day=day, observed=observed, baseline=baseline, threshold=threshold, breached=breached
+        )
+        changed: Final = await self.record_observations(rule, (observation,))
+        if changed:
+            return changed[0]
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                'SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND scope_key=$2 ORDER BY evaluation_day DESC,id DESC LIMIT 1',
+                rule.id,
+                entity_id or "",
+            )
+        )
+        return UsageAlert.model_validate(rows[0].record) if rows else None
+
+    async def record_observations(
+        self, rule: UsageAlertRule, observations: tuple[UsageAlertObservation, ...]
+    ) -> tuple[UsageAlert, ...]:
+        if not observations:
+            return ()
+        if len(observations) > 500 or len({item.entity_id for item in observations}) != len(observations):
+            raise ValueError("An observation batch must contain at most 500 distinct entities")
+        now: Final = datetime.now(timezone.utc)
+        records: Final = []
+        for item in observations:
+            alert_id: Final = str(
+                uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.id}:{item.entity_id or ''}:{item.day.isoformat()}")
+            )
+            incident: Final = UsageAlert(
+                id=alert_id,
+                rule_id=rule.id,
+                rule_name=rule.name,
+                scope_type=rule.scope_type,
+                scope_id=item.entity_id,
+                metric=rule.metric,
+                direction=rule.direction,
+                status="open",
+                day=item.day,
+                baseline_start=item.day - timedelta(days=rule.baseline_days),
+                baseline_end=item.day - timedelta(days=1),
+                observed_value=item.observed,
+                baseline_value=item.baseline,
+                threshold=item.threshold,
+                first_seen=now,
+                last_seen=now,
+                activity_path=activity_path(rule.scope_type, item.entity_id, item.day, alert_id, rule.baseline_days),
+            )
+            records.append(
+                {
+                    "scope_key": item.entity_id or "",
+                    "breached": item.breached,
+                    "day": item.day.isoformat(),
+                    "record": incident.model_dump(mode="json"),
+                }
+            )
+        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
+            await self.database.query_raw(
+                """
+            WITH input AS (SELECT * FROM jsonb_to_recordset($2::jsonb) AS x(scope_key text, breached boolean, day text, record jsonb)),
+            prior AS (
+                SELECT x.*,p.id AS prior_id,p.evaluation_day,p.breached AS prior_breached,p.record AS prior_record
+                FROM input x LEFT JOIN LATERAL (
+                    SELECT * FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND scope_key=x.scope_key ORDER BY evaluation_day DESC,id DESC LIMIT 1
+                ) p ON true
+                WHERE EXISTS (SELECT 1 FROM "LiteLLM_UsageAlertRule" WHERE id=$1 AND enabled AND NOT deleted AND record->>'updated_at'=$3)
+            ), updated AS (
+                UPDATE "LiteLLM_UsageAlertIncident" i SET
+                    evaluation_day=p.day, breached=p.breached,
+                    day=CASE WHEN p.breached THEN p.day ELSE i.day END,
+                    record=(CASE WHEN p.breached THEN p.record || jsonb_build_object(
+                        'id',i.id,'first_seen',i.record->'first_seen',
+                        'activity_path',replace(p.record->>'activity_path',p.record->>'id',i.id)) ELSE i.record END)
+                        || jsonb_build_object('status',CASE WHEN i.status='open' AND NOT p.breached THEN 'resolved' ELSE i.status END),
+                    status=CASE WHEN i.status='open' AND NOT p.breached THEN 'resolved' ELSE i.status END,
+                    updated_at=CURRENT_TIMESTAMP
+                FROM prior p WHERE i.id=p.prior_id AND p.evaluation_day<p.day AND (NOT p.breached OR p.prior_breached)
+                RETURNING i.record
+            ), inserted AS (
+                INSERT INTO "LiteLLM_UsageAlertIncident" (id,rule_id,scope_key,scope_type,scope_id,metric,status,day,evaluation_day,breached,record)
+                SELECT p.record->>'id',$1,p.scope_key,p.record->>'scope_type',p.record->>'scope_id',p.record->>'metric','open',p.day,p.day,true,p.record
+                FROM prior p WHERE p.breached AND (p.prior_id IS NULL OR (NOT p.prior_breached AND p.evaluation_day<p.day))
+                ON CONFLICT(id) DO NOTHING RETURNING record
+            ) SELECT record FROM updated UNION ALL SELECT record FROM inserted
+            """,
+                rule.id,
+                json.dumps(records),
+                rule.model_dump(mode="json")["updated_at"],
+            )
+        )
+        return tuple(UsageAlert.model_validate(row.record) for row in rows)
+
+    async def pending_notifications(self, limit: int = 100) -> tuple[UsageAlertNotification, ...]:
+        rows: Final = TypeAdapter(tuple[_Pending, ...]).validate_python(
+            await self.database.query_raw(
+                """SELECT i.record,r.record AS rule,i.notified_channels FROM "LiteLLM_UsageAlertIncident" i JOIN "LiteLLM_UsageAlertRule" r ON r.id=i.rule_id
+            WHERE i.status='open' AND r.enabled AND NOT r.deleted AND EXISTS (
+                SELECT 1 FROM jsonb_array_elements_text(r.record->'notification_channels') channel WHERE NOT (channel=ANY(i.notified_channels)))
+            ORDER BY i.updated_at,i.id LIMIT $1""",
+                limit,
+            )
+        )
+        return tuple(
+            UsageAlertNotification(
+                incident=UsageAlert.model_validate(row.record),
+                rule=UsageAlertRule.model_validate(row.rule),
+                pending_channels=tuple(
+                    channel
+                    for channel in UsageAlertRule.model_validate(row.rule).notification_channels
+                    if channel not in row.notified_channels
+                ),
+            )
+            for row in rows
+        )
+
+    async def mark_notified(self, incident_id: str, channel: UsageAlertChannel) -> None:
+        await self.database.query_raw(
+            'UPDATE "LiteLLM_UsageAlertIncident" SET notified_channels=array_append(notified_channels,$2::text) WHERE id=$1 AND NOT ($2::text=ANY(notified_channels)) RETURNING id',
+            incident_id,
+            channel,
+        )
+
+    async def prune(self, retention_days: int = 90, limit: int = 500) -> None:
+        await self.database.query_raw(
+            """DELETE FROM "LiteLLM_UsageAlertIncident" WHERE id IN (SELECT id FROM "LiteLLM_UsageAlertIncident" WHERE status<>'open' AND updated_at<CURRENT_TIMESTAMP-($1::int * INTERVAL '1 day') ORDER BY updated_at LIMIT $2) RETURNING id""",
+            retention_days,
+            limit,
+        )
+        await self.database.query_raw(
+            """DELETE FROM "LiteLLM_UsageAlertCheckpoint" WHERE id IN (SELECT id FROM "LiteLLM_UsageAlertCheckpoint" WHERE completed AND expires_at<CURRENT_TIMESTAMP-INTERVAL '45 days' ORDER BY expires_at LIMIT $1) RETURNING id""",
+            limit,
+        )
+
+    async def team_role(self, team_id: str, user_id: str) -> str | None:
+        rows: Final = TypeAdapter(tuple[dict[str, str], ...]).validate_python(
+            await self.database.query_raw(
+                """SELECT member->>'role' AS role FROM "LiteLLM_TeamTable", jsonb_array_elements(members_with_roles) member WHERE team_id=$1 AND member->>'user_id'=$2 LIMIT 1""",
+                team_id,
+                user_id,
+            )
+        )
+        return rows[0]["role"] if rows else None

litellm/types/proxy/usage_alerts.py

@@ -0,0 +1,116 @@
+"""Daily usage alert rules and incidents shared by the API and evaluator."""
+
+from datetime import date, datetime
+from typing import Literal, Self
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+UsageAlertScope = Literal["deployment", "key", "user", "team", "model"]
+UsageAlertMetric = Literal["spend", "requests", "latency", "errors"]
+UsageAlertDirection = Literal["increase", "decrease"]
+UsageAlertStatus = Literal["open", "resolved", "dismissed"]
+UsageAlertChannel = Literal["email", "slack"]
+UsageAlertEvaluationStatus = Literal["pending", "complete", "insufficient_coverage", "insufficient_history"]
+
+
+class UsageAlertRuleInput(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", allow_inf_nan=False)
+
+    name: str = Field(min_length=1, max_length=200)
+    scope_type: UsageAlertScope
+    scope_id: str | None = Field(default=None, min_length=1, max_length=1024)
+    metric: UsageAlertMetric
+    direction: UsageAlertDirection
+    multiplier: float = Field(default=3, gt=1, le=1000)
+    baseline_days: int = Field(default=7, ge=1, le=30)
+    min_history_days: int = Field(default=3, ge=1, le=30)
+    min_requests: int = Field(default=100, ge=0)
+    min_value: float = Field(default=0, ge=0, description="Minimum absolute change in the metric's units")
+    enabled: bool = True
+    notification_channels: tuple[UsageAlertChannel, ...] = Field(default=(), max_length=2)
+    email_recipients: tuple[str, ...] = Field(default=(), max_length=20)
+
+    @model_validator(mode="after")
+    def validate_rule(self) -> Self:
+        if self.min_history_days > self.baseline_days:
+            raise ValueError("min_history_days must not exceed baseline_days")
+        if self.scope_type == "deployment" and self.scope_id is not None:
+            raise ValueError("Deployment alerts cover the whole gateway; omit scope_id")
+        if self.scope_type == "key" and self.scope_id is not None and self.scope_id.startswith("sk-"):
+            raise ValueError("Use the key's reporting ID, never the secret key")
+        if len(set(self.notification_channels)) != len(self.notification_channels):
+            raise ValueError("notification_channels must be unique")
+        if "email" in self.notification_channels and not self.email_recipients:
+            raise ValueError("Email notifications require at least one recipient")
+        if any(
+            "@" not in email or "\n" in email or "\r" in email or len(email) > 254 for email in self.email_recipients
+        ):
+            raise ValueError("Invalid email recipient")
+        return self
+
+
+class UsageAlertRule(UsageAlertRuleInput):
+    id: str
+    created_at: datetime
+    updated_at: datetime
+    created_by: str | None
+    last_evaluated_day: date | None = None
+    evaluation_status: UsageAlertEvaluationStatus = "pending"
+
+
+class UsageAlert(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", allow_inf_nan=False)
+
+    id: str
+    rule_id: str
+    rule_name: str
+    scope_type: UsageAlertScope
+    scope_id: str | None
+    metric: UsageAlertMetric
+    direction: UsageAlertDirection
+    status: UsageAlertStatus
+    day: date
+    baseline_start: date
+    baseline_end: date
+    observed_value: float
+    baseline_value: float
+    threshold: float
+    first_seen: datetime
+    last_seen: datetime
+    activity_path: str
+
+
+class UsageAlertUpdate(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+    status: UsageAlertStatus
+
+
+class UsageAlertRuleList(BaseModel):
+    data: tuple[UsageAlertRule, ...]
+    total: int
+    limit: int
+    offset: int
+
+
+class UsageAlertList(BaseModel):
+    data: tuple[UsageAlert, ...]
+    total: int
+    limit: int
+    offset: int
+
+
+class UsageAlertNotification(BaseModel):
+    model_config = ConfigDict(frozen=True)
+    incident: UsageAlert
+    rule: UsageAlertRule
+    pending_channels: tuple[UsageAlertChannel, ...]
+
+
+class UsageAlertObservation(BaseModel):
+    model_config = ConfigDict(frozen=True, allow_inf_nan=False)
+    entity_id: str | None
+    day: date
+    observed: float
+    baseline: float
+    threshold: float
+    breached: bool

schema.prisma

@@ -2109,3 +2109,43 @@ model LiteLLM_SubscriptionAccountTable {
 
   @@unique([custom_llm_provider, account_id], map: "LiteLLM_SubscriptionAccountTable_provider_account_key")
 }
+
+// Daily alerts consume existing aggregate tables; no request-path writes.
+model LiteLLM_UsageAlertRule {
+  id String @id
+  scope_type String
+  scope_id String?
+  enabled Boolean @default(true)
+  record Json
+  deleted Boolean @default(false)
+  @@index([enabled, deleted, id])
+  @@index([scope_type, scope_id, deleted])
+}
+
+model LiteLLM_UsageAlertIncident {
+  id String @id
+  rule_id String
+  scope_key String
+  scope_type String
+  scope_id String?
+  metric String
+  status String
+  day String
+  evaluation_day String
+  breached Boolean
+  record Json
+  notified_channels String[] @default([])
+  updated_at DateTime @default(now())
+  @@index([rule_id, scope_key, evaluation_day])
+  @@index([scope_type, scope_id, day])
+  @@index([status, day])
+  @@index([updated_at])
+}
+
+model LiteLLM_UsageAlertCheckpoint {
+  id String @id
+  completed Boolean @default(false)
+  owner String?
+  expires_at DateTime @default(now())
+  @@index([expires_at])
+}