fix(router): keep batch and file operations inside their own model group

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/router.py

@@ -122,6 +122,7 @@ from litellm.router_utils.common_utils import (
     _is_proxy_admin_request,
     filter_team_based_models,
     filter_web_search_deployments,
+    model_group_pinned_fallbacks,
 )
 from litellm.router_utils.cooldown_cache import CooldownCache
 from litellm.router_utils.cooldown_handlers import (
@@ -4863,6 +4864,7 @@ class Router:
             kwargs["model"] = model
             kwargs["original_function"] = self._acreate_file
             kwargs["num_retries"] = kwargs.get("num_retries", self.num_retries)
+            kwargs["fallbacks"] = model_group_pinned_fallbacks()
             self._update_kwargs_before_fallbacks(model=model, kwargs=kwargs)
             response = await self.async_function_with_fallbacks(**kwargs)
 
@@ -5124,6 +5126,7 @@ class Router:
             kwargs["model"] = model
             kwargs["original_function"] = self._acreate_batch
             kwargs["num_retries"] = kwargs.get("num_retries", self.num_retries)
+            kwargs["fallbacks"] = model_group_pinned_fallbacks()
             metadata_variable_name = _get_router_metadata_variable_name(function_name="_acreate_batch")
             self._update_kwargs_before_fallbacks(
                 model=model,
@@ -5340,6 +5343,7 @@ class Router:
             kwargs["model"] = model
             kwargs["original_function"] = self._acancel_batch
             kwargs["num_retries"] = kwargs.get("num_retries", self.num_retries)
+            kwargs["fallbacks"] = model_group_pinned_fallbacks()
             metadata_variable_name = _get_router_metadata_variable_name(function_name="_acancel_batch")
             self._update_kwargs_before_fallbacks(
                 model=model,

litellm/router_utils/common_utils.py

@@ -22,6 +22,20 @@ def _is_proxy_admin_request(request_kwargs: Optional[Mapping[str, object]]) -> b
     return getattr(user_api_key_auth, "user_role", None) == "proxy_admin"
 
 
+def model_group_pinned_fallbacks() -> List[str]:
+    """
+    Fallback list for requests that must stay inside the model group they were sent to.
+
+    Batch and file resources are owned by the provider that created them, so retrying a
+    failed batch/file operation on a different model group would create batch state with
+    credentials that do not own the `input_file_id`, and it replaces the owning provider's
+    error with an unrelated one from the fallback provider.
+
+    Intra-group failover (`num_retries`, order-based and weighted failover) still applies.
+    """
+    return []
+
+
 def get_litellm_params_sensitive_credential_hash(litellm_params: dict) -> str:
     """
     Hash of the credential params, used for mapping the file id to the right model

tests/test_litellm/test_router.py

@@ -6114,6 +6114,103 @@ async def test_acreate_batch_request_bedrock_tags_override_deployment_tags():
         assert mock_sign.call_args.kwargs["data"]["tags"] == request_tags
 
 
+def _batch_and_file_router() -> litellm.Router:
+    return litellm.Router(
+        model_list=[
+            {
+                "model_name": "my-gpt",
+                "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-fake"},
+                "model_info": {"id": "my-gpt_unified-1"},
+            },
+            {
+                "model_name": "my-azure-gpt",
+                "litellm_params": {
+                    "model": "azure/gpt-4o-mini",
+                    "api_base": "https://fake.openai.azure.com",
+                    "api_key": "fake",
+                    "api_version": "2024-08-01-preview",
+                },
+                "model_info": {"id": "my-azure-gpt_unified-1"},
+            },
+        ],
+        fallbacks=[{"my-gpt": ["my-azure-gpt"]}],
+        num_retries=0,
+    )
+
+
+@pytest.mark.asyncio
+async def test_acreate_batch_does_not_fall_back_to_another_model_group():
+    """
+    A batch is owned by the provider that owns `input_file_id`, so a failed
+    `batches.create` must surface that provider's error instead of creating batch
+    state on a fallback model group.
+    """
+    from litellm.types.utils import LiteLLMBatch
+
+    attempted_models: list[str] = []
+
+    async def fake_acreate_batch(**kwargs):
+        model = kwargs["model"]
+        attempted_models.append(model)
+        if model.startswith("openai/"):
+            raise litellm.BadRequestError(
+                message="Invalid value for 'completion_window': '5m'. Supported values are: '24h'",
+                model=model,
+                llm_provider="openai",
+            )
+        return LiteLLMBatch(
+            id="batch_from_fallback_group",
+            completion_window="24h",
+            created_at=0,
+            endpoint="/v1/chat/completions",
+            input_file_id=kwargs.get("input_file_id"),
+            object="batch",
+            status="failed",
+        )
+
+    router = _batch_and_file_router()
+    with patch("litellm.acreate_batch", side_effect=fake_acreate_batch):
+        with pytest.raises(litellm.BadRequestError) as exc_info:
+            await router.acreate_batch(
+                model="my-gpt",
+                input_file_id="file-owned-by-openai",
+                endpoint="/v1/chat/completions",
+                completion_window="5m",
+            )
+
+    assert "24h" in str(exc_info.value)
+    assert attempted_models == ["openai/gpt-4o-mini"]
+
+
+@pytest.mark.asyncio
+async def test_acreate_file_does_not_fall_back_to_another_model_group():
+    """
+    The unified file id records the model group the caller asked for, so a failed
+    upload must not silently store the file on a fallback group's provider.
+    """
+    attempted_models: list[str] = []
+
+    async def fake_acreate_file(**kwargs):
+        model = kwargs["model"]
+        attempted_models.append(model)
+        raise litellm.BadRequestError(
+            message="Invalid file format",
+            model=model,
+            llm_provider="openai",
+        )
+
+    router = _batch_and_file_router()
+    with patch("litellm.acreate_file", side_effect=fake_acreate_file):
+        with pytest.raises(litellm.BadRequestError):
+            await router.acreate_file(
+                model="my-gpt",
+                file=("batch.jsonl", b'{"custom_id":"1"}'),
+                purpose="user_data",
+            )
+
+    assert attempted_models == ["openai/gpt-4o-mini"]
+
+
 class TestPreRoutingStrategyRegistryLifecycle:
     """
     Regression tests: a deployment leaving the model_list must release the