diff --git a/basedpyright-code-budget.json b/basedpyright-code-budget.json
index 0b0a61192e6..57ca267e504 100644
--- a/basedpyright-code-budget.json
+++ b/basedpyright-code-budget.json
@@ -84,7 +84,7 @@
     "limit": 56
   },
   "reportPrivateUsage": {
-    "limit": 1808
+    "limit": 1804
   },
   "reportRedeclaration": {
     "limit": 8
@@ -135,7 +135,7 @@
     "limit": 21
   },
   "reportUnusedFunction": {
-    "limit": 138
+    "limit": 136
   },
   "reportUnusedImport": {
     "limit": 542
diff --git a/litellm/router.py b/litellm/router.py
index 3f450661946..3095039ae20 100644
--- a/litellm/router.py
+++ b/litellm/router.py
@@ -403,6 +403,10 @@ _NO_SESSION_KWARGS: Final[Mapping[str, Mapping[str, object]]] = MappingProxyType
 _SESSION_ADAPTER: Final = TypeAdapter(Mapping[str, object])
 
 
+def _as_retry_skipped_deployment_ids(value: object) -> tuple[str, ...]:
+    return tuple(item for item in value if isinstance(item, str)) if isinstance(value, tuple) else ()
+
+
 def _with_router_resolved_session_model(session: object, model_name: str) -> Mapping[str, Mapping[str, object]]:
     """
     Realtime client-secret requests carry the model inside ``session`` as well, and the caller's copy of it still
@@ -7458,6 +7462,21 @@ class Router:
                     Context_Policy_Fallbacks={content_policy_fallbacks}",
             )
 
+    @staticmethod
+    def _deployment_ids_to_skip_on_retry(exception: Exception, already_skipped: object) -> tuple[str, ...]:
+        failed_deployment_id: Final[str | None] = getattr(exception, "failed_deployment_id", None)
+        status_code: Final = getattr(exception, "status_code", None)
+        if not failed_deployment_id or not isinstance(status_code, int):
+            return ()
+        if litellm._should_retry(status_code):  # pyright: ignore[reportPrivateUsage]  # as in should_retry_this_error
+            return ()
+        already_skipped_ids: Final = _as_retry_skipped_deployment_ids(already_skipped)
+        skipped: Final = tuple(sorted(frozenset((*already_skipped_ids, failed_deployment_id))))
+        verbose_router_logger.debug(
+            "Retry skips deployments that already answered %s to this request: %s", status_code, skipped
+        )
+        return skipped
+
     @tracer.wrap()
     async def async_function_with_retries(self, *args, **kwargs):
         verbose_router_logger.debug("Inside async function with retries.")
@@ -7553,6 +7572,12 @@ class Router:
             ## LOGGING
             if num_retries > 0:
                 kwargs = self.log_retry(kwargs=kwargs, e=original_exception)
+                first_skipped_ids: Final = self._deployment_ids_to_skip_on_retry(
+                    exception=original_exception,
+                    already_skipped=kwargs.get("_retry_skipped_deployment_ids"),
+                )
+                if first_skipped_ids:
+                    kwargs["_retry_skipped_deployment_ids"] = first_skipped_ids  # rebind-ok: the next attempt reads it
             else:
                 raise
 
@@ -7622,6 +7647,12 @@ class Router:
                         except Exception:
                             raise e
 
+                    skipped_ids = self._deployment_ids_to_skip_on_retry(
+                        exception=e,
+                        already_skipped=kwargs.get("_retry_skipped_deployment_ids"),
+                    )
+                    if skipped_ids:
+                        kwargs["_retry_skipped_deployment_ids"] = skipped_ids  # rebind-ok: the next attempt reads it
                     _timeout = self._time_to_sleep_before_retry(
                         e=e,
                         remaining_retries=remaining_retries,
@@ -12454,7 +12485,7 @@ class Router:
 
         ## ORDER FILTERING ## -> if user set 'order' in deployments, return deployments with lowest order (e.g. order=1 > order=2)
         _target_order: Final = (request_kwargs or {}).pop("_target_order", None)
-        healthy_deployments = litellm.utils._get_order_filtered_deployments(
+        healthy_deployments = litellm.utils.get_order_filtered_deployments(
             cast(list[dict], healthy_deployments), target_order=_target_order
         )
 
@@ -12462,11 +12493,24 @@ class Router:
         ## this request via weighted-failover. Always honored, regardless of the
         ## router-level flag, so a stale exclusion key on kwargs cannot escape.
         _excluded_deployment_ids: Final = (request_kwargs or {}).pop("_excluded_deployment_ids", None)
-        healthy_deployments = litellm.utils._get_excluded_filtered_deployments(
+        healthy_deployments = litellm.utils.get_excluded_filtered_deployments(
             cast(list[dict], healthy_deployments),
             excluded_deployment_ids=_excluded_deployment_ids,
         )
 
+        ## RETRY SKIP ## -> drop deployments that already refused this request with a
+        ## non-retryable status, unless that leaves nothing, so the caller still gets
+        ## the provider's own error instead of a no-deployments error.
+        _retry_skipped_deployment_ids: Final = _as_retry_skipped_deployment_ids(
+            request_kwargs.pop("_retry_skipped_deployment_ids", None) if request_kwargs else None
+        )
+        healthy_deployments = (
+            litellm.utils.get_excluded_filtered_deployments(
+                healthy_deployments, excluded_deployment_ids=_retry_skipped_deployment_ids
+            )
+            or healthy_deployments
+        )
+
         if len(healthy_deployments) == 0:
             exception: Final = await async_raise_no_deployment_exception(
                 litellm_router_instance=self,
@@ -13359,7 +13403,7 @@ class Router:
 
         ## ORDER FILTERING ## -> if user set 'order' in deployments, return deployments with lowest order (e.g. order=1 > order=2)
         _target_order: Final = (request_kwargs or {}).pop("_target_order", None)
-        healthy_deployments = litellm.utils._get_order_filtered_deployments(
+        healthy_deployments = litellm.utils.get_order_filtered_deployments(
             healthy_deployments, target_order=_target_order
         )
 
@@ -13367,11 +13411,22 @@ class Router:
         ## this request via weighted-failover. See async counterpart in
         ## async_get_healthy_deployments for details.
         _excluded_deployment_ids: Final = (request_kwargs or {}).pop("_excluded_deployment_ids", None)
-        healthy_deployments = litellm.utils._get_excluded_filtered_deployments(
+        healthy_deployments = litellm.utils.get_excluded_filtered_deployments(
             healthy_deployments,
             excluded_deployment_ids=_excluded_deployment_ids,
         )
 
+        ## RETRY SKIP ## -> see async counterpart in async_get_healthy_deployments.
+        _retry_skipped_deployment_ids: Final = _as_retry_skipped_deployment_ids(
+            request_kwargs.pop("_retry_skipped_deployment_ids", None) if request_kwargs else None
+        )
+        healthy_deployments = (
+            litellm.utils.get_excluded_filtered_deployments(
+                healthy_deployments, excluded_deployment_ids=_retry_skipped_deployment_ids
+            )
+            or healthy_deployments
+        )
+
         if len(healthy_deployments) == 0:
             model_ids = self.get_model_ids(model_name=model)
             _cooldown_time = self.cooldown_cache.get_min_cooldown(
diff --git a/litellm/router_strategy/complexity_router/complexity_router.py b/litellm/router_strategy/complexity_router/complexity_router.py
index d8b9b5ea8b2..7d4497fb6f7 100644
--- a/litellm/router_strategy/complexity_router/complexity_router.py
+++ b/litellm/router_strategy/complexity_router/complexity_router.py
@@ -2835,8 +2835,9 @@ class ComplexityRouter(CustomLogger):
         where the prompt never arrives as messages.
 
         Probed on a COPY of request_kwargs because the owner pops routing bookkeeping off the
-        dict it is handed (`_target_order`, `_excluded_deployment_ids`), and this is a
-        speculative question about a model that may never be picked.
+        dict it is handed (`_target_order`, `_excluded_deployment_ids`,
+        `_retry_skipped_deployment_ids`), and this is a speculative question about a model
+        that may never be picked.
 
         Every way the owner says "nothing here can serve this" is a negative verdict: no healthy
         deployment for the group at all (BadRequestError, which ContextWindowExceededError
diff --git a/litellm/utils.py b/litellm/utils.py
index bc2f4a86f12..ebbe643f6ad 100644
--- a/litellm/utils.py
+++ b/litellm/utils.py
@@ -4889,7 +4889,7 @@ def _get_deployment_order(deployment: dict | Any) -> int | None:
     return order
 
 
-def _get_order_filtered_deployments(healthy_deployments: list[dict], target_order: int | None = None) -> list:
+def get_order_filtered_deployments(healthy_deployments: list[dict], target_order: int | None = None) -> list:
     if target_order is not None:
         return [d for d in healthy_deployments if _get_deployment_order(d) == target_order]
 
@@ -4908,7 +4908,7 @@ def _get_order_filtered_deployments(healthy_deployments: list[dict], target_orde
     return healthy_deployments
 
 
-def _get_excluded_filtered_deployments(
+def get_excluded_filtered_deployments(
     healthy_deployments: list[dict],
     excluded_deployment_ids: Iterable[str] | None = None,
 ) -> list:
@@ -4919,10 +4919,12 @@ def _get_excluded_filtered_deployments(
     across the remaining deployments in the same model group after one of them
     has failed.
 
-    If the filter would leave no deployments, an empty list is returned so the
-    caller raises its usual no-deployments error and the weighted-failover
-    helper falls through to the cross-group fallback path. Returning the
-    original unfiltered list here would re-include the just-failed deployment.
+    If the filter would leave no deployments, an empty list is returned and the
+    caller decides what that means. Weighted failover lets it raise the usual
+    no-deployments error and fall through to the cross-group fallback path; the
+    retry skip in `async_get_healthy_deployments` deliberately falls back to the
+    unfiltered list, so a request every deployment refused still comes back with
+    the provider's own error rather than a no-deployments one.
     """
     if not excluded_deployment_ids:
         return healthy_deployments
diff --git a/tests/e2e/coverage_registry/reliability.yaml b/tests/e2e/coverage_registry/reliability.yaml
index b50551ec105..6b69677d490 100644
--- a/tests/e2e/coverage_registry/reliability.yaml
+++ b/tests/e2e/coverage_registry/reliability.yaml
@@ -7,7 +7,7 @@
 - {id: reliability.retry.timeout.succeeds_within_retries, module: reliability, tier: P0, behavior: retry, variant: timeout, assertions: [succeeds_within_retries], exercised_on: [chat_completions, messages], source: "get_retry_from_policy.py:44", rationale: "Timeout retried per policy"}
 - {id: reliability.retry.429.succeeds_within_retries, module: reliability, tier: P0, behavior: retry, variant: "429", assertions: [succeeds_within_retries], exercised_on: [chat_completions, messages], source: "get_retry_from_policy.py:46", rationale: "429 retried per RateLimitErrorRetries policy"}
 - {id: reliability.retry.auth.succeeds_within_retries, module: reliability, tier: P1, behavior: retry, variant: auth, assertions: [succeeds_within_retries], exercised_on: [chat_completions, messages], source: "get_retry_from_policy.py:42", rationale: "Transient auth glitch retry"}
-- {id: reliability.retry.context_window.succeeds_within_retries, module: reliability, tier: P1, behavior: retry, variant: context_window, assertions: [succeeds_within_retries], exercised_on: [chat_completions, messages], source: "get_retry_from_policy.py:51", rationale: "Multi-attempt on context error"}
+- {id: reliability.retry.context_window.succeeds_within_retries, module: reliability, tier: P1, behavior: retry, variant: context_window, assertions: [succeeds_within_retries], exercised_on: [chat_completions, messages], source: "get_retry_from_policy.py:51", fail_before_fix: proven, rationale: "A context-window 400 under BadRequestErrorRetries retries onto a sibling deployment in the same model group, instead of coming straight back as the 400 the deployment that just refused it returned"}
 - {id: reliability.cooldown.5xx.trips_then_recovers, module: reliability, tier: P0, behavior: cooldown, variant: "5xx", assertions: [trips_then_recovers], exercised_on: [chat_completions, messages], source: "cooldown_handlers.py:40", rationale: "Deployment cools after repeated 5xx, recovers after cooldown_time"}
 - {id: reliability.cooldown.429.trips_then_recovers, module: reliability, tier: P0, behavior: cooldown, variant: "429", assertions: [trips_then_recovers], exercised_on: [chat_completions, messages], source: "cooldown_handlers.py:69", rationale: "Cools on 429, avoids hammering exhausted provider"}
 - {id: reliability.cooldown.auth.trips_then_recovers, module: reliability, tier: P1, behavior: cooldown, variant: auth, assertions: [trips_then_recovers], exercised_on: [chat_completions, messages], source: "cooldown_handlers.py:74", rationale: "Cools on 401 auth error"}
diff --git a/tests/e2e/models.py b/tests/e2e/models.py
index 2c6c0e9bbd4..c01687d0b31 100644
--- a/tests/e2e/models.py
+++ b/tests/e2e/models.py
@@ -291,6 +291,7 @@ class RouterSettingsOverride(BaseModel):
     context_window_fallbacks: list[dict[str, list[str]]] | None = None
     content_policy_fallbacks: list[dict[str, list[str]]] | None = None
     num_retries: int | None = None
+    model_group_retry_policy: dict[str, dict[str, int]] | None = None
     enable_tag_filtering: bool | None = None
 
 
diff --git a/tests/e2e/router/reliability_support.py b/tests/e2e/router/reliability_support.py
index 5822058003c..1efcb1a045b 100644
--- a/tests/e2e/router/reliability_support.py
+++ b/tests/e2e/router/reliability_support.py
@@ -73,10 +73,25 @@ def create_always_timing_out_deployment(proxy: ProxyClient, name: str) -> str:
     )
 
 
+def create_always_picked_small_context_deployment(proxy: ProxyClient, name: str) -> str:
+    """The always-picked half of a retry pair on the smallest-context model OpenAI
+    still serves: it holds all of the model group's shuffle weight, so an oversized
+    prompt opens on it and earns a real context-window refusal, which never benches
+    a deployment, so only the retry itself can steer the request off it."""
+    return proxy.register_model(
+        ModelNewBody(
+            model_name=name,
+            litellm_params=LiteLLMParamsBody(model=SMALL_CONTEXT_MODEL, api_key=REAL_KEY, weight=1),
+            model_info=ModelInfoBody(),
+        )
+    )
+
+
 def create_zero_weight_backup_deployment(proxy: ProxyClient, name: str) -> str:
     """The other half of a retry pair: healthy, but weight 0, so the weighted shuffle
-    never opens on it. It is reachable only once its sibling is benched and the
-    weighted pick falls through to a uniform one over what is left."""
+    never opens on it. It is reachable only once its sibling is out of the running,
+    benched by a cooldown or skipped by the retry, and the weighted pick falls through
+    to a uniform one over what is left."""
     return proxy.register_model(
         ModelNewBody(
             model_name=name,
diff --git a/tests/e2e/router/test_reliability_retries_e2e.py b/tests/e2e/router/test_reliability_retries_e2e.py
index 5441412935c..da45cb46a46 100644
--- a/tests/e2e/router/test_reliability_retries_e2e.py
+++ b/tests/e2e/router/test_reliability_retries_e2e.py
@@ -1,13 +1,17 @@
 """Live e2e: a request that fails on its first deployment is retried inside its own
 model group and still comes back a completion.
 
-The model group is a pair: an always-timing-out deployment that holds all of the
-group's shuffle weight, and a healthy backup at weight 0. The weighted pick always
-opens on the timing-out one, its first Timeout benches it (an
-`allowed_fails_policy` of `TimeoutErrorAllowedFails: 0`), and the retry falls
-through to the only deployment left. So the customer sees a completion and the
-proxy reports that it took a retry to get there, with no random first pick in the
-middle of it.
+Each model group is a pair: a deployment that always refuses and holds all of the
+group's shuffle weight, plus a healthy backup at weight 0. The weighted pick always
+opens on the refusing one, so the customer sees a completion only if the retry
+lands on the backup, and the proxy reports that it took a retry to get there, with
+no random first pick in the middle of it.
+
+The timeout pair relies on cooldown: the first Timeout benches the timing-out
+deployment (an `allowed_fails_policy` of `TimeoutErrorAllowedFails: 0`) and the
+retry falls through to the only deployment left. The context-window pair cannot:
+a 400 never benches a deployment, so the retry policy's `BadRequestErrorRetries`
+has to steer the retry off the deployment that just refused the prompt.
 """
 
 from __future__ import annotations
@@ -16,20 +20,48 @@ import pytest
 
 from complexity_router_client import ComplexityRouterClient
 from e2e_config import unique_marker
+from e2e_http import StreamingResponse
 from lifecycle import ResourceManager
 from models import RouterSettingsOverride
 from reliability_support import (
     chat_override,
     completion_tokens_of,
     content_of,
+    create_always_picked_small_context_deployment,
     create_always_timing_out_deployment,
     create_zero_weight_backup_deployment,
     finish_reason_of,
+    oversized_prompt,
 )
 
 pytestmark = pytest.mark.e2e
 
 
+def assert_retry_landed_on_backup(resp: StreamingResponse) -> None:
+    assert resp.status_code == 200, (
+        f"the retry should have landed on the healthy backup, got {resp.status_code}: {resp.body[:300]}"
+    )
+
+    attempted = resp.headers.get("x-litellm-attempted-retries")
+    assert attempted is not None, "response is missing the x-litellm-attempted-retries header"
+    assert int(attempted) >= 1, (
+        f"x-litellm-attempted-retries is {attempted!r}; a 200 with no retry means the request never "
+        "opened on the refusing deployment, so this proves nothing about retries"
+    )
+
+    content = content_of(resp)
+    finish_reason = finish_reason_of(resp)
+    completion_tokens = completion_tokens_of(resp) or 0
+    assert isinstance(content, str), (
+        f"the retry should have returned a completion body, got content {content!r} (body={resp.body[:300]})"
+    )
+    assert content or (finish_reason == "length" and completion_tokens > 0), (
+        f"the retry returned empty content with finish_reason={finish_reason!r}, "
+        f"completion_tokens={completion_tokens}; empty content is only acceptable when the budget "
+        f"was spent on non-visible reasoning (body={resp.body[:300]})"
+    )
+
+
 class TestReliabilityRetries:
     @pytest.mark.covers("reliability.retry.timeout.succeeds_within_retries")
     def test_timeout_on_first_deployment_succeeds_on_retry(
@@ -49,25 +81,27 @@ class TestReliabilityRetries:
             override=RouterSettingsOverride(num_retries=2),
         )
 
-        assert resp.status_code == 200, (
-            f"the retry should have landed on the healthy backup, got {resp.status_code}: {resp.body[:300]}"
+        assert_retry_landed_on_backup(resp)
+
+    @pytest.mark.covers("reliability.retry.context_window.succeeds_within_retries")
+    def test_context_window_refusal_on_first_deployment_succeeds_on_retry(
+        self, client: ComplexityRouterClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        group = f"reliability-retry-{unique_marker()}"
+        small_context = create_always_picked_small_context_deployment(client.proxy, group)
+        resources.defer(lambda: client.proxy.delete_model(small_context))
+        backup = create_zero_weight_backup_deployment(client.proxy, group)
+        resources.defer(lambda: client.proxy.delete_model(backup))
+
+        resp = chat_override(
+            client.proxy,
+            scoped_key,
+            group,
+            oversized_prompt(unique_marker()),
+            override=RouterSettingsOverride(
+                num_retries=2,
+                model_group_retry_policy={group: {"BadRequestErrorRetries": 2}},
+            ),
         )
 
-        attempted = resp.headers.get("x-litellm-attempted-retries")
-        assert attempted is not None, "response is missing the x-litellm-attempted-retries header"
-        assert int(attempted) >= 1, (
-            f"x-litellm-attempted-retries is {attempted!r}; a 200 with no retry means the request never "
-            "opened on the timing-out deployment, so this proves nothing about retries"
-        )
-
-        content = content_of(resp)
-        finish_reason = finish_reason_of(resp)
-        completion_tokens = completion_tokens_of(resp) or 0
-        assert isinstance(content, str), (
-            f"the retry should have returned a completion body, got content {content!r} (body={resp.body[:300]})"
-        )
-        assert content or (finish_reason == "length" and completion_tokens > 0), (
-            f"the retry returned empty content with finish_reason={finish_reason!r}, "
-            f"completion_tokens={completion_tokens}; empty content is only acceptable when the budget "
-            f"was spent on non-visible reasoning (body={resp.body[:300]})"
-        )
+        assert_retry_landed_on_backup(resp)
diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py
index 7c044310e14..a7d4e66bf73 100644
--- a/tests/test_litellm/test_router.py
+++ b/tests/test_litellm/test_router.py
@@ -39,6 +39,7 @@ from litellm.router import (
     _anthropic_stream_should_drop_pre_content_ping,
     _is_retriable_anthropic_status,
 )
+from litellm.router_strategy import simple_shuffle
 from litellm.types.router import DeploymentTypedDict
 
 
@@ -13116,6 +13117,7 @@ async def test_prompt_management_factory_marks_injection_for_every_deployment(mo
         ({"DefaultRetries": 0}, 502, litellm.BadGatewayError, 1),
         ({"DefaultRetries": 0, "ServiceUnavailableErrorRetries": 1}, 503, litellm.ServiceUnavailableError, 2),
         ({"ServiceUnavailableErrorRetries": 0}, 502, litellm.BadGatewayError, 3),
+        ({"BadRequestErrorRetries": 2}, 400, litellm.BadRequestError, 3),
     ],
 )
 async def test_router_retry_policy_controls_upstream_attempt_count(
@@ -13152,6 +13154,323 @@ async def test_router_retry_policy_controls_upstream_attempt_count(
     assert upstream.call_count == expected_upstream_calls
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "retry_policy,upstream_error",
+    [
+        (
+            {"BadRequestErrorRetries": 2},
+            {
+                "message": "This model's maximum context length is 16385 tokens",
+                "type": "invalid_request_error",
+                "code": "context_length_exceeded",
+            },
+        ),
+        (
+            {"ContentPolicyViolationErrorRetries": 2},
+            {
+                "message": "Your request was rejected as a result of our safety system",
+                "type": "invalid_request_error",
+                "code": "content_policy_violation",
+            },
+        ),
+    ],
+)
+async def test_router_retry_policy_400_retries_on_sibling_deployment(
+    monkeypatch: pytest.MonkeyPatch, retry_policy, upstream_error
+):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": "gpt-5.6",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "api_base": "https://rejecting.local/v1",
+                    "weight": 1,
+                },
+                "model_info": {"id": "rejecting"},
+            },
+            {
+                "model_name": "gpt-5.6",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "api_base": "https://accepting.local/v1",
+                    "weight": 0,
+                },
+                "model_info": {"id": "accepting"},
+            },
+        ],
+        num_retries=2,
+        retry_policy=retry_policy,
+        disable_cooldowns=True,
+    )
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        rejecting = respx_mock.post("https://rejecting.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": upstream_error})
+        )
+        accepting = respx_mock.post("https://accepting.local/v1/chat/completions").mock(
+            return_value=httpx.Response(
+                200,
+                json={
+                    "id": "chatcmpl-lit-7036",
+                    "object": "chat.completion",
+                    "created": 1,
+                    "model": "gpt-5.6",
+                    "choices": [
+                        {"index": 0, "message": {"role": "assistant", "content": "hi back"}, "finish_reason": "stop"}
+                    ],
+                    "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
+                },
+            )
+        )
+        response = await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
+
+    assert rejecting.call_count == 1
+    assert accepting.call_count == 1
+    assert response.choices[0].message.content == "hi back"
+    assert response._hidden_params["additional_headers"]["x-litellm-attempted-retries"] == 1
+
+
+_UPSTREAM_400 = {"message": "upstream refused this request", "type": "invalid_request_error", "code": "bad_request"}
+
+
+def _retry_skip_deployment(deployment_id, host, litellm_params=None, model_info=None):
+    return {
+        "model_name": "gpt-5.6",
+        "litellm_params": {
+            "model": "openai/gpt-5.6",
+            "api_key": "sk-fake",
+            "api_base": f"https://{host}.local/v1",
+            **(litellm_params or {}),
+        },
+        "model_info": {"id": deployment_id, **(model_info or {})},
+    }
+
+
+@pytest.mark.parametrize(
+    "status_code,failed_deployment_id,already_skipped,expected",
+    [
+        (400, "rejecting", None, ("rejecting",)),
+        (403, "rejecting", None, ("rejecting",)),
+        (400, "second", ("first",), ("first", "second")),
+        (400, "first", ("first",), ("first",)),
+        (429, "rejecting", None, ()),
+        (503, "rejecting", None, ()),
+        (408, "rejecting", None, ()),
+        (400, None, None, ()),
+        (None, "rejecting", None, ()),
+        ("400", "rejecting", None, ()),
+        (400, "second", 7, ("second",)),
+        (400, "second", "first", ("second",)),
+        (400, "second", ["first"], ("second",)),
+        (400, "second", ("first", 7), ("first", "second")),
+    ],
+)
+def test_router_deployment_ids_to_skip_on_retry(status_code, failed_deployment_id, already_skipped, expected):
+    exception = Exception("upstream refused this request")
+    exception.status_code = status_code
+    exception.failed_deployment_id = failed_deployment_id
+
+    assert litellm.Router._deployment_ids_to_skip_on_retry(exception, already_skipped) == expected
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        (("first", "second"), ("first", "second")),
+        ((), ()),
+        (("first", 7, None, "second"), ("first", "second")),
+        (None, ()),
+        (7, ()),
+        ("first", ()),
+        (["first"], ()),
+        ({"first": True}, ()),
+        (object(), ()),
+    ],
+)
+def test_router_as_retry_skipped_deployment_ids_keeps_only_a_tuple_of_strings(value, expected):
+    from litellm.router import _as_retry_skipped_deployment_ids
+
+    assert _as_retry_skipped_deployment_ids(value) == expected
+
+
+@pytest.mark.parametrize(
+    "deployment_ids,skipped,expected",
+    [
+        (["rejecting", "sibling"], ("rejecting",), ["sibling"]),
+        (["rejecting"], ("rejecting",), ["rejecting"]),
+        (["rejecting", "sibling"], ("rejecting", "sibling"), ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], (), ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], None, ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], ("absent",), ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], 7, ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], "rejecting", ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], ["rejecting"], ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], {"rejecting": True}, ["rejecting", "sibling"]),
+        (["rejecting", "sibling"], ("rejecting", 7), ["sibling"]),
+    ],
+)
+@pytest.mark.asyncio
+async def test_router_healthy_deployments_keep_the_last_candidate_a_retry_skipped(deployment_ids, skipped, expected):
+    router = litellm.Router(
+        model_list=[_retry_skip_deployment(deployment_id, deployment_id) for deployment_id in deployment_ids],
+        disable_cooldowns=True,
+    )
+    request_kwargs = {"_retry_skipped_deployment_ids": skipped}
+
+    healthy_deployments = await router.async_get_healthy_deployments(model="gpt-5.6", request_kwargs=request_kwargs)
+
+    assert sorted(deployment["model_info"]["id"] for deployment in healthy_deployments) == sorted(expected)
+    assert "_retry_skipped_deployment_ids" not in request_kwargs
+
+
+@pytest.mark.parametrize("client_supplied", [7, "rejecting", ["rejecting"], {"rejecting": True}, object()])
+@pytest.mark.asyncio
+async def test_router_retry_policy_400_keeps_upstream_error_when_a_client_forges_the_skip_list(
+    monkeypatch: pytest.MonkeyPatch, client_supplied
+):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router = litellm.Router(
+        model_list=[_retry_skip_deployment("rejecting", "rejecting"), _retry_skip_deployment("sibling", "sibling")],
+        num_retries=2,
+        retry_policy={"BadRequestErrorRetries": 2},
+        disable_cooldowns=True,
+    )
+
+    with respx.mock as respx_mock:
+        respx_mock.post("https://rejecting.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        respx_mock.post("https://sibling.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        with pytest.raises(litellm.BadRequestError) as raised:
+            await router.acompletion(
+                model="gpt-5.6",
+                messages=[{"role": "user", "content": "hi"}],
+                _retry_skipped_deployment_ids=client_supplied,
+            )
+
+    assert "upstream refused this request" in str(raised.value)
+
+
+@pytest.mark.asyncio
+async def test_router_retry_policy_400_keeps_upstream_error_on_order_fallback_hop(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router = litellm.Router(
+        model_list=[
+            _retry_skip_deployment("order1", "order1", litellm_params={"order": 1}),
+            _retry_skip_deployment("order2", "order2", litellm_params={"order": 2}),
+        ],
+        num_retries=2,
+        retry_policy={"BadRequestErrorRetries": 2},
+        disable_cooldowns=True,
+    )
+
+    with respx.mock as respx_mock:
+        order1 = respx_mock.post("https://order1.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        order2 = respx_mock.post("https://order2.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        with pytest.raises(litellm.BadRequestError) as raised:
+            await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
+
+    assert "upstream refused this request" in str(raised.value)
+    assert "No deployments available" not in str(raised.value)
+    assert order1.call_count >= 1
+    assert order2.call_count >= 1
+
+
+@pytest.mark.asyncio
+async def test_router_retry_policy_400_keeps_upstream_error_when_tags_narrow_the_group(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router = litellm.Router(
+        model_list=[
+            _retry_skip_deployment(
+                "tagged", "tagged", litellm_params={"tags": ["free"]}, model_info={"enable_tag_filtering": True}
+            ),
+            _retry_skip_deployment("untagged", "untagged", model_info={"enable_tag_filtering": True}),
+        ],
+        num_retries=2,
+        retry_policy={"BadRequestErrorRetries": 2},
+        disable_cooldowns=True,
+    )
+
+    with respx.mock as respx_mock:
+        tagged = respx_mock.post("https://tagged.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        untagged = respx_mock.post("https://untagged.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        with pytest.raises(litellm.BadRequestError) as raised:
+            await router.acompletion(
+                model="gpt-5.6",
+                messages=[{"role": "user", "content": "hi"}],
+                metadata={"tags": ["free"]},
+            )
+
+    assert "upstream refused this request" in str(raised.value)
+    assert "No deployments available" not in str(raised.value)
+    assert tagged.call_count == 3
+    assert untagged.call_count == 0
+
+
+@pytest.mark.asyncio
+async def test_router_retry_policy_400_never_returns_to_a_deployment_that_already_refused(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(simple_shuffle.random, "choice", lambda deployments: deployments[0])
+    router = litellm.Router(
+        model_list=[
+            _retry_skip_deployment("first-refuser", "first-refuser", litellm_params={"weight": 1}),
+            _retry_skip_deployment("second-refuser", "second-refuser", litellm_params={"weight": 0}),
+            _retry_skip_deployment("accepting", "accepting", litellm_params={"weight": 0}),
+        ],
+        num_retries=3,
+        retry_policy={"BadRequestErrorRetries": 3},
+        disable_cooldowns=True,
+    )
+
+    with respx.mock as respx_mock:
+        first = respx_mock.post("https://first-refuser.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        second = respx_mock.post("https://second-refuser.local/v1/chat/completions").mock(
+            return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
+        )
+        accepting = respx_mock.post("https://accepting.local/v1/chat/completions").mock(
+            return_value=httpx.Response(
+                200,
+                json={
+                    "id": "chatcmpl-lit-7036",
+                    "object": "chat.completion",
+                    "created": 1,
+                    "model": "gpt-5.6",
+                    "choices": [
+                        {"index": 0, "message": {"role": "assistant", "content": "hi back"}, "finish_reason": "stop"}
+                    ],
+                    "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
+                },
+            )
+        )
+        response = await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
+
+    assert first.call_count == 1
+    assert second.call_count == 1
+    assert accepting.call_count == 1
+    assert response.choices[0].message.content == "hi back"
+
+
 def _make_failure_logging_obj():
     return LiteLLMLogging(
         model="gpt-5.6",
diff --git a/tests/test_litellm/test_router_order_fallback.py b/tests/test_litellm/test_router_order_fallback.py
index fde870e5abe..93895bbde08 100644
--- a/tests/test_litellm/test_router_order_fallback.py
+++ b/tests/test_litellm/test_router_order_fallback.py
@@ -18,10 +18,10 @@ from litellm import Router
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.router_utils.prompt_caching_cache import PromptCachingCache
 from litellm.types.router import RouterRateLimitError
-from litellm.utils import _get_deployment_order, _get_order_filtered_deployments
+from litellm.utils import _get_deployment_order, get_order_filtered_deployments
 
 # ---------------------------------------------------------------------------
-# Unit tests for _get_order_filtered_deployments
+# Unit tests for get_order_filtered_deployments
 # ---------------------------------------------------------------------------
 
 
@@ -42,7 +42,7 @@ class TestGetOrderFilteredDeployments:
             self._make_deployment(2, "b"),
             self._make_deployment(1, "c"),
         ]
-        result = _get_order_filtered_deployments(deps)
+        result = get_order_filtered_deployments(deps)
         assert len(result) == 2
         assert all(d["model_info"]["id"] in ("a", "c") for d in result)
 
@@ -52,7 +52,7 @@ class TestGetOrderFilteredDeployments:
             self._make_deployment(2, "b"),
             self._make_deployment(3, "c"),
         ]
-        result = _get_order_filtered_deployments(deps, target_order=2)
+        result = get_order_filtered_deployments(deps, target_order=2)
         assert len(result) == 1
         assert result[0]["model_info"]["id"] == "b"
 
@@ -61,7 +61,7 @@ class TestGetOrderFilteredDeployments:
             self._make_deployment(1, "a"),
             self._make_deployment(2, "b"),
         ]
-        result = _get_order_filtered_deployments(deps, target_order=99)
+        result = get_order_filtered_deployments(deps, target_order=99)
         assert result == []
 
     def test_target_order_no_match_does_not_reselect_lower_order(self):
@@ -70,7 +70,7 @@ class TestGetOrderFilteredDeployments:
             self._make_deployment(2, "b"),
         ]
         remaining_after_pre_call = [deps[0]]
-        result = _get_order_filtered_deployments(remaining_after_pre_call, target_order=2)
+        result = get_order_filtered_deployments(remaining_after_pre_call, target_order=2)
         assert result == []
 
     def test_no_order_set_returns_all(self):
@@ -78,11 +78,11 @@ class TestGetOrderFilteredDeployments:
             self._make_deployment(None, "a"),
             self._make_deployment(None, "b"),
         ]
-        result = _get_order_filtered_deployments(deps)
+        result = get_order_filtered_deployments(deps)
         assert len(result) == 2
 
     def test_empty_list(self):
-        result = _get_order_filtered_deployments([])
+        result = get_order_filtered_deployments([])
         assert result == []
 
     def test_single_order_returns_all_with_that_order(self):
@@ -90,7 +90,7 @@ class TestGetOrderFilteredDeployments:
             self._make_deployment(1, "a"),
             self._make_deployment(1, "b"),
         ]
-        result = _get_order_filtered_deployments(deps)
+        result = get_order_filtered_deployments(deps)
         assert len(result) == 2
 
 
diff --git a/tests/test_litellm/test_router_weighted_failover.py b/tests/test_litellm/test_router_weighted_failover.py
index 162312a8c67..9f05654f23f 100644
--- a/tests/test_litellm/test_router_weighted_failover.py
+++ b/tests/test_litellm/test_router_weighted_failover.py
@@ -15,11 +15,11 @@ import pytest
 
 import litellm
 from litellm import Router
-from litellm.utils import _get_excluded_filtered_deployments
+from litellm.utils import get_excluded_filtered_deployments
 
 
 # ---------------------------------------------------------------------------
-# Unit tests for _get_excluded_filtered_deployments
+# Unit tests for get_excluded_filtered_deployments
 # ---------------------------------------------------------------------------
 
 
@@ -37,17 +37,17 @@ def _make_dep(dep_id: str, weight: Optional[int] = None) -> dict:
 class TestGetExcludedFilteredDeployments:
     def test_no_excluded_returns_all(self):
         deps = [_make_dep("a"), _make_dep("b")]
-        result = _get_excluded_filtered_deployments(deps, excluded_deployment_ids=None)
+        result = get_excluded_filtered_deployments(deps, excluded_deployment_ids=None)
         assert len(result) == 2
 
     def test_empty_excluded_returns_all(self):
         deps = [_make_dep("a"), _make_dep("b")]
-        result = _get_excluded_filtered_deployments(deps, excluded_deployment_ids=[])
+        result = get_excluded_filtered_deployments(deps, excluded_deployment_ids=[])
         assert len(result) == 2
 
     def test_drops_excluded(self):
         deps = [_make_dep("a"), _make_dep("b"), _make_dep("c")]
-        result = _get_excluded_filtered_deployments(deps, excluded_deployment_ids=["b"])
+        result = get_excluded_filtered_deployments(deps, excluded_deployment_ids=["b"])
         ids = sorted(d["model_info"]["id"] for d in result)
         assert ids == ["a", "c"]
 
@@ -57,12 +57,12 @@ class TestGetExcludedFilteredDeployments:
         # error. Returning the original list here would re-include the
         # just-failed deployment and let weighted failover re-pick it.
         deps = [_make_dep("a"), _make_dep("b")]
-        result = _get_excluded_filtered_deployments(deps, excluded_deployment_ids=["a", "b"])
+        result = get_excluded_filtered_deployments(deps, excluded_deployment_ids=["a", "b"])
         assert result == []
 
     def test_excluded_set_with_unknown_ids(self):
         deps = [_make_dep("a"), _make_dep("b")]
-        result = _get_excluded_filtered_deployments(deps, excluded_deployment_ids=["zzz"])
+        result = get_excluded_filtered_deployments(deps, excluded_deployment_ids=["zzz"])
         assert len(result) == 2
 
     def test_handles_missing_model_info(self):
@@ -70,7 +70,7 @@ class TestGetExcludedFilteredDeployments:
             {"model_name": "x", "litellm_params": {"model": "gpt-4o"}},  # no model_info
             _make_dep("b"),
         ]
-        result = _get_excluded_filtered_deployments(deps, excluded_deployment_ids=["b"])
+        result = get_excluded_filtered_deployments(deps, excluded_deployment_ids=["b"])
         assert len(result) == 1