feat(spend): add a per-session auto-router benchmarks rollup

litellm-proxy-extras/litellm_proxy_extras/migrations/20260804000000_add_auto_router_session_rollup/migration.sql

@@ -0,0 +1,37 @@
+CREATE TABLE IF NOT EXISTS "LiteLLM_AutoRouterSession" (
+    api_key            TEXT NOT NULL,
+    session_id         TEXT NOT NULL,
+    model_group        TEXT NOT NULL,
+    router_kind        TEXT NOT NULL,
+    baseline_model     TEXT,
+    turns              INTEGER NOT NULL DEFAULT 0,
+    turns_with_usage   INTEGER NOT NULL DEFAULT 0,
+    total_tokens       BIGINT NOT NULL DEFAULT 0,
+    spend              DOUBLE PRECISION NOT NULL DEFAULT 0,
+    baseline_spend     DOUBLE PRECISION NOT NULL DEFAULT 0,
+    first_visit_turns  INTEGER NOT NULL DEFAULT 0,
+    first_visit_hits   INTEGER NOT NULL DEFAULT 0,
+    warm_turns         INTEGER NOT NULL DEFAULT 0,
+    warm_hits          INTEGER NOT NULL DEFAULT 0,
+    expired_turns      INTEGER NOT NULL DEFAULT 0,
+    expired_hits       INTEGER NOT NULL DEFAULT 0,
+    unordered_turns    INTEGER NOT NULL DEFAULT 0,
+    unordered_hits     INTEGER NOT NULL DEFAULT 0,
+    unknown_ttl_turns        INTEGER NOT NULL DEFAULT 0,
+    unknown_ttl_hits         INTEGER NOT NULL DEFAULT 0,
+    cache_5m_turns           INTEGER NOT NULL DEFAULT 0,
+    cache_1h_turns           INTEGER NOT NULL DEFAULT 0,
+    cache_ttl_unknown_turns INTEGER NOT NULL DEFAULT 0,
+    tiers              JSONB NOT NULL DEFAULT '{}',
+    first_turn_at      TIMESTAMP(3) NOT NULL,
+    last_turn_at       TIMESTAMP(3) NOT NULL,
+    updated_at         TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "LiteLLM_AutoRouterSession_pkey" PRIMARY KEY (api_key, session_id, model_group)
+);
+
+CREATE INDEX IF NOT EXISTS "idx_auto_router_session_started"
+    ON "LiteLLM_AutoRouterSession" (first_turn_at);
+
+CREATE INDEX IF NOT EXISTS "idx_auto_router_session_activity"
+    ON "LiteLLM_AutoRouterSession" (last_turn_at);

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -1393,6 +1393,45 @@ model LiteLLM_AdaptiveRouterSession {
   @@index([last_activity_at], map: "idx_adaptive_router_session_activity")
 }
 
+model LiteLLM_AutoRouterSession {
+  api_key            String
+  session_id         String
+  model_group        String
+  router_kind        String
+  baseline_model     String?
+
+  turns              Int      @default(0)
+  turns_with_usage   Int      @default(0)
+  total_tokens       BigInt   @default(0)
+  spend              Float    @default(0)
+  baseline_spend     Float    @default(0)
+
+  first_visit_turns  Int      @default(0)
+  first_visit_hits   Int      @default(0)
+  warm_turns         Int      @default(0)
+  warm_hits          Int      @default(0)
+  expired_turns      Int      @default(0)
+  expired_hits       Int      @default(0)
+  unordered_turns    Int      @default(0)
+  unordered_hits     Int      @default(0)
+  unknown_ttl_turns  Int      @default(0)
+  unknown_ttl_hits   Int      @default(0)
+
+  cache_5m_turns          Int @default(0)
+  cache_1h_turns          Int @default(0)
+  cache_ttl_unknown_turns Int @default(0)
+
+  tiers              Json     @default("{}")
+
+  first_turn_at      DateTime
+  last_turn_at       DateTime
+  updated_at         DateTime @default(now()) @updatedAt
+
+  @@id([api_key, session_id, model_group])
+  @@index([first_turn_at], map: "idx_auto_router_session_started")
+  @@index([last_turn_at], map: "idx_auto_router_session_activity")
+}
+
 // ---------------------------------------------------------------------------
 // Workflow Run Tracking
 //

litellm/proxy/db/db_spend_update_writer.py

@@ -51,6 +51,7 @@
     ToolDiscoveryQueue,
 )
 from litellm.proxy.route_llm_request import ROUTE_ENDPOINT_MAPPING
+from litellm.proxy.spend_tracking.auto_router_sessions import AutoRouterSessionQueue, build_turn_facts
 from litellm.proxy.spend_tracking.compression_savings import (
     extract_compression_saved_tokens,
 )
@@ -132,6 +133,46 @@ def __init__(
         self.daily_agent_spend_update_queue = DailySpendUpdateQueue()
         self.daily_org_spend_update_queue = DailySpendUpdateQueue()
         self.daily_tag_spend_update_queue = DailySpendUpdateQueue()
+        self.auto_router_session_queue = AutoRouterSessionQueue()
+
+    def _enqueue_auto_router_turn(self, payload: SpendLogsPayload) -> None:
+        """Stage one auto-routed turn for the benchmarks rollup.
+
+        Runs once per request rather than beside the daily transactions, which are built
+        per entity type and would count every turn six times over. Independent of
+        ``disable_spend_logs``, because the rollup is what the benchmarks dashboard reads
+        now. Never raises, and never blocks: a full queue drops the turn rather than
+        applying backpressure to spend tracking.
+        """
+        try:
+            metadata: Final[SpendLogsMetadata] = json.loads(payload["metadata"])
+            if not metadata.get("routing_decision"):
+                return
+            usage_raw: Final = metadata.get("usage_object")
+            usage_obj: Final = usage_raw if isinstance(usage_raw, dict) else None
+            cache_read_tokens: Final = _extract_cache_read_tokens(usage_obj) if usage_obj is not None else 0
+            savings: Final = compute_savings_spend(
+                model=payload.get("model", None),
+                custom_llm_provider=payload.get("custom_llm_provider", None),
+                compression_saved_tokens=extract_compression_saved_tokens(metadata),
+                cache_read_input_tokens=cache_read_tokens,
+                routing_decision=metadata.get("routing_decision"),
+                model_id=payload.get("model_id"),
+                llm_router=_get_llm_router,
+                usage_object=usage_obj,
+                cost_breakdown=metadata.get("cost_breakdown"),
+            )
+            turn: Final = build_turn_facts(
+                payload=payload,
+                metadata=metadata,
+                autorouter_savings=savings.autorouter,
+                cache_read_tokens=cache_read_tokens,
+                cache_creation_tokens=_extract_cache_creation_tokens(usage_obj) if usage_obj is not None else 0,
+            )
+            if turn is not None:
+                self.auto_router_session_queue.update_queue.put_nowait(turn)
+        except Exception as e:  # noqa: BLE001  # a dashboard rollup must never fail spend tracking
+            verbose_proxy_logger.debug("auto_router_sessions: could not stage turn (%s)", e)
 
     async def update_database(
         # LiteLLM management object fields
@@ -209,6 +250,8 @@ async def update_database(
                     "disable_spend_logs=True. Skipping writing spend logs to db. Other spend updates - Key/User/Team table will still occur."
                 )
 
+            self._enqueue_auto_router_turn(payload=payload)
+
             # Single task replaces 11 create_task() calls
             asyncio.create_task(
                 self._batch_database_updates(

litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py

@@ -15,6 +15,9 @@
 from litellm.proxy.db.db_transaction_queue.spend_logs_partition_manager import (
     SpendLogsPartitionManager,
 )
+from litellm.proxy.spend_tracking.auto_router_benchmarks import (
+    AUTO_ROUTER_SESSION_RETENTION_DAYS,
+)
 from litellm.proxy.utils import PrismaClient
 
 
@@ -186,22 +189,42 @@ async def _delete_old_tool_index_rows(self, prisma_client: PrismaClient, cutoff_
             time_column="start_time",
         )
 
+    async def _delete_old_auto_router_sessions(self, prisma_client: PrismaClient, cutoff_date: datetime) -> int:
+        """Expire auto-router rollups past the given cutoff.
+
+        Keyed on last activity rather than session start, so a conversation still running
+        when the cutoff passes is not pruned out from under itself.
+        """
+        return await self._delete_old_rows_batched(
+            prisma_client,
+            cutoff_date,
+            table_name="LiteLLM_AutoRouterSession",
+            key_columns=("api_key", "session_id", "model_group"),
+            time_column="last_turn_at",
+        )
+
+    def _auto_router_session_cutoff(self, retention_seconds: float | None) -> datetime:
+        """Rows the benchmarks endpoint can no longer read are collected at the read
+        horizon even with no retention configured; a shorter configured retention wins."""
+        now: Final = datetime.now(timezone.utc)
+        horizon: Final = now - timedelta(days=AUTO_ROUTER_SESSION_RETENTION_DAYS)
+        return horizon if retention_seconds is None else max(horizon, now - timedelta(seconds=retention_seconds))
+
     async def cleanup_old_spend_logs(self, prisma_client: PrismaClient) -> None:
         """
         Main cleanup function. Deletes old spend logs in batches.
         If pod_lock_manager is available, ensures only one pod runs cleanup.
         If no pod_lock_manager, runs cleanup without distributed locking.
+
+        Spend logs and their tool index only expire when
+        ``maximum_spend_logs_retention_period`` is configured. The auto-router rollup is
+        collected on every run regardless; see ``_auto_router_session_cutoff``.
         """
         lock_acquired = False
         try:
             verbose_proxy_logger.info("Cleanup job triggered at %s", datetime.now())
 
-            if not self._should_delete_spend_logs():
-                return
-
-            if self.retention_seconds is None:
-                verbose_proxy_logger.error("Retention seconds is None, cannot proceed with cleanup")
-                return
+            retention_seconds: Final = self.retention_seconds if self._should_delete_spend_logs() else None
 
             # If we have a pod lock manager, try to acquire the lock
             if self.pod_lock_manager and self.pod_lock_manager.redis_cache:
@@ -219,31 +242,39 @@ async def cleanup_old_spend_logs(self, prisma_client: PrismaClient) -> None:
                     verbose_proxy_logger.info("Another pod is already running cleanup")
                     return
 
-            cutoff_date: Final = datetime.now(timezone.utc) - timedelta(seconds=float(self.retention_seconds))
-            verbose_proxy_logger.info("Removing logs older than %s", cutoff_date.isoformat())
+            if retention_seconds is not None:
+                cutoff_date: Final = datetime.now(timezone.utc) - timedelta(seconds=float(retention_seconds))
+                verbose_proxy_logger.info("Removing logs older than %s", cutoff_date.isoformat())
 
-            if self.general_settings.get(
-                "use_spend_logs_partitioning", False
-            ) and await self.partition_manager.is_partitioned(prisma_client):
-                await self.partition_manager.ensure_partitions(prisma_client)
-                dropped: Final = await self.partition_manager.drop_partitions_older_than(prisma_client, cutoff_date)
-                verbose_proxy_logger.info(
-                    "Dropped %d expired spend-log partitions: %s",
-                    len(dropped),
-                    dropped,
-                )
-                # DROP only reclaims whole expired partitions. Expired rows can
-                # still sit in the DEFAULT partition (backfill, coverage gaps)
-                # or in a partition that spans the cutoff, so retention must
-                # also delete those stragglers row-wise.
-                total_deleted = await self._delete_old_logs(prisma_client, cutoff_date)
-                verbose_proxy_logger.info("Deleted %s expired logs not covered by dropped partitions", total_deleted)
-            else:
-                total_deleted = await self._delete_old_logs(prisma_client, cutoff_date)
-                verbose_proxy_logger.info("Deleted %s logs", total_deleted)
+                if self.general_settings.get(
+                    "use_spend_logs_partitioning", False
+                ) and await self.partition_manager.is_partitioned(prisma_client):
+                    await self.partition_manager.ensure_partitions(prisma_client)
+                    dropped: Final = await self.partition_manager.drop_partitions_older_than(prisma_client, cutoff_date)
+                    verbose_proxy_logger.info(
+                        "Dropped %d expired spend-log partitions: %s",
+                        len(dropped),
+                        dropped,
+                    )
+                    # DROP only reclaims whole expired partitions. Expired rows can
+                    # still sit in the DEFAULT partition (backfill, coverage gaps)
+                    # or in a partition that spans the cutoff, so retention must
+                    # also delete those stragglers row-wise.
+                    total_deleted = await self._delete_old_logs(prisma_client, cutoff_date)
+                    verbose_proxy_logger.info(
+                        "Deleted %s expired logs not covered by dropped partitions", total_deleted
+                    )
+                else:
+                    total_deleted = await self._delete_old_logs(prisma_client, cutoff_date)
+                    verbose_proxy_logger.info("Deleted %s logs", total_deleted)
 
-            index_deleted: Final = await self._delete_old_tool_index_rows(prisma_client, cutoff_date)
-            verbose_proxy_logger.info("Deleted %s expired tool index rows", index_deleted)
+                index_deleted: Final = await self._delete_old_tool_index_rows(prisma_client, cutoff_date)
+                verbose_proxy_logger.info("Deleted %s expired tool index rows", index_deleted)
+
+            sessions_deleted: Final = await self._delete_old_auto_router_sessions(
+                prisma_client, self._auto_router_session_cutoff(retention_seconds)
+            )
+            verbose_proxy_logger.info("Deleted %s expired auto-router session rollups", sessions_deleted)
 
         except Exception as e:
             # .exception() captures the traceback; str(e) alone on a Prisma/DB