fix(proxy): satisfy lint and type-discipline gates for the block gate

Use read-only Sequence/Mapping annotations so the fallback types stop tripping
LIT001, fold the blocked-and-no-fallback decision into one Router method so the
proxy makes a single cross-class call, validate router fallbacks through the
TypeAdapter, and register the bounded reachability walker with the recursion
detector.
This commit is contained in:
Awshesh12 2026-08-13 11:11:51 +00:00
parent 37edd13e5b
commit 298ab31d22
4 changed files with 41 additions and 18 deletions

View file

@ -1,5 +1,5 @@
import asyncio
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Literal
import httpx
@ -74,9 +74,16 @@ def _is_a2a_agent_model(model_name: Any) -> bool:
return isinstance(model_name, str) and model_name.startswith("a2a/")
def _validated_block_fallbacks(raw: object) -> Sequence[Mapping[str, Sequence[str]] | str] | None:
try:
return _BLOCK_GATE_FALLBACKS_ADAPTER.validate_python(raw)
except ValidationError:
return None
def _reachable_block_fallbacks(
llm_router: LitellmRouter, data: dict, route_type: str
) -> list[dict[str, list[str]] | str] | None:
llm_router: LitellmRouter, data: Mapping[str, object], route_type: str
) -> Sequence[Mapping[str, Sequence[str]] | str] | None:
"""Fallbacks the router would actually attempt for a blocked primary, or None when
none can run: eval routes bypass the router, disabled fallbacks skip the chain, and a
request-supplied list replaces the router-level one, matching what the router does."""
@ -84,30 +91,25 @@ def _reachable_block_fallbacks(
return None
if data.get("disable_fallbacks") is True:
return None
request_fallbacks: Final[object] = data.get("fallbacks")
raw_fallbacks: Final[object] = request_fallbacks if isinstance(request_fallbacks, list) else llm_router.fallbacks
try:
return _BLOCK_GATE_FALLBACKS_ADAPTER.validate_python(raw_fallbacks)
except ValidationError:
return None
if "fallbacks" in data:
return _validated_block_fallbacks(data.get("fallbacks"))
# Router.fallbacks is an untyped list attribute; the adapter validates it into a typed view.
return _validated_block_fallbacks(llm_router.fallbacks) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # untyped attr, validated by adapter
def _raise_if_model_fully_blocked(
llm_router: LitellmRouter,
model_name: Any,
team_id: str | None,
reachable_fallbacks: list[dict[str, list[str]] | str] | None,
reachable_fallbacks: Sequence[Mapping[str, Sequence[str]] | str] | None,
) -> None:
if not isinstance(model_name, str) or not model_name:
return
if not isinstance(llm_router, litellm.Router):
return
deployments: Final = llm_router.get_model_list(model_name=model_name, team_id=team_id) or []
if not llm_router._are_all_deployments_blocked(deployments):
return
if reachable_fallbacks is not None and llm_router._has_reachable_fallback(
if not llm_router._is_blocked_without_reachable_fallback(
model_name=model_name,
fallbacks=reachable_fallbacks,
reachable_fallbacks=reachable_fallbacks,
team_id=team_id,
):
return

View file

@ -9983,7 +9983,7 @@ class Router:
def _has_reachable_fallback(
self,
model_name: str,
fallbacks: list[dict[str, list[str]] | str],
fallbacks: Sequence[Mapping[str, Sequence[str]] | str],
team_id: str | None = None,
visited: frozenset[str] = frozenset(),
) -> bool:
@ -10005,6 +10005,26 @@ class Router:
for group in fallback_model_group
)
def _is_blocked_without_reachable_fallback(
self,
model_name: str,
reachable_fallbacks: Sequence[Mapping[str, Sequence[str]] | str] | None,
team_id: str | None,
) -> bool:
"""
True when every deployment of `model_name` is blocked and no reachable fallback
can serve the request. `reachable_fallbacks` is the already-resolved chain the
caller would attempt, or None when no fallback can run on this path.
"""
deployments: Final = self.get_model_list(model_name=model_name, team_id=team_id) or []
if not self._are_all_deployments_blocked(deployments):
return False
if reachable_fallbacks is not None and self._has_reachable_fallback(
model_name=model_name, fallbacks=reachable_fallbacks, team_id=team_id
):
return False
return True
async def async_get_fully_unhealthy_model_names(self) -> set[str]:
"""
Returns the set of model names where every backing deployment is currently

View file

@ -1,6 +1,6 @@
import hashlib
import json
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from enum import Enum
from typing import TYPE_CHECKING, Any, Final
@ -214,7 +214,7 @@ def _check_stripped_model_group(model_group: str, fallback_key: str) -> bool:
return False
def get_fallback_model_group(fallbacks: list[Any], model_group: str) -> tuple[list[str] | None, int | None]:
def get_fallback_model_group(fallbacks: Sequence[Any], model_group: str) -> tuple[list[str] | None, int | None]:
"""
Returns:
- fallback_model_group: List[str] of fallback model groups. example: ["gpt-4", "gpt-3.5-turbo"]

View file

@ -57,6 +57,7 @@ IGNORE_FUNCTIONS = [
"_filter_argument_value", # max depth set (DEFAULT_MAX_RECURSE_DEPTH); fails closed by blocking the tool call at the cap.
"_redact_scanned_content", # max depth set (DEFAULT_MAX_RECURSE_DEPTH); fails closed by returning "[REDACTED]" at the cap.
"_iter_fallback_targets", # max depth set (2 * ROUTER_MAX_FALLBACKS); fails closed by raising ValueError at the cap.
"_has_reachable_fallback", # bounded by the visited set: every model group is expanded at most once, so a cycle terminates.
"json_string_leaves", # max depth set (MAX_STRUCTURED_CONTENT_SCAN_DEPTH); fails closed by raising at the cap so nothing goes unscanned.
"with_json_string_leaves", # transitively bounded: only runs on a tree json_string_leaves already walked under the cap.
"json_unrewritable_labels", # max depth set (MAX_STRUCTURED_CONTENT_SCAN_DEPTH); returns the None sentinel at the cap so the caller blocks.