fix(autorouter): compare historical and new savings consistently

litellm/models/autorouter_session.py

@@ -34,15 +34,12 @@ class LiteLLM_AutoRouterSession(LiteLLMPydanticObjectBase):
 
     @property
     def baseline_model(self) -> str | None:
-        """The baseline most covered turns were priced against, or None when none were estimated.
-
-        A router reconfigured mid-session leaves turns priced against two baselines; the row keeps both
-        counts, and the label is the one that priced the most money-carrying turns rather than whatever the
-        router is configured with now.
-        """
-        if not self.savings_estimated_baseline_models:
+        """A recorded baseline label when excluded turns cannot change the selected model."""
+        if not self.baseline_models:
+            return None
+        if self.savings_estimated_turns < self.turns and len(self.baseline_models) > 1:
             return None
         return max(
-            self.savings_estimated_baseline_models,
-            key=lambda model: (self.savings_estimated_baseline_models[model], model),
+            self.baseline_models,
+            key=lambda model: (self.baseline_models[model], model),
         )

litellm/proxy/db/autorouter_savings_comparison.py

@@ -0,0 +1,147 @@
+from collections.abc import Mapping
+from contextlib import AbstractAsyncContextManager
+from datetime import timedelta
+from math import isclose
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, Protocol, cast
+
+from pydantic import BaseModel, ConfigDict, TypeAdapter
+
+from litellm._logging import verbose_proxy_logger
+from litellm.constants import MAX_SPENDLOG_ROWS_TO_QUERY
+from litellm.proxy.db.autorouter_session_rollup import AUTOROUTER_SESSION_WINDOW_SQL
+from litellm.proxy.db.create_views import SupportsRawQueries
+
+if TYPE_CHECKING:
+    from litellm.proxy.utils import PrismaClient
+
+
+class SessionSavingsComparison(BaseModel):
+    model_config = ConfigDict(frozen=True, allow_inf_nan=False)
+
+    router_name: str
+    router_type: str
+    turns: int
+    estimated_turns: int
+    actual_spend: float
+    classifier_cost: float | None
+    saved_spend: float
+    complete: bool
+
+    def coverage_fields(self, recorded_savings: float, recorded_turns: int) -> Mapping[str, float | int]:
+        if self.turns != recorded_turns or not self.complete:
+            return MappingProxyType({})
+        if not isclose(self.saved_spend, recorded_savings, rel_tol=1e-9, abs_tol=1e-9):
+            return MappingProxyType({})
+        return MappingProxyType(
+            {
+                "savings_estimated_turns": self.estimated_turns,
+                "savings_estimated_actual_spend": self.actual_spend,
+                "savings_estimated_saved_spend": self.saved_spend,
+            }
+        )
+
+
+class _ReadTransactions(Protocol):
+    def tx(self, *, timeout: timedelta, max_wait: timedelta) -> AbstractAsyncContextManager[SupportsRawQueries]: ...
+
+
+_COMPARISONS: Final = TypeAdapter(tuple[SessionSavingsComparison, ...])
+
+
+async def historical_session_comparisons(
+    prisma_client: "PrismaClient",
+    start_date: str,
+    end_date: str,
+    api_key: str | None,
+    user_id: str | None,
+    session_id: str | None = None,
+) -> Mapping[tuple[str, str], SessionSavingsComparison]:
+    try:
+        reader: Final = cast(_ReadTransactions, prisma_client.read_db)  # cast-ok: untyped Prisma transaction delegate
+        async with reader.tx(timeout=timedelta(seconds=3), max_wait=timedelta(seconds=1)) as transaction:
+            await transaction.execute_raw("SET TRANSACTION READ ONLY")
+            await transaction.execute_raw("SET LOCAL statement_timeout = 2000")
+            rows: Final = await transaction.query_raw(
+                HISTORICAL_SESSION_COMPARISONS_SQL,
+                start_date,
+                end_date,
+                api_key,
+                user_id,
+                session_id,
+            )
+            comparisons: Final = _COMPARISONS.validate_python(rows or ())
+            return MappingProxyType({(row.router_name, row.router_type): row for row in comparisons})
+    except Exception:  # noqa: BLE001  # missing retained logs must not discard recorded dollar savings
+        verbose_proxy_logger.warning("Historical auto-router cost comparison unavailable; preserving recorded savings")
+        return MappingProxyType({})
+
+
+HISTORICAL_SESSION_COMPARISONS_SQL: Final = f"""
+WITH {AUTOROUTER_SESSION_WINDOW_SQL}, scoped AS MATERIALIZED (
+    SELECT * FROM windowed WHERE $5::text IS NULL OR session_id = $5::text
+), limited_logs AS MATERIALIZED (
+    SELECT session.api_key, session.session_id, session.router_name, session.router_type, session.comparison_user_id,
+        session.classifier_cost_recorded_turns = session.turns AS classifier_cost_tracked,
+        logs.spend, logs.prompt_tokens + logs.completion_tokens AS tokens,
+        logs.metadata::jsonb -> 'routing_decision' AS decision,
+        logs.metadata::jsonb -> 'autorouter_savings' AS savings,
+        logs.metadata::jsonb -> 'autorouter_savings_estimate' AS estimate
+    FROM scoped AS session JOIN "LiteLLM_SpendLogs" AS logs
+      ON logs.api_key = session.api_key
+      AND CASE WHEN char_length(logs.session_id) > 256
+          THEN 'sha256:' || encode(sha256(convert_to(logs.session_id, 'UTF8')), 'hex')
+          ELSE logs.session_id END = session.session_id
+      AND (session.comparison_user_id IS NULL OR logs."user" = session.comparison_user_id)
+      AND logs."startTime" BETWEEN session.first_turn_at AND session.last_turn_at
+      AND COALESCE(logs.metadata::jsonb #>> '{{routing_decision,router_model_name}}', logs.model_group)
+          = session.router_name
+    WHERE session.savings_estimated_turns < session.turns
+      AND logs.status = 'success' AND COALESCE(logs.metadata::jsonb ->> 'internal_call_origin', '') = ''
+    LIMIT {MAX_SPENDLOG_ROWS_TO_QUERY + 1}
+), facts AS (
+    SELECT *,
+        CASE WHEN jsonb_typeof(decision -> 'classifier_cost') = 'number'
+            THEN (decision ->> 'classifier_cost')::float8
+            WHEN classifier_cost_tracked THEN 0 END AS classifier,
+        CASE WHEN jsonb_typeof(savings) = 'number' AND (
+            estimate IS NULL OR estimate = 'null'::jsonb OR (
+                jsonb_typeof(estimate -> 'version') = 'number' AND estimate ->> 'version' IN ('1', '2', '3')
+                AND estimate ->> 'status' = 'estimated'
+            )
+        ) THEN savings::text::float8 END AS saved
+    FROM limited_logs
+), compared AS (
+    SELECT api_key, session_id, router_name, router_type, comparison_user_id,
+        COUNT(*) AS turns, SUM(spend + COALESCE(classifier, 0)) AS spend, SUM(tokens) AS total_tokens,
+        COUNT(saved) AS estimated_turns,
+        COALESCE(SUM(spend + COALESCE(classifier, 0)) FILTER (WHERE saved IS NOT NULL), 0)::float8 AS actual_spend,
+        CASE WHEN COUNT(saved) = COUNT(classifier) FILTER (WHERE saved IS NOT NULL)
+            THEN COALESCE(SUM(classifier) FILTER (WHERE saved IS NOT NULL), 0)::float8
+        END AS estimated_classifier_cost,
+        COALESCE(SUM(saved), 0)::float8 AS saved_spend
+    FROM facts GROUP BY 1, 2, 3, 4, 5
+), reconciled AS (
+    SELECT session.*, logs.estimated_turns, logs.actual_spend, logs.estimated_classifier_cost,
+        COALESCE((SELECT COUNT(*) FROM limited_logs) <= {MAX_SPENDLOG_ROWS_TO_QUERY}
+            AND logs.turns = session.turns AND logs.total_tokens = session.total_tokens
+            AND ABS(logs.spend - session.spend) <= GREATEST(1e-9, ABS(session.spend) * 1e-9)
+            AND ABS(logs.saved_spend - session.saved_spend) <= GREATEST(1e-9, ABS(session.saved_spend) * 1e-9), FALSE
+        ) AS recovered
+    FROM scoped AS session LEFT JOIN compared AS logs
+        ON logs.api_key = session.api_key AND logs.session_id = session.session_id
+        AND logs.router_name = session.router_name AND logs.router_type = session.router_type
+        AND logs.comparison_user_id IS NOT DISTINCT FROM session.comparison_user_id
+)
+SELECT router_name, router_type,
+    SUM(turns)::bigint AS turns,
+    SUM(CASE WHEN recovered THEN estimated_turns ELSE savings_estimated_turns END)::bigint AS estimated_turns,
+    SUM(CASE WHEN recovered THEN actual_spend ELSE savings_estimated_actual_spend END)::float8 AS actual_spend,
+    CASE WHEN BOOL_AND(CASE WHEN recovered THEN estimated_classifier_cost IS NOT NULL
+        ELSE savings_estimated_turns = turns AND classifier_cost_recorded_turns = turns END)
+        THEN SUM(CASE WHEN recovered THEN estimated_classifier_cost ELSE classifier_cost END)::float8
+    END AS classifier_cost,
+    SUM(saved_spend)::float8 AS saved_spend,
+    BOOL_AND(recovered OR savings_estimated_turns = turns) AS complete
+FROM reconciled GROUP BY router_name, router_type
+"""

litellm/proxy/db/autorouter_session_rollup.py

@@ -7,8 +7,8 @@
 key and user session rollups with one atomic statement per turn: each upsert classifies
 the turn (same model, first visit, return to a model the session already used, out of
 order) against the row's own columns, so nothing is read before the write and concurrent
-pods compose. The benchmarks endpoint aggregates these rows and never touches
-LiteLLM_SpendLogs.
+pods compose. The benchmarks endpoint aggregates these rows and can recover matching historical
+costs from retained spend logs when estimate coverage predates these columns.
 """
 
 from __future__ import annotations
@@ -45,20 +45,24 @@
     savings_estimated_baseline_models
 """
 
-AUTOROUTER_BENCHMARKS_SQL: Final = f"""
-WITH windowed AS (
-    SELECT {_SESSION_COLUMNS} FROM "LiteLLM_AutoRouterSession"
+AUTOROUTER_SESSION_WINDOW_SQL: Final = f"""
+windowed AS (
+    SELECT {_SESSION_COLUMNS}, NULL::text AS comparison_user_id FROM "LiteLLM_AutoRouterSession"
     WHERE $4::text IS NULL
       AND last_turn_at >= $1::timestamp
       AND first_turn_at < $2::timestamp
       AND ($3::text IS NULL OR api_key = $3::text)
     UNION ALL
-    SELECT {_SESSION_COLUMNS} FROM "LiteLLM_AutoRouterUserSession"
+    SELECT {_SESSION_COLUMNS}, user_id AS comparison_user_id FROM "LiteLLM_AutoRouterUserSession"
     WHERE (($4::text IS NOT NULL AND user_id = $4::text) OR ($4::text IS NULL AND api_key = ''))
       AND last_turn_at >= $1::timestamp
       AND first_turn_at < $2::timestamp
       AND ($3::text IS NULL OR api_key = $3::text)
-),
+)
+"""
+
+AUTOROUTER_BENCHMARKS_SQL: Final = f"""
+WITH {AUTOROUTER_SESSION_WINDOW_SQL},
 tier_maps AS (
     SELECT router_name, router_type, jsonb_object_agg(tier, tier_turns) AS tier_turns
     FROM (
@@ -95,6 +99,8 @@
     COALESCE(SUM(saved_spend), 0)::float8 AS saved_spend,
     COALESCE(SUM(savings_estimated_turns), 0)::int AS savings_estimated_turns,
     COALESCE(SUM(savings_estimated_actual_spend), 0)::float8 AS savings_estimated_actual_spend,
+    CASE WHEN BOOL_AND(savings_estimated_turns = turns AND classifier_cost_recorded_turns = turns)
+        THEN SUM(classifier_cost)::float8 END AS savings_estimated_classifier_cost,
     COALESCE(SUM(savings_estimated_saved_spend), 0)::float8 AS savings_estimated_saved_spend,
     COALESCE(SUM(classifier_cost), 0)::float8 AS classifier_cost,
     COALESCE(SUM(classifier_cost_recorded_turns), 0)::int AS classifier_cost_recorded_turns,