feat(proxy): add config source precedence toggle

litellm/proxy/_types.py

@@ -94,6 +94,11 @@ class SupportedDBObjectType(str, enum.Enum):
         return str(self.value)
 
 
+class ConfigSourceOfTruth(str, enum.Enum):
+    DATABASE = "database"
+    CONFIG_FILE = "config_file"
+
+
 class LiteLLMTeamRoles(enum.Enum):
     # team admin
     TEAM_ADMIN = "admin"
@@ -2470,6 +2475,14 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
         None,
         description="If True, models and config are stored in and loaded from the database. Default is False.",
     )
+    config_source_of_truth: ConfigSourceOfTruth = Field(
+        ConfigSourceOfTruth.DATABASE,
+        description=(
+            "Controls precedence when store_model_in_db is enabled. 'database' preserves the default behavior where "
+            "LiteLLM_Config values override the deployed config. 'config_file' makes explicit deployed config values "
+            "override database values while still loading database-only values. Set this in the deployed config file."
+        ),
+    )
     forward_client_headers_to_llm_api: bool | None = Field(
         None,
         description="If True, forwards client headers (e.g. Authorization) to the LLM API. Required for Claude Code with Max subscription.",

litellm/proxy/management_endpoints/coordination_redis_endpoints.py

@@ -187,23 +187,25 @@ async def get_persisted_coordination_redis_settings() -> dict[str, object] | Non
 async def _current_coordination_redis_settings() -> dict[str, object] | None:
     """The coordination_redis block the proxy would boot with.
 
-    The persisted row wins over the yaml-loaded config state because startup
-    applies the DB `general_settings` row over the file config.
+    The configured source-of-truth mode determines whether an explicit file
+    value or the persisted row wins.
     """
     from litellm.proxy.proxy_server import proxy_config
 
+    config_state: Final = _SETTINGS_ADAPTER.validate_python(proxy_config.get_config_state())
+    general_settings_value: Final = config_state.get(_GENERAL_SETTINGS_PARAM_NAME)
+    from_file_value: Final = (
+        general_settings_value.get(_COORDINATION_REDIS_KEY) if isinstance(general_settings_value, Mapping) else None
+    )
+    from_file: Final = _SETTINGS_ADAPTER.validate_python(from_file_value) if isinstance(from_file_value, dict) else None
+    if proxy_config.config_file_general_setting_is_authoritative(_COORDINATION_REDIS_KEY):
+        return from_file
+
     persisted: Final = await get_persisted_coordination_redis_settings()
     if persisted is not None:
         return persisted
 
-    config_state: Final = _SETTINGS_ADAPTER.validate_python(proxy_config.get_config_state())
-    general_settings: Final = config_state.get(_GENERAL_SETTINGS_PARAM_NAME)
-    if not isinstance(general_settings, dict):
-        return None
-    from_file: Final = general_settings.get(_COORDINATION_REDIS_KEY)
-    if isinstance(from_file, dict):
-        return _SETTINGS_ADAPTER.validate_python(from_file)
-    return None
+    return from_file
 
 
 def _coordination_redis_source(settings: Mapping[str, object] | None) -> CoordinationRedisSource | None:

litellm/proxy/proxy_server.py

@@ -76,6 +76,7 @@ from litellm.proxy._types import (
     ConfigFieldUpdate,
     ConfigGeneralSettings,
     ConfigList,
+    ConfigSourceOfTruth,
     ConfigYAML,
     CoordinationRedisParams,
     EnterpriseLicenseData,
@@ -4028,6 +4029,7 @@ class ProxyConfig:
         self.worker_registry: list[WorkerRegistryEntry] = []
         self.config_sync_subscriber: ConfigSyncSubscriber | None = None
         self.auth_cache_invalidation_subscriber: AuthCacheInvalidationSubscriber | None = None
+        self._config_source_of_truth: ConfigSourceOfTruth = ConfigSourceOfTruth.DATABASE
         from litellm.litellm_core_utils.get_model_cost_map import (
             get_model_cost_map_loaded_at,
         )
@@ -4037,10 +4039,45 @@ class ProxyConfig:
         # whether an existing request predates the prices it just fetched, and re-serving one
         # costs a single fetch where skipping one leaves it priced wrong indefinitely
         self.model_cost_map_applied_revision: int = 0
-        # Keys explicitly set in the YAML config file. Used to give YAML
-        # precedence over stale DB-cached values for these specific keys
-        # during periodic config reloads (_update_general_settings).
-        self._yaml_general_settings_keys: set[str] = set()  # mutable-ok: populated once at startup, read-only thereafter  # fmt: skip
+        self._yaml_general_settings_keys: frozenset[str] = frozenset()
+
+    def _record_deployed_config(self, config: Mapping[str, object]) -> None:
+        deployed_general_settings: Final = config.get("general_settings")
+        if not isinstance(deployed_general_settings, Mapping):
+            self._yaml_general_settings_keys = frozenset()
+            self._config_source_of_truth = ConfigSourceOfTruth.DATABASE
+            return
+
+        self._yaml_general_settings_keys = frozenset(deployed_general_settings)
+        configured_source: Final = deployed_general_settings.get("config_source_of_truth")
+        if configured_source is None:
+            self._config_source_of_truth = ConfigSourceOfTruth.DATABASE
+        elif isinstance(configured_source, ConfigSourceOfTruth):
+            self._config_source_of_truth = configured_source
+        elif isinstance(configured_source, str):
+            self._config_source_of_truth = ConfigSourceOfTruth(configured_source)
+        else:
+            raise ValueError("general_settings.config_source_of_truth must be 'database' or 'config_file'")
+
+    def config_file_general_setting_is_authoritative(self, key: str) -> bool:
+        return (
+            self._config_source_of_truth is ConfigSourceOfTruth.CONFIG_FILE and key in self._yaml_general_settings_keys
+        )
+
+    @classmethod
+    def _merge_deployed_over_database(cls, database_value: object, deployed_value: object) -> object:
+        if not isinstance(database_value, Mapping) or not isinstance(deployed_value, Mapping):
+            return copy.deepcopy(deployed_value)
+        database_snapshot: Final = MappingProxyType(
+            {key: copy.deepcopy(value) for key, value in database_value.items()}
+        )
+        deployed_snapshot: Final = MappingProxyType(
+            {
+                key: cls._merge_deployed_over_database(database_value.get(key), value)
+                for key, value in deployed_value.items()
+            }
+        )
+        return database_snapshot | deployed_snapshot
 
     def is_yaml(self, config_file_path: str) -> bool:
         if not os.path.isfile(config_file_path):
@@ -4402,6 +4439,8 @@ class ProxyConfig:
 
             config = await self._get_config_from_file(config_file_path=config_file_path)
 
+        self._record_deployed_config(config)
+
         ## UPDATE CONFIG WITH DB
         if prisma_client is not None and store_model_in_db is True:
             config = await self._update_config_from_db(
@@ -4973,11 +5012,6 @@ class ProxyConfig:
         _hc_staleness = None
         _hc_ignore_transient = False
         if general_settings:
-            # Record which keys were explicitly set in the YAML config file.
-            # These keys take precedence over DB-cached values during periodic
-            # reloads (see _update_general_settings).
-            self._yaml_general_settings_keys = set(general_settings.keys())  # mutable-ok: snapshot of YAML keys at load time  # fmt: skip
-
             ### LOAD KEY MANAGEMENT SETTINGS FIRST (needed for custom secret manager) ###
             key_management_settings: Final = general_settings.get("key_management_settings", None)
             if key_management_settings is not None:
@@ -5853,7 +5887,7 @@ class ProxyConfig:
         return encrypted_env_vars
 
     def _decrypt_and_set_db_env_variables(
-        self, environment_variables: dict, return_original_value: bool = False
+        self, environment_variables: Mapping[str, object], return_original_value: bool = False
     ) -> dict:
         """
         Decrypts a dictionary of environment variables and then sets them in the environment
@@ -6006,9 +6040,16 @@ class ProxyConfig:
                 and db_router_settings is not None
                 and isinstance(db_router_settings.param_value, dict)
             ):
-                from litellm.utils import _update_dictionary
+                if self._config_source_of_truth is ConfigSourceOfTruth.CONFIG_FILE:
+                    combined_router_settings = self._merge_deployed_over_database(
+                        db_router_settings.param_value, config_router_settings
+                    )
+                else:
+                    from litellm.utils import _update_dictionary
 
-                combined_router_settings = _update_dictionary(config_router_settings, db_router_settings.param_value)
+                    combined_router_settings = _update_dictionary(
+                        config_router_settings, db_router_settings.param_value
+                    )
             elif config_router_settings is not None and isinstance(config_router_settings, dict):
                 combined_router_settings = config_router_settings
             elif db_router_settings is not None and isinstance(db_router_settings.param_value, dict):
@@ -6155,14 +6196,20 @@ class ProxyConfig:
                 except ValueError:
                     verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value")
 
-    async def _update_general_settings(self, db_general_settings: Json | None):
+    async def _update_general_settings(self, db_general_settings: Mapping[str, JsonValue] | None):
         """
         Pull from DB, read general settings value
         """
         global general_settings, store_model_in_db
         if db_general_settings is None:
             return
-        _general_settings: Final = dict(db_general_settings)
+        _general_settings: Final[Mapping[str, JsonValue]] = MappingProxyType(
+            {
+                key: value
+                for key, value in db_general_settings.items()
+                if not self.config_file_general_setting_is_authoritative(key)
+            }
+        )
         ## MAX PARALLEL REQUESTS ##
         if "max_parallel_requests" in _general_settings:
             general_settings["max_parallel_requests"] = _general_settings["max_parallel_requests"]
@@ -6311,38 +6358,79 @@ class ProxyConfig:
         # DB-sourced ``s3://`` value would otherwise reach
         # ``_load_instance_from_remote_storage`` without going through
         # the runtime gate.
-        db_param_value = _scrub_db_overlay_remote_module_loads(section=param_name, db_value=db_param_value)
+        scrubbed_db_param_value: Final = _scrub_db_overlay_remote_module_loads(
+            section=param_name, db_value=db_param_value
+        )
 
         if param_name == "environment_variables":
-            decrypted_env_vars = self._decrypt_and_set_db_env_variables(db_param_value, return_original_value=True)
+            deployed_env_vars_value: Final = current_config.get("environment_variables")
+            deployed_env_vars: Final[Mapping[object, object]] = (
+                deployed_env_vars_value if isinstance(deployed_env_vars_value, Mapping) else MappingProxyType({})
+            )
+            deployed_env_keys: Final = frozenset(str(key).upper() for key in deployed_env_vars)
+            db_env_vars: Final = (
+                MappingProxyType(
+                    {
+                        key: value
+                        for key, value in scrubbed_db_param_value.items()
+                        if str(key).upper() not in deployed_env_keys
+                    }
+                )
+                if self._config_source_of_truth is ConfigSourceOfTruth.CONFIG_FILE
+                and isinstance(scrubbed_db_param_value, Mapping)
+                else scrubbed_db_param_value
+            )
+            decrypted_env_vars = self._decrypt_and_set_db_env_variables(db_env_vars, return_original_value=True)
             # Normalize keys when loading from DB so services expecting uppercase
             # (e.g. Datadog) can read them even if stored in lowercase.
-            merged_env_vars: Final[dict] = {}
             for key, value in decrypted_env_vars.items():
-                merged_env_vars[key] = value
-                upper_key = key.upper()
-                merged_env_vars[upper_key] = value
-                os.environ[upper_key] = value
+                os.environ[key.upper()] = value
+            merged_env_vars: Final[Mapping[str, str]] = MappingProxyType(
+                {
+                    normalized_key: value
+                    for key, value in decrypted_env_vars.items()
+                    for normalized_key in (key, key.upper())
+                }
+            )
 
-            current_config.setdefault("environment_variables", {}).update(merged_env_vars)
+            if self._config_source_of_truth is ConfigSourceOfTruth.CONFIG_FILE:
+                current_config["environment_variables"] = self._merge_deployed_over_database(
+                    merged_env_vars, deployed_env_vars
+                )
+            else:
+                current_config.setdefault("environment_variables", {}).update(merged_env_vars)
             return current_config
-        elif param_name == "litellm_settings" and isinstance(db_param_value, dict):
-            for key, value in db_param_value.items():
+        elif (
+            param_name == "litellm_settings"
+            and self._config_source_of_truth is ConfigSourceOfTruth.DATABASE
+            and isinstance(scrubbed_db_param_value, dict)
+        ):
+            for key, value in scrubbed_db_param_value.items():
                 if key in LITELLM_SETTINGS_SAFE_DB_OVERRIDES:  # params that are safe to override with db values
                     setattr(litellm, key, value)
 
+        if self._config_source_of_truth is ConfigSourceOfTruth.CONFIG_FILE and param_name in current_config:
+            current_config[param_name] = self._merge_deployed_over_database(
+                scrubbed_db_param_value, current_config[param_name]
+            )
+            if param_name == "litellm_settings" and isinstance(current_config[param_name], dict):
+                for key, value in current_config[param_name].items():
+                    if key in LITELLM_SETTINGS_SAFE_DB_OVERRIDES:
+                        setattr(litellm, key, value)
+            return current_config
+
         # If param doesn't exist in config, add it
         if param_name not in current_config:
-            current_config[param_name] = db_param_value
+            current_config[param_name] = scrubbed_db_param_value
 
             return current_config
 
         # For dictionary values, update only non-none values
-        if isinstance(current_config[param_name], dict) and isinstance(db_param_value, dict):
-            _deep_merge_dicts(current_config[param_name], db_param_value)
+        if isinstance(current_config[param_name], dict) and isinstance(scrubbed_db_param_value, dict):
+            _deep_merge_dicts(current_config[param_name], scrubbed_db_param_value)
         else:
             # Non-dict or mismatched types: DB value replaces config (unchanged behavior)
-            current_config[param_name] = db_param_value
+            current_config[param_name] = scrubbed_db_param_value
 
         return current_config
 
@@ -8324,6 +8412,8 @@ class ProxyStartupEvent:
         Returns None when nothing is persisted or the persisted block names no
         connection target, leaving the file/env resolution untouched.
         """
+        if proxy_config.config_file_general_setting_is_authoritative("coordination_redis"):
+            return None
         try:
             persisted: Final = await get_persisted_coordination_redis_settings()
         except Exception as e:  # noqa: BLE001  # a config-row read failure must not block proxy startup
@@ -8696,7 +8786,11 @@ class ProxyStartupEvent:
         # If store_model_in_db is still False, check DB for override.
         # This breaks the chicken-and-egg where DB has store_model_in_db=True
         # but YAML config has False.
-        if store_model_in_db is not True and prisma_client is not None:
+        if (
+            store_model_in_db is not True
+            and prisma_client is not None
+            and not proxy_config.config_file_general_setting_is_authoritative("store_model_in_db")
+        ):
             try:
                 _db_gs_record: Final[_ConfigParamRow | None] = await ConfigRepository(prisma_client).table.find_first(
                     where={"param_name": "general_settings"}

tests/test_litellm/proxy/management_endpoints/test_coordination_redis_endpoints.py

@@ -59,11 +59,15 @@ def _prisma_with_general_settings(general_settings: dict | None) -> MagicMock:
     return mock_prisma
 
 
-def _proxy_config(file_general_settings: dict | None = None) -> MagicMock:
+def _proxy_config(
+    file_general_settings: dict | None = None,
+    config_file_authoritative: bool = False,
+) -> MagicMock:
     proxy_config = MagicMock()
     proxy_config.get_config_state = MagicMock(
         return_value={"general_settings": file_general_settings or {}},
     )
+    proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=config_file_authoritative)
     return proxy_config
 
 
@@ -136,6 +140,29 @@ async def test_get_source_reads_block_from_yaml_config_when_db_row_absent(monkey
     assert response.values["host"] == "yaml-redis"
 
 
+@pytest.mark.asyncio
+async def test_get_source_prefers_config_file_block_when_authoritative(monkeypatch):
+    monkeypatch.setattr(litellm, "cache", None)
+    monkeypatch.delenv("REDIS_HOST", raising=False)
+
+    with (
+        patch(
+            "litellm.proxy.proxy_server.prisma_client",
+            _prisma_with_general_settings({"coordination_redis": {"host": "database-redis"}}),
+        ),
+        patch(
+            "litellm.proxy.proxy_server.proxy_config",
+            _proxy_config(
+                {"coordination_redis": {"host": "deployed-redis"}},
+                config_file_authoritative=True,
+            ),
+        ),
+    ):
+        response = await get_coordination_redis_settings(user_api_key_dict=_admin_auth())
+
+    assert response.values["host"] == "deployed-redis"
+
+
 @pytest.mark.parametrize("cache_backend_cls", [RedisCache, RedisClusterCache])
 @pytest.mark.asyncio
 async def test_get_source_is_cache_backend_when_no_block(monkeypatch, cache_backend_cls):

tests/test_litellm/proxy/proxy_server/test_proxy_config.py

@@ -17,6 +17,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 import litellm
+from litellm.proxy._types import ConfigGeneralSettings, ConfigSourceOfTruth
 from litellm.proxy.proxy_server import (
     ProxyConfig,
     _is_remote_module_url,
@@ -2167,6 +2168,38 @@ async def test_ProxyConfig__add_router_settings_from_db_config_updates_router():
     }
 
 
+@pytest.mark.asyncio
+async def test_ProxyConfig__add_router_settings_config_file_wins_conflicts():
+    pc = ProxyConfig()
+    pc._record_deployed_config({"general_settings": {"config_source_of_truth": "config_file"}})
+    fake_router = MagicMock()
+    fake_prisma = MagicMock()
+    fake_prisma.db.litellm_config.find_first = AsyncMock(
+        return_value=SimpleNamespace(
+            param_value={
+                "timeout": 30,
+                "retry_policy": {"database": True, "conflict": "database"},
+            }
+        )
+    )
+
+    await pc._add_router_settings_from_db_config(
+        config_data={
+            "router_settings": {
+                "timeout": 10,
+                "retry_policy": {"deployed": True, "conflict": "deployed"},
+            }
+        },
+        llm_router=fake_router,
+        prisma_client=fake_prisma,
+    )
+
+    fake_router.update_settings.assert_called_once_with(
+        timeout=10,
+        retry_policy={"database": True, "deployed": True, "conflict": "deployed"},
+    )
+
+
 @pytest.mark.asyncio
 async def test_ProxyConfig__add_router_settings_from_db_config_none_router_noop():
     pc = ProxyConfig()
@@ -2349,6 +2382,129 @@ def test_ProxyConfig__update_config_fields_merges_dict():
     assert out == {"general_settings": {"a": 1, "b": 3, "c": 4, "d": 5}}
 
 
+@pytest.mark.parametrize(
+    "source",
+    [ConfigSourceOfTruth.DATABASE, ConfigSourceOfTruth.CONFIG_FILE],
+)
+def test_ConfigGeneralSettings_validates_config_source_of_truth(source):
+    settings = ConfigGeneralSettings(config_source_of_truth=source)
+
+    assert settings.config_source_of_truth is source
+
+
+def test_ConfigGeneralSettings_config_source_of_truth_defaults_to_database():
+    assert ConfigGeneralSettings().config_source_of_truth is ConfigSourceOfTruth.DATABASE
+
+
+def test_ConfigGeneralSettings_rejects_invalid_config_source_of_truth():
+    with pytest.raises(ValueError):
+        ConfigGeneralSettings.model_validate({"config_source_of_truth": "invalid"})
+
+
+def test_ProxyConfig__update_config_fields_config_file_wins_conflicts():
+    pc = ProxyConfig()
+    pc._record_deployed_config({"general_settings": {"config_source_of_truth": "config_file"}})
+    current = {
+        "general_settings": {
+            "max_parallel_requests": 10,
+            "allowed_models": [],
+            "nested": {"deployed": True, "conflict": None},
+        }
+    }
+
+    out = pc._update_config_fields(
+        current_config=current,
+        param_name="general_settings",
+        db_param_value={
+            "max_parallel_requests": 20,
+            "allowed_models": ["db-model"],
+            "db_only": True,
+            "nested": {"database": True, "conflict": "database"},
+        },
+    )
+
+    assert out["general_settings"] == {
+        "max_parallel_requests": 10,
+        "allowed_models": [],
+        "db_only": True,
+        "nested": {"database": True, "deployed": True, "conflict": None},
+    }
+
+
+def test_ProxyConfig__update_config_fields_config_file_env_wins_case_insensitively(monkeypatch):
+    pc = ProxyConfig()
+    pc._record_deployed_config({"general_settings": {"config_source_of_truth": "config_file"}})
+    monkeypatch.delenv("API_KEY", raising=False)
+    monkeypatch.delenv("DB_ONLY", raising=False)
+
+    out = pc._update_config_fields(
+        current_config={"environment_variables": {"api_key": "deployed"}},
+        param_name="environment_variables",
+        db_param_value={"API_KEY": "database", "DB_ONLY": "database-only"},
+    )
+
+    assert out["environment_variables"]["api_key"] == "deployed"
+    assert out["environment_variables"]["DB_ONLY"] == "database-only"
+    assert "API_KEY" not in os.environ
+    assert os.environ["DB_ONLY"] == "database-only"
+
+
+@pytest.mark.asyncio
+async def test_ProxyConfig_get_config_reads_source_before_database_merge(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    pc = ProxyConfig()
+    deployed = {
+        "general_settings": {
+            "config_source_of_truth": "config_file",
+            "max_parallel_requests": 10,
+        }
+    }
+    monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
+    monkeypatch.setattr(proxy_server, "store_model_in_db", True)
+    monkeypatch.setattr(pc, "_get_config_from_file", AsyncMock(return_value=deployed))
+
+    async def merge_after_source_is_recorded(**kwargs):
+        assert pc._config_source_of_truth is ConfigSourceOfTruth.CONFIG_FILE
+        assert pc._yaml_general_settings_keys == {
+            "config_source_of_truth",
+            "max_parallel_requests",
+        }
+        return kwargs["config"]
+
+    monkeypatch.setattr(pc, "_update_config_from_db", merge_after_source_is_recorded)
+
+    await pc.get_config(config_file_path="config.yaml")
+
+
+@pytest.mark.asyncio
+async def test_ProxyConfig__update_general_settings_respects_config_file_source(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    pc = ProxyConfig()
+    pc._record_deployed_config(
+        {
+            "general_settings": {
+                "config_source_of_truth": "config_file",
+                "max_parallel_requests": 10,
+            }
+        }
+    )
+    monkeypatch.setattr(proxy_server, "general_settings", {"max_parallel_requests": 10})
+
+    await pc._update_general_settings(
+        db_general_settings={
+            "max_parallel_requests": 20,
+            "global_max_parallel_requests": 30,
+        }
+    )
+
+    assert proxy_server.general_settings == {
+        "max_parallel_requests": 10,
+        "global_max_parallel_requests": 30,
+    }
+
+
 def test_ProxyConfig__update_config_fields_invalid_param_raises():
     pc = ProxyConfig()
     with pytest.raises(Exception):

tests/test_litellm/proxy/test_proxy_server.py

@@ -702,6 +702,7 @@ async def test_initialize_scheduled_jobs_credentials(monkeypatch):
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
 
     with (
         patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
@@ -762,6 +763,7 @@ async def test_periodic_reload_job_scheduled_without_store_model_in_db(monkeypat
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
     scheduler = AsyncIOScheduler()
 
     try:
@@ -803,6 +805,7 @@ async def test_initialize_scheduled_jobs_uses_configured_config_reload_interval(
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
     mock_scheduler = MagicMock()
 
     configured_interval = 47
@@ -852,6 +855,7 @@ async def test_initialize_scheduled_jobs_rejects_non_positive_config_reload_inte
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
     mock_scheduler = MagicMock()
 
     with (
@@ -899,6 +903,7 @@ async def test_initialize_scheduled_jobs_hydrates_mcp_when_store_model_in_db_fal
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
 
     with (
         patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
@@ -6862,6 +6867,7 @@ async def test_store_model_in_db_db_override_when_config_false():
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
 
     with (
         patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
@@ -6888,6 +6894,42 @@ async def test_store_model_in_db_db_override_when_config_false():
         assert mock_proxy_config.get_credentials.call_count == 1
 
 
+@pytest.mark.asyncio
+async def test_store_model_in_db_config_file_false_blocks_database_override():
+    from litellm.proxy.proxy_server import ProxyStartupEvent
+    from litellm.proxy.utils import ProxyLogging
+
+    mock_prisma_client = MagicMock()
+    mock_db_record = MagicMock()
+    mock_db_record.param_value = {"store_model_in_db": True}
+    mock_prisma_client.db.litellm_config.find_first = AsyncMock(return_value=mock_db_record)
+    mock_proxy_logging = MagicMock(spec=ProxyLogging)
+    mock_proxy_logging.slack_alerting_instance = MagicMock()
+    mock_proxy_logging.db_spend_update_writer = MagicMock()
+    mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=True)
+
+    with (
+        patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
+        patch("litellm.proxy.proxy_server.store_model_in_db", False),
+        patch("litellm.proxy.proxy_server.get_secret_bool", return_value=False),
+    ):
+        await ProxyStartupEvent.initialize_scheduled_background_jobs(
+            general_settings={},
+            prisma_client=mock_prisma_client,
+            proxy_budget_rescheduler_min_time=1,
+            proxy_budget_rescheduler_max_time=2,
+            proxy_batch_write_at=5,
+            proxy_logging_obj=mock_proxy_logging,
+        )
+
+        import litellm.proxy.proxy_server as ps
+
+        assert ps.store_model_in_db is False
+        mock_prisma_client.db.litellm_config.find_first.assert_not_awaited()
+        mock_proxy_config.add_deployment.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_store_model_in_db_db_check_skipped_when_already_true(monkeypatch):
     """
@@ -6905,6 +6947,7 @@ async def test_store_model_in_db_db_check_skipped_when_already_true(monkeypatch)
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
 
     with (
         patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
@@ -6948,6 +6991,7 @@ async def test_store_model_in_db_db_failure_graceful(monkeypatch):
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
 
     with (
         patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
@@ -10383,6 +10427,11 @@ async def test_startup_applies_coordination_redis_saved_in_database():
         patch.object(proxy_server_module, "litellm_config_cache", fresh_config_cache),
         patch.object(proxy_server_module, "RedisCache", _EnvBuiltRedisCache),
         patch.object(proxy_server_module, "RedisClusterCache", _EnvBuiltClusterCache),
+        patch.object(
+            proxy_server_module.proxy_config,
+            "config_file_general_setting_is_authoritative",
+            return_value=False,
+        ),
         patch.object(
             proxy_server_module,
             "get_persisted_coordination_redis_settings",
@@ -10400,6 +10449,30 @@ async def test_startup_applies_coordination_redis_saved_in_database():
     assert fresh_config_cache.redis_cache is result
 
 
+@pytest.mark.asyncio
+async def test_startup_keeps_config_file_coordination_redis_when_authoritative():
+    persisted_settings = AsyncMock(return_value={"host": "database-redis"})
+    with (
+        patch.object(
+            proxy_server_module.proxy_config,
+            "config_file_general_setting_is_authoritative",
+            return_value=True,
+        ),
+        patch.object(
+            proxy_server_module,
+            "get_persisted_coordination_redis_settings",
+            persisted_settings,
+        ),
+    ):
+        result = await proxy_server_module.ProxyStartupEvent._init_coordination_redis_from_db(
+            litellm_settings={},
+            llm_router=None,
+        )
+
+    assert result is None
+    persisted_settings.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_startup_ignores_database_coordination_redis_without_connection_target():
     """A persisted block with no host/url/cluster/sentinel must be ignored rather
@@ -10408,6 +10481,11 @@ async def test_startup_ignores_database_coordination_redis_without_connection_ta
         patch.object(proxy_server_module, "spend_counter_cache", DualCache()),
         patch.object(proxy_server_module, "litellm_config_cache", types.SimpleNamespace(redis_cache=None)),
         patch.object(proxy_server_module, "RedisCache", _EnvBuiltRedisCache),
+        patch.object(
+            proxy_server_module.proxy_config,
+            "config_file_general_setting_is_authoritative",
+            return_value=False,
+        ),
         patch.object(
             proxy_server_module,
             "get_persisted_coordination_redis_settings",
@@ -10426,6 +10504,11 @@ async def test_startup_ignores_database_coordination_redis_without_connection_ta
 async def test_startup_survives_database_read_failure_for_coordination_redis():
     """A config-row read failure must not block proxy startup."""
     with (
+        patch.object(
+            proxy_server_module.proxy_config,
+            "config_file_general_setting_is_authoritative",
+            return_value=False,
+        ),
         patch.object(
             proxy_server_module,
             "get_persisted_coordination_redis_settings",
@@ -10745,6 +10828,7 @@ async def _run_scheduled_background_jobs():
     mock_proxy_logging.slack_alerting_instance = MagicMock()
     mock_proxy_logging.db_spend_update_writer = MagicMock()
     mock_proxy_config = AsyncMock()
+    mock_proxy_config.config_file_general_setting_is_authoritative = MagicMock(return_value=False)
 
     with (
         patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),