fix(proxy): treat a reset to the bundled catalog value as an echo even after router registration

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
ryan 2026-09-20 09:12:26 +00:00
parent ccab0d4c01
commit 721e3b86f8
4 changed files with 56 additions and 6 deletions

View file

@ -10,6 +10,7 @@ export LITELLM_LOCAL_MODEL_COST_MAP=True
"""
import asyncio
import functools
import hashlib
import json
import os
@ -17,14 +18,16 @@ import random
import sys
import threading
import time
from collections.abc import Awaitable, Callable
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from importlib.resources import files
from pathlib import Path
from types import MappingProxyType
from typing import Final, Protocol
import httpx
from pydantic import TypeAdapter
from typing_extensions import ReadOnly, TypedDict
from litellm import verbose_logger
@ -37,6 +40,7 @@ from litellm.litellm_core_utils.fallback_generalizations import (
)
FALLBACK_GENERALIZATIONS_KEY: Final = "fallback_generalizations"
_BUNDLED_CATALOG_ADAPTER: Final = TypeAdapter(dict[str, dict[str, object]])
_CLI_ENTRYPOINT_NAMES: Final = frozenset({"lite", "litellm-proxy"})
@ -88,6 +92,13 @@ class GetModelCostMap:
"""Load the local backup model cost map bundled with the package."""
return GetModelCostMap.load_local_model_cost_map_with_revision().model_cost_map
@staticmethod
@functools.lru_cache(maxsize=1)
def bundled_model_cost_map() -> Mapping[str, Mapping[str, object]]:
"""The bundled catalog as shipped, untouched by ``register_model`` or router registrations."""
raw: Final = _BUNDLED_CATALOG_ADAPTER.validate_python(GetModelCostMap.load_local_model_cost_map())
return MappingProxyType({key: MappingProxyType(entry) for key, entry in raw.items()})
@classmethod
def _get_backup_model_count(cls) -> int:
"""Return the number of models in the local backup (cached int)."""

View file

@ -28,6 +28,7 @@ import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.constants import LITELLM_PROXY_ADMIN_NAME
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.ptu_pricing import (
CUSTOM_PRICING_FIELDS,
PTU_EMPTIED_PRICING_FIELDS,
@ -138,6 +139,7 @@ from litellm.types.router import (
updateLiteLLMParams,
)
from litellm.types.utils import (
COST_MAP_LOOKUP_KEY,
echoed_cost_map_fields,
echoed_cost_map_pricing_fields,
without_server_derived_pricing,
@ -886,6 +888,13 @@ def _cost_map_entry(db_model: Deployment, incoming_model_info: Mapping[str, obje
return MappingProxyType({})
def _bundled_cost_map_entry(incoming_model_info: Mapping[str, object]) -> Mapping[str, object]:
catalog_key: Final = incoming_model_info.get(COST_MAP_LOOKUP_KEY)
if not isinstance(catalog_key, str):
return MappingProxyType({})
return GetModelCostMap.bundled_model_cost_map().get(catalog_key, MappingProxyType({}))
def update_db_model(db_model: Deployment, updated_patch: updateDeployment) -> PrismaCompatibleUpdateDBModel:
if updated_patch.model_info is not None:
_raise_if_ptu_cost_attribution_disabled(updated_patch.model_info.model_dump(exclude_none=True))
@ -910,7 +919,9 @@ def update_db_model(db_model: Deployment, updated_patch: updateDeployment) -> Pr
if updated_patch.model_info:
incoming_model_info: Final = updated_patch.model_info.model_dump(exclude_none=True)
echoed_fields: Final = echoed_cost_map_fields(
incoming_model_info, _cost_map_entry(db_model, incoming_model_info)
incoming_model_info,
_cost_map_entry(db_model, incoming_model_info),
_bundled_cost_map_entry(incoming_model_info),
)
merged_model_info.update(
MappingProxyType(

View file

@ -3779,16 +3779,20 @@ def echoed_cost_map_pricing_fields(model_info: Mapping[str, Any]) -> tuple[str,
return tuple(sorted(k for k in model_info if is_server_derived_pricing_key(k)))
def echoed_cost_map_fields(model_info: Mapping[str, Any], cost_map_entry: Mapping[str, Any]) -> tuple[str, ...]:
def echoed_cost_map_fields(model_info: Mapping[str, Any], *cost_map_entries: Mapping[str, Any]) -> tuple[str, ...]:
"""Fields a ``/model/info`` echo copied from the cost map unchanged.
Only ``litellm.get_model_info`` emits ``key``, so a blob carrying it is an echo of that
response. Anything in it that still equals the resolved cost-map entry is a display value
nobody typed; a value the operator edited differs and stays a real override.
response. Anything in it that still equals a resolved cost-map entry is a display value
nobody typed; a value the operator edited differs from every entry and stays a real override.
Callers pass both the live entry, which the router rewrites with each deployment's own
overrides, and the bundled entry, so a reset to the catalog value reads as an echo either way.
"""
if COST_MAP_LOOKUP_KEY not in model_info:
return ()
return tuple(sorted(k for k, v in model_info.items() if k in cost_map_entry and cost_map_entry[k] == v))
return tuple(
sorted(k for k, v in model_info.items() if any(k in entry and entry[k] == v for entry in cost_map_entries))
)
def pricing_override_fields(*sources: Mapping[str, Any]) -> tuple[str, ...]:

View file

@ -4090,6 +4090,30 @@ class TestModelInfoCostMapEchoFilter:
assert info["mode"] == "chat"
assert info["access_groups"] == ["staging"]
def test_reset_is_recognised_after_the_router_registered_the_override(self, monkeypatch: pytest.MonkeyPatch):
import litellm
from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
pristine = litellm.get_model_info("openai/gpt-5.6")
polluted = {**pristine, "max_input_tokens": 2048}
monkeypatch.setattr(litellm, "get_model_info", lambda model, **_: polluted)
db_model = Deployment(
model_name="gpt-5.6",
litellm_params=LiteLLM_Params(model="openai/gpt-5.6"),
model_info=ModelInfo(id="dep-echo-8", max_input_tokens=2048),
)
echo = {**pristine, "id": "dep-echo-8", "db_model": True}
result = update_db_model(
db_model=db_model,
updated_patch=updateDeployment(model_info=ModelInfo(**echo)),
)
info = json.loads(result["model_info"])
assert "max_input_tokens" not in info, info
def test_echo_is_compared_against_the_deployments_lookup_not_the_key(self):
import litellm