Merge 6c4fd246d3 into 252c71c0b2

litellm/proxy/batches_endpoints/endpoints.py

@@ -222,6 +222,7 @@ async def create_batch(
                 llm_router=llm_router,
                 model_id=model_from_file_id,
                 operation_context="batch creation (file created with model)",
+                team_id=user_api_key_dict.team_id,
             )
 
             original_file_id: Final = get_original_file_id(input_file_id)
@@ -314,6 +315,7 @@ async def create_batch(
                     llm_router=llm_router,
                     model_id=model_param,
                     operation_context="batch creation",
+                    team_id=user_api_key_dict.team_id,
                 )
 
                 prepare_data_with_credentials(
@@ -544,6 +546,7 @@ async def retrieve_batch(
                 llm_router=llm_router,
                 model_id=model_from_id,
                 operation_context="batch retrieval (batch created with model)",
+                team_id=user_api_key_dict.team_id,
             )
 
             original_batch_id: Final = get_original_file_id(batch_id)
@@ -768,6 +771,7 @@ async def list_batches(
                 llm_router=llm_router,
                 model_id=model_param,
                 operation_context="batch listing",
+                team_id=user_api_key_dict.team_id,
             )
 
             prepare_data_with_credentials(data=data, credentials=credentials)
@@ -956,6 +960,7 @@ async def cancel_batch(
                 llm_router=llm_router,
                 model_id=model_from_id,
                 operation_context="batch cancellation (batch created with model)",
+                team_id=user_api_key_dict.team_id,
             )
 
             original_batch_id: Final = get_original_file_id(batch_id)

litellm/proxy/openai_files_endpoints/common_utils.py

@@ -346,6 +346,7 @@ def get_credentials_for_model(
     llm_router,  # Router instance
     model_id: str,
     operation_context: str = "file operation",
+    team_id: "str | None" = None,
 ):
     """
     Retrieve API credentials for a model from the LLM Router.
@@ -354,6 +355,8 @@ def get_credentials_for_model(
         llm_router: LiteLLM Router instance
         model_id: Model name or deployment ID
         operation_context: Description for error messages (e.g., "file upload", "batch creation")
+        team_id: Caller's team id; unlocks that team's own deployments and keeps
+            shared model names from resolving another team's credentials
 
     Returns:
         Dictionary with credentials (api_key, api_base, custom_llm_provider, etc.)
@@ -369,7 +372,7 @@ def get_credentials_for_model(
             detail={"error": "Router not initialized. Cannot use model-based routing."},
         )
 
-    credentials: Final = llm_router.get_deployment_credentials_with_provider(model_id=model_id)
+    credentials: Final = llm_router.get_deployment_credentials_with_provider(model_id=model_id, team_id=team_id)
 
     if credentials is None:
         raise HTTPException(
@@ -539,7 +542,16 @@ def add_internal_model_credentials(
     if model_id is None:
         return
     try:
-        credentials: Final = llm_router.get_deployment_credentials_with_provider(model_id=model_id)
+        # Server-side snapshot for a deployment the router already picked, so it
+        # resolves with the deployment's own owner team; the resolver's team
+        # guard would otherwise drop a team-owned (BYOK) deployment here and
+        # silently lose that batch's cost.
+        deployment: Final = llm_router.get_deployment(model_id=model_id)
+        model_info: Final = deployment.model_info if deployment is not None else None
+        owner_team_id: Final = model_info.get("team_id") if model_info is not None else None
+        credentials: Final = llm_router.get_deployment_credentials_with_provider(
+            model_id=model_id, team_id=owner_team_id
+        )
     except Exception:  # noqa: BLE001  # the snapshot only enables cost accounting; a batch whose deployment no longer resolves must still be retrievable
         return
     if credentials is None:

litellm/router.py

@@ -6468,10 +6468,15 @@ class Router:
                 request_kwargs=kwargs,
             )
 
-            selected_deployment_id: Final = (deployment.get("model_info") or {}).get("id")
+            selected_model_info: Final = deployment.get("model_info") or {}
+            selected_deployment_id: Final = selected_model_info.get("id")
             data: Final = deployment["litellm_params"].copy()
+            # async_get_available_deployment already team-authorized this
+            # deployment; re-resolve with its owner team so the resolver's
+            # team guard doesn't reject the deployment it was handed.
             resolved_credentials: Final = self.get_deployment_credentials_with_provider(
-                model_id=selected_deployment_id or model
+                model_id=selected_deployment_id or model,
+                team_id=selected_model_info.get("team_id"),
             )
             if resolved_credentials is not None:
                 data.update(resolved_credentials)
@@ -10204,6 +10209,67 @@ class Router:
                     raise Exception(f"Model Name invalid - {type(model)}")
         return None
 
+    def _resolve_unblocked_deployment(self, model_id: str, team_id: "str | None" = None) -> "Deployment | None":
+        """
+        Resolve a model id, model-group alias, or wildcard pattern to a single
+        deployment, returning None when nothing matches or the match is paused
+        via ``LiteLLM_ProxyModelTable.blocked``.
+
+        Both the credential resolver and the alias-to-model resolver delegate
+        here so a mixed model group can never hand one caller the credentials
+        of one deployment and the model of another. Every lookup path -
+        exact deployment id, model-group name, and wildcard - skips
+        deployments owned by a team other than ``team_id``, so a caller who
+        knows another team's deployment id cannot resolve its credentials or
+        model; passing ``team_id`` also unlocks that team's own deployments
+        (exact team public model name and team wildcard patterns).
+        """
+        deployment = self.get_deployment(model_id=model_id)
+        if deployment is not None and not self._deployment_usable_by_team(deployment, team_id):
+            deployment = None
+
+        if deployment is None:
+            deployment = self._get_model_group_deployment_usable_by_team(model_group_name=model_id, team_id=team_id)
+
+        # Team-scoped deployments whose team public model name exactly matches
+        # model_id (wildcard team names are matched via team_pattern_routers
+        # below).
+        if deployment is None and team_id is not None:
+            team_indices = self.team_model_to_deployment_indices.get((team_id, model_id)) or ()
+            team_match = next((self.model_list[idx] for idx in team_indices), None)
+            if isinstance(team_match, dict):
+                deployment = Deployment(**team_match)
+            elif isinstance(team_match, Deployment):
+                deployment = team_match
+
+        # Wildcard pattern matches. Team wildcard matches take priority so a
+        # global pattern (e.g. "openai/*") doesn't shadow the team's own entry.
+        if deployment is None:
+            team_pattern_router = self.team_pattern_routers.get(team_id) if team_id is not None else None
+            team_wildcard_match = (
+                next(iter(team_pattern_router.route(model_id) or ()), None) if team_pattern_router else None
+            )
+            wildcard_match = (
+                team_wildcard_match
+                if team_wildcard_match is not None
+                else next(
+                    (
+                        wildcard_model
+                        for wildcard_model in (self.pattern_router.route(model_id) or ())
+                        if self._deployment_usable_by_team(wildcard_model, team_id)
+                    ),
+                    None,
+                )
+            )
+            if isinstance(wildcard_match, dict):
+                deployment = Deployment(**wildcard_match)
+            elif isinstance(wildcard_match, Deployment):
+                deployment = wildcard_match
+
+        if deployment is None or self._is_deployment_blocked(deployment):
+            return None
+        return deployment
+
     @staticmethod
     def _deployment_usable_by_team(model: Mapping | Deployment, team_id: str | None) -> bool:
         """
@@ -10352,10 +10418,11 @@ class Router:
             model_id: Model ID or model name from model_list (e.g., "gpt-4o-litellm")
             team_id: Optional team id of the caller. When set, team-scoped
                 deployments (indexed by team public model name, including team
-                wildcard models like "openai/*") are also considered. Name and
-                wildcard lookups never resolve a deployment owned by a
-                different team, so shared model names can't leak another
-                team's credentials.
+                wildcard models like "openai/*") are also considered. No lookup
+                path - exact deployment id, model-group name, or wildcard -
+                ever resolves a deployment owned by a different team, so
+                neither shared model names nor known deployment ids can leak
+                another team's credentials.
 
         Returns:
             Dictionary containing api_key, api_base, custom_llm_provider, etc.
@@ -10367,43 +10434,8 @@ class Router:
             credentials = router.get_deployment_credentials_with_provider("gpt-4o-litellm")
             # Returns: {"api_key": "sk-...", "custom_llm_provider": "openai", "model": "gpt-4o", ...}
         """
-        # Try to get deployment by model_id first
-        deployment = self.get_deployment(model_id=model_id)
-
-        # If not found, try by model_group_name
+        deployment: Final = self._resolve_unblocked_deployment(model_id=model_id, team_id=team_id)
         if deployment is None:
-            deployment = self._get_model_group_deployment_usable_by_team(model_group_name=model_id, team_id=team_id)
-
-        # If not found, check team-scoped deployments whose team public model
-        # name exactly matches model_id (wildcard team names are matched via
-        # team_pattern_routers below).
-        if deployment is None and team_id is not None:
-            team_indices: Final = self.team_model_to_deployment_indices.get((team_id, model_id), [])
-            if team_indices:
-                team_model: Final = self.model_list[team_indices[0]]
-                deployment = Deployment(**team_model) if isinstance(team_model, dict) else team_model
-
-        # If still not found, check for wildcard pattern matches. Team wildcard
-        # matches take priority so a global pattern (e.g. "openai/*") doesn't
-        # shadow the team's own entry.
-        if deployment is None:
-            team_pattern_router: Final = self.team_pattern_routers.get(team_id) if team_id is not None else None
-            team_wildcard_models: Final = (team_pattern_router.route(model_id) or []) if team_pattern_router else []
-            global_wildcard_models: Final = [
-                wildcard_model
-                for wildcard_model in (self.pattern_router.route(model_id) or [])
-                if self._deployment_usable_by_team(wildcard_model, team_id)
-            ]
-            potential_wildcard_models: Final = team_wildcard_models or global_wildcard_models
-            if potential_wildcard_models:
-                # Use the first matching wildcard deployment
-                deployment_dict: Final = potential_wildcard_models[0]
-                if isinstance(deployment_dict, dict):
-                    deployment = Deployment(**deployment_dict)
-                elif isinstance(deployment_dict, Deployment):
-                    deployment = deployment_dict
-
-        if deployment is None or self._is_deployment_blocked(deployment):
             return None
 
         # Get basic credentials

tests/test_litellm/proxy/batches_endpoints/test_endpoints.py

@@ -160,7 +160,7 @@ class Harness:
         return dict(self.router_acreate.call_args.kwargs)
 
 
-def _creds_lookup(*, model_id: str) -> Dict[str, str]:
+def _creds_lookup(*, model_id: str, team_id: Optional[str] = None) -> Dict[str, str]:
     # KeyError on an unknown/hardcoded model_id - the bug cannot hide.
     return dict(CREDS[model_id])
 
@@ -273,7 +273,7 @@ async def test_create__model_encoded_file_id(harness):
     harness.router_acreate.assert_not_called()
 
     # 2. CREDENTIALS - resolved for the model decoded FROM the file id.
-    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
     # 3. SEAM PAYLOAD - exact, whole dict. A new forwarded key breaks this.
     assert harness.acreate_kwargs() == {
@@ -329,7 +329,7 @@ async def test_create__model_encoded_file_id__resolver_gets_decoded_model(harnes
 
     await call_create(harness)
 
-    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
 
 # =========================================================================== #
@@ -353,7 +353,7 @@ async def test_create__model_from_body(harness):
 
     assert harness.litellm_acreate.call_count == 1
     harness.router_acreate.assert_not_called()
-    harness.creds_resolver.assert_called_once_with(model_id="vertex-model")
+    harness.creds_resolver.assert_called_once_with(model_id="vertex-model", team_id=None)
     payload = harness.acreate_kwargs()
     assert payload["custom_llm_provider"] == "vertex_ai"
     assert payload["input_file_id"] == "file-plain"
@@ -373,7 +373,7 @@ async def test_create__model_from_header(harness):
 
     await call_create(harness, headers={"x-litellm-model": "vertex-model"})
 
-    harness.creds_resolver.assert_called_once_with(model_id="vertex-model")
+    harness.creds_resolver.assert_called_once_with(model_id="vertex-model", team_id=None)
     harness.router_acreate.assert_not_called()
 
 
@@ -390,7 +390,7 @@ async def test_create__model_from_query(harness):
 
     await call_create(harness, query={"model": "vertex-model"})
 
-    harness.creds_resolver.assert_called_once_with(model_id="vertex-model")
+    harness.creds_resolver.assert_called_once_with(model_id="vertex-model", team_id=None)
     harness.router_acreate.assert_not_called()
 
 
@@ -413,7 +413,7 @@ async def test_create__body_model_beats_header_and_query(harness):
         query={"model": "vertex-model"},
     )
 
-    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
 
 # =========================================================================== #
@@ -780,7 +780,7 @@ async def test_create__model_encoded_beats_unified(harness):
 
     assert harness.litellm_acreate.call_count == 1
     harness.router_acreate.assert_not_called()
-    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
 
 # =========================================================================== #
@@ -826,7 +826,7 @@ async def test_create__model_encoded_beats_loadbalancing(harness):
 
     assert harness.litellm_acreate.call_count == 1
     harness.router_acreate.assert_not_called()
-    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
 
 @pytest.mark.asyncio
@@ -1162,6 +1162,7 @@ def retrieve_harness():
 
     router = MagicMock(spec=Router)
     router.aretrieve_batch = AsyncMock(return_value=make_batch())
+    router.get_deployment = MagicMock(return_value=None)
     router.get_deployment_credentials_with_provider = MagicMock(side_effect=_creds_lookup)
 
     pre_call = AsyncMock(side_effect=lambda **kw: (data_holder["data"], MagicMock()))
@@ -1260,7 +1261,7 @@ async def test_retrieve__model_encoded_id(retrieve_harness):
     retrieve_harness.router_aretrieve.assert_not_called()
 
     # 2. CREDENTIALS - resolved for the model decoded FROM the batch id.
-    retrieve_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    retrieve_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
     # 3. SEAM PAYLOAD - exact, whole dict forwarded to the provider call.
     #    Note `model` is the DECODED model, not the deployment from creds: the
@@ -1318,7 +1319,7 @@ async def test_retrieve__model_encoded_beats_loadbalancing(retrieve_harness):
 
     assert retrieve_harness.litellm_aretrieve.call_count == 1
     retrieve_harness.router_aretrieve.assert_not_called()
-    retrieve_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    retrieve_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
 
 # --------------------------------------------------------------------------- #
@@ -1340,7 +1341,7 @@ async def test_retrieve__unified_batch_id_routes_to_router(retrieve_harness):
     # Credentials are resolved for the deployment behind the unified id so the batch's
     # output file can be read for cost accounting. This id resolves to nothing here, and
     # the retrieve must still serve the batch rather than fail on the lookup.
-    retrieve_harness.creds_resolver.assert_called_once_with(model_id="gpt-4o-mini")
+    retrieve_harness.creds_resolver.assert_called_once_with(model_id="gpt-4o-mini", team_id=None)
 
     # router receives the (still-encoded) batch id verbatim - this layer does
     # not decode it for the unified path.
@@ -1815,7 +1816,7 @@ async def test_list__model_from_body_routes_and_encodes(list_harness):
 
     assert list_harness.litellm_alist.call_count == 1
     list_harness.router_alist.assert_not_called()
-    list_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    list_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
     assert resp.data[0].id == encode_file_id_with_model("batch-1", "azure/gpt-4o", id_type="batch")
     assert resp.data[1].id == encode_file_id_with_model("batch-2", "azure/gpt-4o", id_type="batch")
 
@@ -2107,7 +2108,7 @@ async def test_cancel__model_encoded_id(cancel_harness):
     cancel_harness.router_acancel.assert_not_called()
 
     # CREDENTIALS - resolved for the model decoded from the batch id.
-    cancel_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    cancel_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
     # SEAM PAYLOAD - exact dict. NOTE current behavior: `model` is the
     # DEPLOYMENT name from creds, NOT the decoded model (cancel, unlike
@@ -2145,7 +2146,7 @@ async def test_cancel__model_encoded_beats_unified(cancel_harness):
 
     assert cancel_harness.litellm_acancel.call_count == 1
     cancel_harness.router_acancel.assert_not_called()
-    cancel_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o")
+    cancel_harness.creds_resolver.assert_called_once_with(model_id="azure/gpt-4o", team_id=None)
 
 
 # --------------------------------------------------------------------------- #

tests/test_litellm/proxy/openai_files_endpoint/test_files_common_utils.py

@@ -377,6 +377,7 @@ def test_add_internal_model_credentials_attaches_an_immutable_snapshot():
     )
 
     router = MagicMock()
+    router.get_deployment = MagicMock(return_value=None)
     router.get_deployment_credentials_with_provider = MagicMock(
         return_value={"s3_bucket_name": "configured-bucket", "aws_region_name": "us-east-1"}
     )
@@ -389,7 +390,37 @@ def test_add_internal_model_credentials_attaches_an_immutable_snapshot():
     assert isinstance(snapshot, MappingProxyType)
     with pytest.raises(TypeError):
         snapshot["s3_bucket_name"] = "attacker-bucket"
-    router.get_deployment_credentials_with_provider.assert_called_once_with(model_id="deployment-1")
+    router.get_deployment_credentials_with_provider.assert_called_once_with(model_id="deployment-1", team_id=None)
+
+
+def test_add_internal_model_credentials_snapshots_a_team_owned_deployment():
+    """The router already picked this deployment, so the snapshot resolves with the
+    deployment's own owner team. Without that, the resolver's team guard drops a
+    team-owned (BYOK) deployment and the batch's cost is silently never recorded."""
+    from litellm import Router
+    from litellm.proxy.openai_files_endpoints.common_utils import (
+        add_internal_model_credentials,
+    )
+
+    deployment_id = "team-owned-bedrock-deployment"
+    router = Router(
+        model_list=[
+            {
+                "model_name": "bedrock-batch-haiku",
+                "litellm_params": {
+                    "model": "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
+                    "aws_region_name": "us-east-1",
+                    "s3_bucket_name": "team-batch-bucket",
+                },
+                "model_info": {"id": deployment_id, "team_id": "team-1"},
+            }
+        ]
+    )
+    data = {"batch_id": "unified-batch-id"}
+
+    add_internal_model_credentials(data=data, llm_router=router, model_id=deployment_id)
+
+    assert data["_litellm_internal_model_credentials"]["s3_bucket_name"] == "team-batch-bucket"
 
 
 @pytest.mark.parametrize(

tests/test_litellm/proxy/vector_store_endpoints/test_vector_store_endpoints.py

@@ -302,7 +302,7 @@ async def test_vector_store_file_list_resolves_credentials_from_model_query_para
     assert result["model"] == "openai/gpt-4o-mini"
     assert "custom_llm_provider" not in result
     llm_router.get_deployment_credentials_with_provider.assert_called_once_with(
-        model_id="team-openai"
+        model_id="team-openai", team_id=None
     )
 
 

tests/test_litellm/proxy/vector_store_endpoints/test_vector_store_tenant_guard.py

@@ -138,7 +138,7 @@ async def test_vector_store_file_list_resolves_managed_vector_store_before_team_
 
     llm_router = MagicMock()
 
-    def get_credentials(model_id):
+    def get_credentials(model_id, team_id=None):
         return {
             "api_key": f"sk-{model_id}",
             "api_base": "https://api.openai.com/v1",
@@ -171,7 +171,7 @@ async def test_vector_store_file_list_resolves_managed_vector_store_before_team_
     assert captured_data["api_key"] == "sk-managed-deployment"
     assert captured_data["model"] == "openai/managed-deployment"
     llm_router.get_deployment_credentials_with_provider.assert_called_once_with(
-        model_id="managed-deployment"
+        model_id="managed-deployment", team_id=None
     )
 
 

tests/test_litellm/test_router.py

@@ -7884,6 +7884,110 @@ def test_is_deployment_blocked_static_helper_reflects_blocked_flag():
     )
 
 
+def test_deployment_credentials_are_scoped_to_the_callers_team():
+    """
+    Model and credential resolution must pick the SAME deployment for a caller.
+
+    Regression: with a team-owned deployment listed before a shared one under
+    the same alias, an unscoped alias lookup returned the team deployment's
+    model while the team-aware credential resolver returned the shared
+    deployment's credentials, so an outside caller's batch reached the provider
+    with team A's private model id on the shared account.
+    """
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": "bedrock-batch",
+                "litellm_params": {
+                    "model": "bedrock/team-a-private-model",
+                    "aws_region_name": "team-a-region",
+                },
+                "model_info": {
+                    "id": "team-a-dep",
+                    "team_id": "team-a",
+                    "team_public_model_name": "bedrock-batch",
+                },
+            },
+            {
+                "model_name": "bedrock-batch",
+                "litellm_params": {
+                    "model": "bedrock/shared-model",
+                    "aws_region_name": "shared-region",
+                },
+                "model_info": {"id": "shared-dep"},
+            },
+        ]
+    )
+
+    for team_id, expected_model, expected_region in [
+        (None, "bedrock/shared-model", "shared-region"),
+        ("team-b", "bedrock/shared-model", "shared-region"),
+        ("team-a", "bedrock/team-a-private-model", "team-a-region"),
+    ]:
+        credentials = router.get_deployment_credentials_with_provider(model_id="bedrock-batch", team_id=team_id)
+        assert credentials is not None
+        assert credentials["model"] == expected_model, f"team_id={team_id}"
+        assert credentials["aws_region_name"] == expected_region, (
+            f"team_id={team_id}: credentials came from a different deployment than the model"
+        )
+
+    # A caller who knows another team's exact deployment id must not resolve
+    # its credentials through it either.
+    for outsider_team_id in [None, "team-b"]:
+        assert router.get_deployment_credentials_with_provider(model_id="team-a-dep", team_id=outsider_team_id) is None
+    own_team_credentials = router.get_deployment_credentials_with_provider(model_id="team-a-dep", team_id="team-a")
+    assert own_team_credentials is not None
+    assert own_team_credentials["model"] == "bedrock/team-a-private-model"
+
+
+def test_resolve_unblocked_deployment_resolves_alias_id_and_wildcard():
+    """
+    _resolve_unblocked_deployment underpins both the credential resolver and the
+    batch-create alias swap, so it must resolve a deployment by model-group
+    alias, by deployment id, and by wildcard pattern, returning a Deployment
+    whose litellm_params carry the real provider model.
+    """
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": "bedrock-batch-haiku",
+                "litellm_params": {
+                    "model": "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
+                },
+                "model_info": {"id": "bedrock-batch-dep-0"},
+            },
+            {
+                "model_name": "openai/*",
+                "litellm_params": {"model": "openai/*"},
+            },
+        ]
+    )
+
+    by_alias = router._resolve_unblocked_deployment(model_id="bedrock-batch-haiku")
+    assert by_alias is not None
+    assert (
+        by_alias.litellm_params.model
+        == "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
+    )
+
+    by_id = router._resolve_unblocked_deployment(model_id="bedrock-batch-dep-0")
+    assert by_id is not None
+    assert by_id.model_info.id == "bedrock-batch-dep-0"
+
+    by_wildcard = router._resolve_unblocked_deployment(model_id="openai/gpt-4o")
+    assert by_wildcard is not None
+    assert by_wildcard.litellm_params.model == "openai/gpt-4o"
+
+
+def test_resolve_unblocked_deployment_returns_none_for_unknown_and_blocked():
+    router = _router_with_two_deployments([True, False])
+    assert router._resolve_unblocked_deployment(model_id="missing") is None
+    assert router._resolve_unblocked_deployment(model_id="dep-0") is None
+    unblocked = router._resolve_unblocked_deployment(model_id="dep-1")
+    assert unblocked is not None
+    assert unblocked.model_info.id == "dep-1"
+
+
 class TestRouterRequestTimeoutPropagation:
     """litellm_settings.request_timeout must act as an independent per-attempt timeout.