refactor(router): make the encrypted-content marker lookup a public entry point

The basedpyright gate failed on reportPrivateUsage: the pre-routing hook reached
into `EncryptedContentAffinityCheck._extract_model_id_from_input`. Reading a
request's marker is genuinely part of that callback's contract now, because the
Router needs the answer before a model group exists, so the method is public
rather than reached into. Only two call sites referenced the private name, both
inside its own module, and no test named it.

Also tightens the annotations the same gate counts: the tier mapping is read
through `Mapping[str, str | Sequence[str]]`, `input` is `Sequence[object]`
instead of a bare `list`, and the resolved request body is declared `object`,
which is all `extract_model_id_from_input` requires. Errors added over the gate
base drop from 12 to 8, with reportPrivateUsage, reportMissingTypeArgument and
reportUnknownParameterType back at base.
This commit is contained in:
aishwary-dongre 2026-09-09 12:12:21 +05:30
parent af1c9f85e8
commit b1bb36eaef
2 changed files with 12 additions and 6 deletions

View file

@ -13119,9 +13119,10 @@ class Router:
if not isinstance(strategy, ComplexityRouter):
return frozenset()
tiers: Final[Mapping[str, str | Sequence[str]]] = strategy.config.tiers
tier_models: Final = frozenset(
model_name
for tier_value in strategy.config.tiers.values()
for tier_value in tiers.values()
for model_name in ((tier_value,) if isinstance(tier_value, str) else tuple(tier_value))
)
default_model: Final = strategy.config.default_model
@ -13137,7 +13138,7 @@ class Router:
selected_strategy: "TaggedPreRoutingStrategy[PreRoutingStrategy]",
request_kwargs: Mapping[str, object],
messages: list[dict[str, Any]] | None = None,
input: str | list | None = None,
input: str | Sequence[object] | None = None,
) -> Optional["PreRoutingHookResponse"]:
"""Keep a follow-up carrying encrypted reasoning on the deployment that produced it.
@ -13164,8 +13165,8 @@ class Router:
if not affinity_registered:
return None
request_input: Final = input if input is not None else request_kwargs.get("input")
originating_model_id: Final = EncryptedContentAffinityCheck._extract_model_id_from_input(request_input)
request_input: Final[object] = input if input is not None else request_kwargs.get("input")
originating_model_id: Final = EncryptedContentAffinityCheck.extract_model_id_from_input(request_input)
if not originating_model_id:
return None

View file

@ -108,11 +108,16 @@ class EncryptedContentAffinityCheck(CustomLogger):
)
@staticmethod
def _extract_model_id_from_input(request_input: object) -> str | None:
def extract_model_id_from_input(request_input: object) -> str | None:
"""
Scan ``input`` items for litellm-encoded encrypted-content markers and
return the ``model_id`` embedded in the first one found.
Public because the Router's pre-routing hook needs the same answer before a
model group has been chosen: a complexity router would otherwise reclassify the
follow-up into a tier on the far side of the encryption boundary, and this
callback's deployment filtering only ever runs within one group.
Checks both:
1. Encoded item IDs (encitem_...) - for clients that send IDs
2. Wrapped encrypted_content (litellm_enc:...) - for clients like Codex that don't send IDs
@ -249,7 +254,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
request_kwargs["litellm_metadata"]["encrypted_content_affinity_enabled"] = True
request_input: Final = request_kwargs.get("input")
model_id: Final = self._extract_model_id_from_input(request_input)
model_id: Final = self.extract_model_id_from_input(request_input)
if not model_id:
return typed_healthy_deployments