scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 207

feat(proxy): give Codex accurate model entries for every model in /v1/models (#49825)

Browse source 
* feat(proxy): give Codex accurate model entries for every model in /v1/models

* fix(router): read a deployment's upstream model without a not-required key access

* fix(proxy): keep Codex tiers for unpriced deployments and match stock entries by declared provider

A deployment with no token price no longer votes a Codex service tier off,
an undescribed model keeps Codex's default reasoning summary parameter,
a Bedrock deployment naming its provider in custom_llm_provider gets Codex's
stock entry, and Bedrock provider namespaces follow the AWS model id format.

* fix(proxy): import cost tracking and provider helpers lazily in the Codex catalog

lite codex imports the catalog before the SDK is initialized, so importing
cost tracking, the Bedrock helpers or provider resolution at module level hit
a circular import and broke every lite command.

* fix(proxy): drop Codex's fast speed tier wherever its priority tier is dropped

Codex shows /fast for a leftover fast speed tier even when service_tiers
no longer offers priority, so an alias, an empty configured tier list or
a configured list without priority showed a /fast toggle that sends nothing.
One helper now sets service tiers, speed tiers and the default together.

* fix(proxy): give an older Codex a translated entry for models past its version

Codex's own entries name the oldest release that can read them in minimal_client_version. A release older than that can fail to decode the entry, such as a reasoning effort it has no name for, which fails the whole catalog, so it now gets the translated entry, whose levels are trimmed to that release
This commit is contained in:
ryan-crabbe-berri 2026-11-05 19:31:32 -08:00 • committed by GitHub
parent a1b87bcca8
commit 9a473fbbb0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 1324 additions and 200 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/llms/bedrock/common_utils.py
View file

@ -1538,6 +1538,19 @@ def get_bedrock_base_model(model: str) -> str:
return model
def bedrock_model_name_without_provider(model: str) -> str:
"""The model id without Bedrock's routing and region prefixes and without its provider namespace.
A Bedrock foundation model id is ``<provider>.<model>`` with a one-word provider (``anthropic.``,
``deepseek.``, ``moonshotai.``), so a base model whose first dot-separated part is one word drops it:
``bedrock/us.openai.gpt-6.1-sol`` and ``us-east-1/openai.gpt-6.1-sol`` are ``gpt-6.1-sol``,
``us.deepseek.r1-v1:0`` is ``r1-v1:0``. Any other base model, such as ``gpt-5.5`` or an imported model's
ARN id, is returned as is."""
base: Final = get_bedrock_base_model(model)
provider, separator, name = base.partition(".")
return name if separator and provider.isalnum() else base
def bedrock_converse_accepts_redacted_thinking(model: str, base_model: str | None) -> bool:
"""
Whether ``reasoningContent.redactedContent`` may be replayed to this Converse target.

187
litellm/proxy/common_utils/codex_bundled_models_0.159.3.json → litellm/proxy/common_utils/codex_bundled_models_0.166.0.json
View file

File diff suppressed because one or more lines are too long

295
litellm/proxy/common_utils/codex_model_catalog.py
View file

@ -3,37 +3,50 @@
Codex CLI discovers a provider's models with ``GET <model_catalog_url>?client_version=<its version>``
(the provider's ``model_catalog_url`` pointed at ``/v1/models``, with its ``api_key_model_discovery``
feature on) and decodes the body as its own ``ModelsResponse``, ``{"models": [ModelInfo, ...]}``,
never the OpenAI list shape. This module builds that body from the rows ``/v1/models`` already
lists: a model Codex knows keeps Codex's own entry for it, any other model gets the fallback entry
Codex itself uses for an unknown slug, and ``model_info.service_tiers`` becomes the entry's
never the OpenAI list shape. Codex then ignores its own bundled catalog, so this body is all it knows
about the provider's models. It is built from the rows ``/v1/models`` already lists: a model Codex
knows, by its id or by the upstream model its deployments call once the provider's own prefixes are
stripped, keeps Codex's own entry for it, minus any service tier some deployment behind it prices tokens
but not that tier for; any other model gets an entry translated from the capabilities its deployments' model info
declares (see ``codex_model_translation``). ``model_info.service_tiers`` becomes the entry's
``service_tiers``, which Codex turns into slash commands (``/ultrafast``) that send ``service_tier``
upstream. A ``model_info.max_input_tokens`` the operator sets replaces a known model's
``context_window``, up to the most Codex lets its own ``model_context_window`` raise that model to.
A fallback entry offers the model group's ``supported_reasoning_efforts`` as its reasoning levels,
A translated entry offers the model group's ``supported_reasoning_efforts`` as its reasoning levels,
trimmed to the ones the requesting Codex release can parse.
The stock entries are Codex 0.159.3's bundled catalog, vendored unchanged from
https://github.com/openai/codex/blob/rust-v0.159.3/codex-rs/models-manager/models.json (Apache-2.0)
as ``codex_bundled_models_0.159.3.json``, and ``codex_base_instructions.md`` is
The stock entries are Codex 0.166.0's bundled catalog, vendored unchanged from
https://github.com/openai/codex/blob/rust-v0.166.0/codex-rs/models-manager/models.json (Apache-2.0)
as ``codex_bundled_models_0.166.0.json``, and ``codex_base_instructions.md`` is
``codex-rs/models-manager/prompt.md`` at the same tag. Refresh both by downloading those two paths at
a newer tag.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from functools import cache
from itertools import accumulate, pairwise
from pathlib import Path
from types import MappingProxyType, NoneType
from typing import TYPE_CHECKING, Annotated, Final, Literal, TypeAlias
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal, TypeAlias
from packaging.version import Version
from pydantic import ConfigDict, Field, StringConstraints, TypeAdapter, ValidationError
from pydantic import ConfigDict, TypeAdapter, ValidationError
from litellm._logging import verbose_proxy_logger
from litellm.litellm_core_utils.reasoning_effort_strength import nearest_declared_reasoning_effort
from litellm.proxy.common_utils.codex_model_translation import (
TOKEN_PRICE_KEY,
CodexDeploymentCapabilities,
CodexFallbackModel,
CodexServiceTier,
TierId,
codex_fallback_entry,
combined_capabilities,
no_priced_deployment_lacks_tier,
tier_price_keys,
)
from litellm.types.llms.base_model import LiteLLMBaseModel
if TYPE_CHECKING:
@ -42,89 +55,24 @@ if TYPE_CHECKING:
from litellm.types.proxy.model_listing import ModelInfoResponse
CODEX_BASE_INSTRUCTIONS_PATH: Final = Path(__file__).with_name("codex_base_instructions.md")
CODEX_BUNDLED_MODELS_PATH: Final = Path(__file__).with_name("codex_bundled_models_0.159.3.json")
CODEX_BUNDLED_MODELS_PATH: Final = Path(__file__).with_name("codex_bundled_models_0.166.0.json")
CODEX_CATALOG_BYTE_LIMIT: Final = 1024 * 1024
CODEX_CHAT_MODES: Final = frozenset({"chat", "responses"})
_TierId = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
_BODY_PREFIX: Final = '{"models":['
_BODY_SUFFIX: Final = "]}"
_FIRST_CODEX_WITH_CUSTOM_EFFORTS: Final = Version("0.138.0")
_EFFORTS_EVERY_CODEX_PARSES: Final = frozenset({"none", "minimal", "low", "medium", "high", "xhigh"})
_PREFERRED_DEFAULT_EFFORT: Final = "medium"
_CODEX_UNSET_EFFORT: Final = "none"
class CodexServiceTier(LiteLLMBaseModel):
"""A `ModelServiceTier` as Codex reads it: the slash command is `name` lowercased, and toggling
it sends `id` as the request's `service_tier`."""
model_config = ConfigDict(frozen=True)
id: _TierId
name: str
description: str
_NO_UPDATE: Final[Mapping[str, object]] = MappingProxyType({})
class _ConfiguredTier(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
id: _TierId
id: TierId
name: str | None = None
description: str | None = None
_CONFIGURED_TIERS: Final = TypeAdapter(tuple[_TierId | _ConfiguredTier, ...])
class CodexFallbackReasoningLevel(LiteLLMBaseModel):
"""A `ReasoningEffortPreset` as Codex reads it: picking it sends `effort` as the request's reasoning effort."""
model_config = ConfigDict(frozen=True)
effort: str
description: str
class CodexTruncationPolicy(LiteLLMBaseModel):
mode: Literal["bytes"] = "bytes"
limit: int = 10_000
class CodexFallbackModel(LiteLLMBaseModel):
"""One `ModelInfo` entry of a Codex model catalog for a model Codex does not know.
Every field that some Codex release since `model_catalog_json` appeared
(0.105.0) deserializes without a default is spelled out here, so one catalog
parses on all of them; the values match the fallback metadata Codex uses for
a model slug it does not know, so picking such a proxy model behaves the
same as `codex -m` did.
"""
model_config = ConfigDict(frozen=True)
slug: str
display_name: str
description: NoneType = None
default_reasoning_level: str | None = Field(default=None, exclude_if=lambda level: level is None)
supported_reasoning_levels: tuple[CodexFallbackReasoningLevel, ...] = ()
shell_type: Literal["unified_exec"] = "unified_exec"
visibility: Literal["list"] = "list"
supported_in_api: Literal[True] = True
priority: int
service_tiers: tuple[CodexServiceTier, ...] = ()
availability_nux: NoneType = None
upgrade: NoneType = None
support_verbosity: Literal[False] = False
supports_reasoning_summaries: Literal[False] = False
supports_parallel_tool_calls: Literal[False] = False
default_verbosity: NoneType = None
apply_patch_tool_type: NoneType = None
truncation_policy: CodexTruncationPolicy = CodexTruncationPolicy()
experimental_supported_tools: tuple[()] = ()
context_window: int | None
base_instructions: str
supports_search_tool: bool = False
_CONFIGURED_TIERS: Final = TypeAdapter(tuple[TierId | _ConfiguredTier, ...])
class CodexStockUpgrade(LiteLLMBaseModel):
@ -156,11 +104,13 @@ class CodexStockModel(LiteLLMBaseModel):
supported_in_api: bool = True
upgrade: CodexStockUpgrade | None = None
service_tiers: tuple[CodexServiceTier, ...] = ()
additional_speed_tiers: tuple[str, ...] = ()
default_service_tier: str | None = None
supports_search_tool: bool = False
supported_reasoning_levels: tuple[CodexReasoningLevel, ...] = ()
context_window: int | None = None
max_context_window: int | None = None
minimal_client_version: str | tuple[int, ...] | None = None
class CodexStockCatalog(LiteLLMBaseModel):
@ -174,22 +124,26 @@ CodexAliasDefaults: TypeAlias = Literal["codex", "deployment"]
@dataclass(frozen=True, slots=True)
class CodexCatalogRow:
"""What the catalog needs to know about one listed model: its public id, the listing's mode and
input limit, the upstream model behind it, the reasoning effort its deployments pin, and the
input limit, the upstream model behind it and the provider its deployment declares, the reasoning
effort its deployments pin, and the
`model_info` values the operator set; `service_tiers` holds one raw `model_info.service_tiers`
value per deployment behind the id, None where a deployment sets none, and
`configured_max_input_tokens` one `model_info.max_input_tokens` per deployment the same way.
`supported_reasoning_efforts` is the model group's, None when unknown."""
`deployment_capabilities` holds what each deployment's model info, merged with the cost map,
declares. `supported_reasoning_efforts` is the model group's, None when unknown."""
id: str
mode: str | None = None
max_input_tokens: int | None = None
upstream_model: str | None = None
upstream_provider: str | None = None
reasoning_effort: str | None = None
display_name: str | None = None
service_tiers: tuple[object, ...] = ()
supported_reasoning_efforts: tuple[str, ...] | None = None
search_tool_support: tuple[bool | None, ...] = ()
configured_max_input_tokens: tuple[int | None, ...] = ()
deployment_capabilities: tuple[CodexDeploymentCapabilities, ...] = ()
@cache
@ -292,44 +246,87 @@ def _codex_can_drive(model_id: str, mode: str | None) -> bool:
return "*" not in model_id and (mode is None or mode in CODEX_CHAT_MODES)
def _parses_custom_efforts(client_version: str | None) -> bool:
@cache
def _provider_model_names() -> Mapping[str, Callable[[str], str]]:
"""Imported on first use for the same reason as `tier_price_keys`."""
from litellm.llms.bedrock.common_utils import bedrock_model_name_without_provider
return MappingProxyType(
{"bedrock": bedrock_model_name_without_provider, "bedrock_mantle": bedrock_model_name_without_provider}
)
def _provider_model_name(upstream_model: str, upstream_provider: str | None) -> str:
"""The upstream model without the prefixes its provider adds, the provider being the one the deployment
declares in `custom_llm_provider`, else the model's `provider/` prefix."""
from litellm.litellm_core_utils.get_llm_provider_logic import declared_provider
provider: Final = declared_provider(upstream_model, upstream_provider)
if provider is None:
return upstream_model
model: Final = upstream_model.removeprefix(f"{provider}/")
name_for: Final = _provider_model_names().get(provider)
return model if name_for is None else name_for(model)
def _stock_for(row: CodexCatalogRow, stock: Mapping[str, CodexStockModel]) -> CodexStockModel | None:
"""Codex's entry for the row's id, else for its upstream model as written, else for that model's name
once its provider's own prefixes are stripped (`bedrock/us.openai.gpt-6.1-sol` is `gpt-6.1-sol`)."""
names: Final = (
(row.id,)
if row.upstream_model is None
else (row.id, row.upstream_model, _provider_model_name(row.upstream_model, row.upstream_provider))
)
return next((stock[name] for name in names if name in stock), None)
def _readable_by(stock_model: CodexStockModel, client_version: str | None) -> bool:
"""Whether Codex's own server would hand this entry to the requesting release: one at or past the entry's
`minimal_client_version`. An older release can fail to decode it, and one entry it cannot decode (a reasoning
effort it has no name for) fails the whole catalog, so it gets a translated entry. No version keeps the entry."""
minimum: Final = stock_model.minimal_client_version
if minimum is None or client_version is None:
return True
try:
return client_version is not None and Version(client_version) >= _FIRST_CODEX_WITH_CUSTOM_EFFORTS
return Version(client_version) >= Version(minimum if isinstance(minimum, str) else ".".join(map(str, minimum)))
except ValueError:
return False
def codex_reasoning_levels(
efforts: Sequence[str], client_version: str | None
) -> tuple[tuple[CodexFallbackReasoningLevel, ...], str | None]:
"""The reasoning levels a fallback entry offers for a model group's efforts, and its default level.
def _tier_fields(stock_model: CodexStockModel, tiers: tuple[CodexServiceTier, ...]) -> Mapping[str, object]:
"""The tier fields of an entry offering `tiers`. Codex turns `/fast` on off a speed tier as much as off a
service tier, so a speed tier stays only while a service tier billed the same way is still offered, and the
default tier only while it is still offered."""
from litellm.litellm_core_utils.llm_cost_calc.utils import get_service_tier_cost_key
Codex before 0.138.0 fails the whole catalog on an effort it has no name for, so for it, and for
a version that does not parse, only the efforts every release knows are offered. The default is
set whenever a level is offered, because Codex reads a missing one as `none`: `none` itself when the
group accepts it, so a user who picks no level sends what Codex sent before, else medium, else the
nearest weaker declared level, so picking no level never costs more reasoning than Codex's usual medium,
else the weakest declared level, else the first one."""
parses_any: Final = _parses_custom_efforts(client_version)
offered: Final = tuple(effort for effort in efforts if parses_any or effort in _EFFORTS_EVERY_CODEX_PARSES)
if not offered:
return (), None
nearest: Final = nearest_declared_reasoning_effort(_PREFERRED_DEFAULT_EFFORT, offered, "down")
levels: Final = tuple(
CodexFallbackReasoningLevel(effort=effort, description=f"Sends reasoning_effort={effort} upstream")
for effort in offered
offered: Final = frozenset(tier.id for tier in tiers)
offered_prices: Final = frozenset(get_service_tier_cost_key(TOKEN_PRICE_KEY, tier) for tier in offered)
speed_tiers: Final = tuple(
tier
for tier in stock_model.additional_speed_tiers
if get_service_tier_cost_key(TOKEN_PRICE_KEY, tier) in offered_prices - {TOKEN_PRICE_KEY}
)
return MappingProxyType(
{
"service_tiers": tiers,
**({"additional_speed_tiers": speed_tiers} if stock_model.additional_speed_tiers else {}),
"default_service_tier": (
stock_model.default_service_tier if stock_model.default_service_tier in offered else None
),
}
)
if _CODEX_UNSET_EFFORT in offered:
return levels, _CODEX_UNSET_EFFORT
return levels, nearest if nearest in offered else offered[0]
def _stock_for(row: CodexCatalogRow, stock: Mapping[str, CodexStockModel]) -> CodexStockModel | None:
listed: Final = stock.get(row.id)
if listed is not None or row.upstream_model is None:
return listed
upstream: Final = stock.get(row.upstream_model)
return upstream if upstream is not None else stock.get(row.upstream_model.partition("/")[2])
def _priced_tiers(row: CodexCatalogRow, stock_model: CodexStockModel) -> Mapping[str, object]:
"""Codex's own tiers for the model without the ones some deployment behind the row prices tokens but not
that tier for, or nothing when none is dropped. Sending a tier the upstream does not sell fails the
request, and the cost map prices a tier for exactly the models that sell it, while a deployment it has no
token price for at all says nothing either way."""
deployments: Final = row.deployment_capabilities
tiers: Final = tuple(
tier for tier in stock_model.service_tiers if no_priced_deployment_lacks_tier(tier.id, deployments)
)
return _NO_UPDATE if tiers == stock_model.service_tiers else _tier_fields(stock_model, tiers)
def _alias_context_window(row: CodexCatalogRow, stock_model: CodexStockModel) -> Mapping[str, int]:
@ -369,8 +366,7 @@ def _deployment_defaults(row: CodexCatalogRow, stock_model: CodexStockModel) ->
none otherwise, and a window no larger than the alias's own limit."""
return MappingProxyType(
{
"service_tiers": (),
"default_service_tier": None,
**_tier_fields(stock_model, ()),
"default_verbosity": None,
"upgrade": None,
"availability_nux": None,
@ -389,24 +385,24 @@ def _entry(
alias_defaults: CodexAliasDefaults,
client_version: str | None,
) -> CodexCatalogEntry:
stock_model: Final = _stock_for(row, stock)
matched: Final = _stock_for(row, stock)
stock_model: Final = matched if matched is not None and _readable_by(matched, client_version) else None
stock_tiers: Final = MappingProxyType(
{tier.id: tier for tier in (stock_model.service_tiers if stock_model is not None else ())}
)
configured: Final = configured_service_tiers(row.service_tiers, row.id, stock_tiers)
search_tool: Final = offered_search_tool(row.search_tool_support)
if stock_model is None:
levels, default_level = codex_reasoning_levels(row.supported_reasoning_efforts or (), client_version)
return CodexFallbackModel(
return codex_fallback_entry(
slug=row.id,
display_name=row.display_name if row.display_name is not None else row.id,
default_reasoning_level=default_level,
supported_reasoning_levels=levels,
priority=index,
capabilities=combined_capabilities(row.deployment_capabilities, row.max_input_tokens),
instructions=instructions,
service_tiers=configured if configured is not None else (),
context_window=row.max_input_tokens,
base_instructions=instructions,
supports_search_tool=search_tool is True,
reasoning_efforts=row.supported_reasoning_efforts or (),
client_version=client_version,
)
upgrade: Final = (
stock_model.upgrade if stock_model.upgrade is not None and stock_model.upgrade.model in served else None
@ -425,17 +421,8 @@ def _entry(
**_configured_window(row, stock_model),
}
if configured is None:
return stock_model.model_copy(update=listing)
configured_ids: Final = frozenset(tier.id for tier in configured)
return stock_model.model_copy(
update={
**listing,
"service_tiers": configured,
"default_service_tier": (
stock_model.default_service_tier if stock_model.default_service_tier in configured_ids else None
),
}
)
return stock_model.model_copy(update={**_priced_tiers(row, stock_model), **listing})
return stock_model.model_copy(update={**listing, **_tier_fields(stock_model, configured)})
def _serialized(entry: CodexCatalogEntry) -> str:
@ -503,8 +490,10 @@ def codex_models_response_json(
`max_context_window`. A row listed under another name than the Codex model its `upstream_model` names (an alias) takes that model's
entry; `alias_defaults="deployment"` drops the defaults Codex picks for that model in favor of
what the alias's deployments pin, while `"codex"` keeps them, as Codex would use them for the
model's own name. `client_version` is the requesting Codex's version, which decides the reasoning
levels a fallback entry may offer; None counts as a version too old for custom levels.
model's own name. Codex's own service tiers stay on a known model's entry only when no deployment
behind the row prices tokens without pricing them, and only while the operator configures none. `client_version` is the
requesting Codex's version, which decides the reasoning levels a fallback entry may offer; None counts
as a version too old for custom levels.
"""
known: Final = bundled_codex_models() if stock is None else stock
prompt: Final = codex_base_instructions() if instructions is None else instructions
@ -531,6 +520,35 @@ def _group_reasoning_efforts(
return group.supported_reasoning_efforts if group is not None else None
def deployment_capabilities(model_info: Mapping[str, object] | None, model_id: str) -> CodexDeploymentCapabilities:
"""One deployment's model info, merged with the cost map, as the capabilities a Codex entry describes.
Model info that does not validate declares nothing, logged, so one bad value never fails the listing."""
if model_info is None:
return CodexDeploymentCapabilities()
prices_tokens: Final = model_info.get(TOKEN_PRICE_KEY) is not None
priced: Final = frozenset(key for key in tier_price_keys() if model_info.get(key) is not None)
try:
return CodexDeploymentCapabilities.model_validate(
{**model_info, "prices_tokens": prices_tokens, "priced_service_tier_keys": priced}
)
except ValidationError as e:
verbose_proxy_logger.warning(
"model info of a deployment of %s declares no capability to Codex: %s", model_id, e.errors()[0]["msg"]
)
return CodexDeploymentCapabilities(prices_tokens=prices_tokens, priced_service_tier_keys=priced)
def _deployment_capabilities(
lookup_id: str, registry: RouterRegistry | None, team_id: str | None
) -> tuple[CodexDeploymentCapabilities, ...]:
if registry is None:
return ()
return tuple(
deployment_capabilities(model_info, lookup_id)
for model_info in registry.group(lookup_id, field="deployment_model_info", team_id=team_id)
)
def _catalog_row(
row: ModelInfoResponse, lookup_id: str, label_id: str, llm_router: Router | None, team_id: str | None
) -> CodexCatalogRow:
@ -540,6 +558,9 @@ def _catalog_row(
mode=row.get("mode"),
max_input_tokens=row.get("max_input_tokens"),
upstream_model=registry.resolve(lookup_id, kind="upstream", team_id=team_id) if registry is not None else None,
upstream_provider=(
registry.resolve(lookup_id, kind="upstream_provider", team_id=team_id) if registry is not None else None
),
display_name=registry.group(label_id, field="display_name") if registry is not None else None,
service_tiers=registry.group(lookup_id, field="service_tiers", team_id=team_id) if registry is not None else (),
supported_reasoning_efforts=_group_reasoning_efforts(row, lookup_id, registry, team_id),
@ -549,6 +570,7 @@ def _catalog_row(
configured_max_input_tokens=(
registry.group(lookup_id, field="max_input_tokens", team_id=team_id) if registry is not None else ()
),
deployment_capabilities=_deployment_capabilities(lookup_id, registry, team_id),
)
@ -562,8 +584,9 @@ def codex_catalog_rows(
"""`rows` joined with the router's configured metadata, looked up by each entry's internal id so
team-scoped rows resolve the way the Anthropic listing's display names do, a `model_group_alias`
reading its target's deployments under its own name; `team_id` is the requesting key's team, so the
upstream model that picks Codex's stock entry, the tiers, and the configured input limits are read
only off the deployments its requests can route to, and the reasoning efforts are the ones
upstream model that picks Codex's stock entry, the tiers, the configured input limits and each
deployment's capabilities are read only off the deployments its requests can route to, and the
reasoning efforts are the ones
`/model_group/info` reports for the entry. Display names are read under `label_entries`' ids when given."""
lookup_ids: Final = MappingProxyType(dict(entries))
label_ids: Final = lookup_ids if label_entries is None else MappingProxyType(dict(label_entries))

250
litellm/proxy/common_utils/codex_model_translation.py Normal file
View file

@ -0,0 +1,250 @@
"""Codex's `ModelInfo` entry for a model Codex does not know, translated from what LiteLLM knows about the
deployments behind it.
Field names, types and enum values follow `codex-rs/protocol/src/openai_models.rs` at rust-v0.166.0, and every
field a release back to 0.140.0 decodes without a default is spelled out, so one body parses on all of them.
Codex rejects the whole catalog when one entry fails to decode, so only enum values every one of those
releases knows are sent.
"""
from __future__ import annotations
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from functools import cache
from types import NoneType
from typing import Annotated, Final, Literal, TypeAlias
from packaging.version import Version
from pydantic import ConfigDict, Field, StringConstraints
from litellm.litellm_core_utils.reasoning_effort_strength import nearest_declared_reasoning_effort
from litellm.types.llms.base_model import LiteLLMBaseModel
TierId: TypeAlias = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
CodexInputModality: TypeAlias = Literal["text", "image"]
_TEXT_ONLY: Final[tuple[CodexInputModality, ...]] = ("text",)
_TEXT_AND_IMAGE: Final[tuple[CodexInputModality, ...]] = ("text", "image")
TOKEN_PRICE_KEY: Final = "input_cost_per_token"
_FIRST_CODEX_WITH_CUSTOM_EFFORTS: Final = Version("0.138.0")
_EFFORTS_EVERY_CODEX_PARSES: Final = frozenset({"none", "minimal", "low", "medium", "high", "xhigh"})
_PREFERRED_DEFAULT_EFFORT: Final = "medium"
_CODEX_UNSET_EFFORT: Final = "none"
def _unset(value: object) -> bool:
return value is None
class CodexServiceTier(LiteLLMBaseModel):
"""A `ModelServiceTier` as Codex reads it: the slash command is `name` lowercased, and toggling
it sends `id` as the request's `service_tier`."""
model_config = ConfigDict(frozen=True)
id: TierId
name: str
description: str
class CodexFallbackReasoningLevel(LiteLLMBaseModel):
"""A `ReasoningEffortPreset` as Codex reads it: picking it sends `effort` as the request's reasoning effort."""
model_config = ConfigDict(frozen=True)
effort: str
description: str
class CodexTruncationPolicy(LiteLLMBaseModel):
"""How Codex truncates a long tool output before sending it back, the policy its own entries use."""
model_config = ConfigDict(frozen=True)
mode: Literal["tokens"] = "tokens"
limit: int = 10_000
class CodexFallbackModel(LiteLLMBaseModel):
"""One `ModelInfo` entry of a Codex model catalog for a model Codex does not know.
`supports_reasoning_summaries` and `supports_parallel_tool_calls` are read only by Codex before 0.150.0,
which requires them, and `base_instructions` is the prompt field every release still decodes.
"""
model_config = ConfigDict(frozen=True)
slug: str
display_name: str
description: NoneType = None
default_reasoning_level: str | None = Field(default=None, exclude_if=_unset)
supported_reasoning_levels: tuple[CodexFallbackReasoningLevel, ...] = ()
shell_type: Literal["unified_exec"] = "unified_exec"
visibility: Literal["list"] = "list"
supported_in_api: Literal[True] = True
priority: int
service_tiers: tuple[CodexServiceTier, ...] = ()
availability_nux: NoneType = None
upgrade: NoneType = None
support_verbosity: Literal[False] = False
supports_reasoning_summary_parameter: bool | None = Field(default=None, exclude_if=_unset)
supports_reasoning_summaries: bool = False
supports_parallel_tool_calls: bool = False
default_verbosity: NoneType = None
apply_patch_tool_type: Literal["freeform"] | None = None
truncation_policy: CodexTruncationPolicy = CodexTruncationPolicy()
experimental_supported_tools: tuple[()] = ()
input_modalities: tuple[CodexInputModality, ...] = _TEXT_AND_IMAGE
context_window: int | None
base_instructions: str
supports_search_tool: bool = False
use_responses_lite: Literal[False] = False
class CodexDeploymentCapabilities(LiteLLMBaseModel):
"""What one deployment's model info says about the capabilities a Codex entry describes, None where
it says nothing, whether it has a token price at all, and the input price keys of the service tiers it
has a price for."""
model_config = ConfigDict(frozen=True, extra="ignore")
max_input_tokens: int | None = None
supports_vision: bool | None = None
supports_function_calling: bool | None = None
supports_parallel_function_calling: bool | None = None
supports_reasoning: bool | None = None
prices_tokens: bool = False
priced_service_tier_keys: frozenset[str] = frozenset()
@cache
def tier_price_keys() -> tuple[str, ...]:
"""The input price key of every service tier cost tracking prices. Cost tracking is imported on first use
because `lite codex` imports this module before the SDK is initialized, and cost tracking imports the rest
of the SDK."""
from litellm.litellm_core_utils.llm_cost_calc.utils import SERVICE_TIER_COST_KEY_SUFFIXES
return tuple(f"{TOKEN_PRICE_KEY}{suffix}" for suffix in SERVICE_TIER_COST_KEY_SUFFIXES)
def no_priced_deployment_lacks_tier(tier_id: str, deployments: Sequence[CodexDeploymentCapabilities]) -> bool:
"""Whether every deployment that has a token price also has one for `tier_id`, the way cost tracking
prices a request sent at that tier. A deployment with no token price at all (a subscription upstream,
a deployment name the cost map does not know) says nothing about tiers, so it does not vote, and a tier
cost tracking has no price key for is lacked by every deployment that prices tokens."""
from litellm.litellm_core_utils.llm_cost_calc.utils import get_service_tier_cost_key
price_key: Final = get_service_tier_cost_key(TOKEN_PRICE_KEY, tier_id)
return all(
price_key in deployment.priced_service_tier_keys for deployment in deployments if deployment.prices_tokens
)
@dataclass(frozen=True, slots=True)
class CodexCapabilities:
"""What a request to a listed name can count on whichever deployment serves it: a capability is False
when one deployment lacks it, True when one has it and none lacks it, None when none says, and the
context window is the smallest one any deployment declares."""
context_window: int | None = None
supports_vision: bool | None = None
supports_function_calling: bool | None = None
supports_parallel_function_calling: bool | None = None
supports_reasoning: bool | None = None
def _combined(flags: Iterable[bool | None]) -> bool | None:
said: Final = frozenset(flag for flag in flags if flag is not None)
return None if not said else False not in said
def combined_capabilities(
deployments: Sequence[CodexDeploymentCapabilities], listed_context_window: int | None = None
) -> CodexCapabilities:
"""The capabilities every one of `deployments` offers, the window falling back to
`listed_context_window` when no deployment declares one."""
windows: Final = tuple(
deployment.max_input_tokens
for deployment in deployments
if deployment.max_input_tokens is not None and deployment.max_input_tokens > 0
)
return CodexCapabilities(
context_window=min(windows) if windows else listed_context_window,
supports_vision=_combined(deployment.supports_vision for deployment in deployments),
supports_function_calling=_combined(deployment.supports_function_calling for deployment in deployments),
supports_parallel_function_calling=_combined(
deployment.supports_parallel_function_calling for deployment in deployments
),
supports_reasoning=_combined(deployment.supports_reasoning for deployment in deployments),
)
def _parses_custom_efforts(client_version: str | None) -> bool:
try:
return client_version is not None and Version(client_version) >= _FIRST_CODEX_WITH_CUSTOM_EFFORTS
except ValueError:
return False
def codex_reasoning_levels(
efforts: Sequence[str], client_version: str | None
) -> tuple[tuple[CodexFallbackReasoningLevel, ...], str | None]:
"""The reasoning levels a fallback entry offers for a model group's efforts, and its default level.
Codex before 0.138.0 fails the whole catalog on an effort it has no name for, so for it, and for
a version that does not parse, only the efforts every release knows are offered. The default is
set whenever a level is offered, because Codex reads a missing one as `none`: `none` itself when the
group accepts it, so a user who picks no level sends what Codex sent before, else medium, else the
nearest weaker declared level, so picking no level never costs more reasoning than Codex's usual medium,
else the weakest declared level, else the first one."""
parses_any: Final = _parses_custom_efforts(client_version)
offered: Final = tuple(effort for effort in efforts if parses_any or effort in _EFFORTS_EVERY_CODEX_PARSES)
if not offered:
return (), None
nearest: Final = nearest_declared_reasoning_effort(_PREFERRED_DEFAULT_EFFORT, offered, "down")
levels: Final = tuple(
CodexFallbackReasoningLevel(effort=effort, description=f"Sends reasoning_effort={effort} upstream")
for effort in offered
)
if _CODEX_UNSET_EFFORT in offered:
return levels, _CODEX_UNSET_EFFORT
return levels, nearest if nearest in offered else offered[0]
def codex_fallback_entry(
*,
slug: str,
display_name: str,
priority: int,
capabilities: CodexCapabilities,
instructions: str,
service_tiers: tuple[CodexServiceTier, ...] = (),
supports_search_tool: bool = False,
reasoning_efforts: Sequence[str] = (),
client_version: str | None = None,
) -> CodexFallbackModel:
"""Codex's entry for a model it does not know, from the capabilities its deployments share.
A capability no deployment declares keeps Codex's own default for an unknown model, except the freeform
`apply_patch` tool, which goes out only to a model that calls tools: `supports_reasoning_summary_parameter`
is left out, which Codex reads as true, and the fields only Codex before 0.150.0 reads keep the false
it requires.
"""
levels, default_level = codex_reasoning_levels(reasoning_efforts, client_version)
return CodexFallbackModel(
slug=slug,
display_name=display_name,
default_reasoning_level=default_level,
supported_reasoning_levels=levels,
priority=priority,
service_tiers=service_tiers,
supports_reasoning_summary_parameter=capabilities.supports_reasoning,
supports_reasoning_summaries=capabilities.supports_reasoning is True,
supports_parallel_tool_calls=capabilities.supports_parallel_function_calling is True,
apply_patch_tool_type="freeform" if capabilities.supports_function_calling is True else None,
input_modalities=_TEXT_ONLY if capabilities.supports_vision is False else _TEXT_AND_IMAGE,
context_window=capabilities.context_window,
base_instructions=instructions,
supports_search_tool=supports_search_tool,
)

38
litellm/router_api.py
View file

@ -145,6 +145,7 @@ ResolveKind: TypeAlias = Literal[
"team_model",
"recognized",
"upstream",
"upstream_provider",
"group_of_id",
"wildcard_access_group",
"team_model_alternatives",
@ -158,6 +159,7 @@ GroupField: TypeAlias = Literal[
"mode",
"service_tiers",
"max_input_tokens",
"deployment_model_info",
"token_limits",
"listing_info",
]
@ -757,7 +759,7 @@ class RouterRegistry:
self,
name: str,
*,
kind: Literal["alias", "upstream"],
kind: Literal["alias", "upstream", "upstream_provider"],
team_id: str | None = None,
custom_llm_provider: str | None = None,
served_model: str | None = None,
@ -837,6 +839,8 @@ class RouterRegistry:
- ``team_model``: a team's public name to its group (was ``map_team_model``; ``team_id`` required)
- ``recognized``: whether the router knows the name (was ``is_recognized_model``)
- ``upstream``: the upstream model a name routes to (was ``get_routable_upstream_model``)
- ``upstream_provider``: the provider the deployment ``upstream`` reads declares, by its
``custom_llm_provider`` or its model's ``provider/`` prefix
- ``group_of_id``: the group of a deployment id (was ``get_model_group``)
- ``wildcard_access_group``: whether an access group grants a wildcard route
(was ``is_model_access_group_for_wildcard_route``)
@ -866,10 +870,8 @@ class RouterRegistry:
return name if served_model is None else r._core_resolve_served_model_name(name, served_model)
if kind == "recognized":
return r._core_is_recognized_model(name)
if kind == "upstream":
return r._core_get_routable_upstream_model(name, team_id=team_id)
if kind == "search_tool_support":
return r._core_get_search_tool_support(name, team_id=team_id)
if kind in ("upstream", "upstream_provider", "search_tool_support"):
return self._resolve_routable(name, kind, team_id)
if kind == "group_of_id":
return r._core_get_model_group(id=name)
if kind == "wildcard_access_group":
@ -878,6 +880,19 @@ class RouterRegistry:
return r._core_team_model_has_alternatives(name)
raise ValueError(f"unknown resolve kind {kind!r}")
def _resolve_routable(
self,
name: str,
kind: Literal["upstream", "upstream_provider", "search_tool_support"],
team_id: str | None,
) -> str | tuple[bool | None, ...] | None:
r: Final = self._router
if kind == "upstream":
return r._core_get_routable_upstream_model(name, team_id=team_id)
if kind == "upstream_provider":
return r._core_get_routable_upstream_provider(name, team_id=team_id)
return r._core_get_search_tool_support(name, team_id=team_id)
def aliases(
self,
) -> "dict[str, str | RouterModelGroupAliasItem]": # mutable-ok: the model_group_alias attribute, the live dict
@ -939,6 +954,16 @@ class RouterRegistry:
team_id: str | None = None,
) -> tuple[int | None, ...]: ...
@overload
def group(
self,
name: str,
*,
field: Literal["deployment_model_info"],
include_hidden: bool = False,
team_id: str | None = None,
) -> "tuple[ModelMapInfo | None, ...]": ...
@overload
def group(
self,
@ -977,6 +1002,7 @@ class RouterRegistry:
``get_configured_token_limits``, ``get_model_listing_info``)
- ``service_tiers``: (was ``get_configured_service_tiers``)
- ``max_input_tokens``: each routable deployment's own ``model_info.max_input_tokens``
- ``deployment_model_info``: each routable deployment's model info, merged with the pricing map
"""
r: Final = self._router
if field == "info":
@ -991,6 +1017,8 @@ class RouterRegistry:
return r._core_get_configured_service_tiers(name, team_id=team_id)
if field == "max_input_tokens":
return r._core_get_configured_max_input_tokens(name, team_id=team_id)
if field == "deployment_model_info":
return r._core_get_routable_deployment_model_info(name, team_id=team_id)
if field == "token_limits":
return r._core_get_configured_token_limits(name)
if field == "listing_info":

36
litellm/router_rust/surface/registry.py
View file

@ -65,6 +65,7 @@ from litellm.litellm_core_utils.core_helpers import as_str_mapping, without
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
from litellm.litellm_core_utils.get_llm_provider_logic import (
declared_authenticating_provider,
declared_provider,
is_registered_custom_provider,
resolve_model_and_provider_for_metadata,
)
@ -141,6 +142,7 @@ if TYPE_CHECKING:
from litellm.router_rust.steps.prerouting import PreRoutingStrategies
from litellm.router_strategy.budget_limiter import RouterBudgetLimiting
from litellm.rust_bridge._native import RouterCoreHandle
from litellm.types.utils import ModelInfo as ModelMapInfo
# The PTU share teams and the team ceilings read so far.
_PTUMemo: TypeAlias = tuple[
@ -679,7 +681,7 @@ class HostRegistry(HostModelInfo):
self._cached_get_model_group_info: Callable[[str], ModelGroupInfo | None] = lru_cache( # pyright: ignore[reportIncompatibleMethodOverride] # the per-router cache shadows the uncached method, as the core keeps it
maxsize=DEFAULT_MAX_LRU_CACHE_SIZE
)(self._core_get_model_group_info)
self._core_cached_deployment_model_info: Callable[[str, str], object] = lru_cache(
self._core_cached_deployment_model_info: Callable[[str, str], ModelMapInfo | None] = lru_cache(
maxsize=DEFAULT_MAX_LRU_CACHE_SIZE
)(self._core_get_deployment_model_info)
self.cached_deployment_tracks_usage: Callable[[str, str], bool] = lru_cache(maxsize=DEFAULT_MAX_LRU_CACHE_SIZE)(
@ -2079,6 +2081,38 @@ class HostRegistry(HostModelInfo):
model: Final = _params_of(deployment).get("model")
return None if model is None else str(model)
def _core_get_routable_upstream_provider(self, model_name: str, team_id: str | None = None) -> str | None:
"""The provider the deployment ``_core_get_routable_upstream_model`` reads declares: its
``litellm_params.custom_llm_provider``, else its model's ``provider/`` prefix. None when it declares neither."""
deployment: Final = next(iter(self._routable_deployments(model_name, team_id)), None)
if deployment is None:
return None
params: Final = _params_of(deployment)
model: Final = params.get("model")
provider: Final = params.get("custom_llm_provider")
return declared_provider(
model if isinstance(model, str) else None, provider if isinstance(provider, str) else None
)
def _core_get_routable_deployment_model_info(
self, model_name: str, team_id: str | None = None
) -> "tuple[ModelMapInfo | None, ...]":
"""Each deployment a request from ``team_id`` to ``model_name`` can route to, read one at a time the way
cost tracking reads it: its upstream model's cost map entry merged with its own ``model_info``, None where
neither is known. Nothing is merged across deployments, so a caller can tell a deployment that lacks a
capability from one that says nothing."""
return tuple(
self._routable_deployment_model_info(deployment)
for deployment in self._routable_deployments(model_name, team_id)
)
def _routable_deployment_model_info(self, deployment: Mapping[str, object]) -> "ModelMapInfo | None":
deployment_id: Final = _model_info_of(deployment).get("id")
model: Final = _params_of(deployment).get("model")
if not isinstance(deployment_id, str) or not isinstance(model, str):
return None
return self._core_cached_deployment_model_info(deployment_id, model)
def _core_map_team_model(
self, team_model_name: str | None, team_id: str, custom_llm_provider: str | None = None
) -> str | None:

2
packaging/litellm-core/pyproject.toml
View file

@ -61,7 +61,7 @@ include = [
"litellm/router_strategy/complexity_router/fuse_presets.json",
"litellm/proxy/model_insights_tasks.json",
"litellm/proxy/common_utils/codex_base_instructions.md",
"litellm/proxy/common_utils/codex_bundled_models_0.159.3.json",
"litellm/proxy/common_utils/codex_bundled_models_0.166.0.json",
]
exclude = [
"litellm/proxy/_experimental/out",

2
pyproject.toml
View file

@ -342,7 +342,7 @@ include = [
"litellm/proxy/model_insights_tasks.json",
"litellm/proxy/public_endpoints/whats_new.json",
"litellm/proxy/common_utils/codex_base_instructions.md",
"litellm/proxy/common_utils/codex_bundled_models_0.159.3.json",
"litellm/proxy/common_utils/codex_bundled_models_0.166.0.json",
{ path = "model_prices_and_context_window.json", format = "sdist" },
]
exclude = [

2
tests/integration/compatibility/test_codex_model_catalog.py
View file

@ -43,7 +43,7 @@ RELOAD_MARGIN_SECONDS: Final = 5.0
PROXY_WORKERS: Final = int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1"))
WORKER_SYNC_SECONDS: Final = 0.0 if PROXY_WORKERS == 1 else PROXY_CONFIG_RELOAD_INTERVAL_SECONDS + RELOAD_MARGIN_SECONDS
ROOT: Final = Path(os.environ.get("INTEGRATION_PROXY_ROOT") or Path(__file__).resolve().parents[3])
BUNDLED_STOCK_PATH: Final = ROOT / "litellm" / "proxy" / "common_utils" / "codex_bundled_models_0.159.3.json"
BUNDLED_STOCK_PATH: Final = ROOT / "litellm" / "proxy" / "common_utils" / "codex_bundled_models_0.166.0.json"
BASE_INSTRUCTIONS_PATH: Final = ROOT / "litellm" / "proxy" / "common_utils" / "codex_base_instructions.md"
FRESH_CONNECTION: Final = {"Connection": "close"}
ANTHROPIC_HEADERS: Final = {"anthropic-version": "2023-06-01"}

2
tests/integration/configuration/test_codex_catalog_boot.py
View file

@ -130,7 +130,7 @@ def test_yaml_tiers_and_alias_in_owned_proxy(gateway: Gateway, tmp_path: Path) -
ROOT: Final = Path(os.environ.get("INTEGRATION_PROXY_ROOT") or Path(__file__).resolve().parents[3])
BUNDLED_STOCK_PATH: Final = ROOT / "litellm" / "proxy" / "common_utils" / "codex_bundled_models_0.159.3.json"
BUNDLED_STOCK_PATH: Final = ROOT / "litellm" / "proxy" / "common_utils" / "codex_bundled_models_0.166.0.json"
STOCK_UPSTREAM: Final = "gpt-5.5"

21
tests/unit/llms/bedrock/test_bedrock_common_utils.py
View file

@ -11,6 +11,7 @@ from litellm.llms.bedrock.common_utils import(
NativeOutputFallback,
bedrock_native_output_is_opt_in,
bedrock_strict_tools_fallback,
bedrock_model_name_without_provider,
deployment_model_info,
extract_model_name_from_bedrock_arn,
for_bedrock_json_tool_retry,
@ -305,6 +306,26 @@ def test_legacy_mantle_route_prefix_stripped_for_cost_lookup():
assert BedrockModelInfo.get_bedrock_route("bedrock/mantle/anthropic.claude-sonnet-5") == "mantle"
@pytest.mark.parametrize(
("model", "expected"),
[
("bedrock/us.openai.gpt-6.1-sol", "gpt-6.1-sol"),
("global.openai.gpt-6.1-sol", "gpt-6.1-sol"),
("bedrock/converse/eu.anthropic.claude-sonnet-5-5", "claude-sonnet-5-5"),
("us-east-1/openai.gpt-6.1-sol", "gpt-6.1-sol"),
("openai.gpt-5.5", "gpt-5.5"),
("arn:aws:bedrock:us-east-1:123456789012:inference-profile/us.meta.llama4-scout-17b", "llama4-scout-17b"),
("us.deepseek.r1-v1:0", "r1-v1:0"),
("moonshotai.kimi-k2-thinking", "kimi-k2-thinking"),
("bedrock/qwen.qwen3-32b-v1:0", "qwen3-32b-v1:0"),
("gpt-5.5", "gpt-5.5"),
("arn:aws:bedrock:us-east-1:123456789012:imported-model/abc123", "abc123"),
],
)
def test_bedrock_model_name_without_provider_strips_routing_region_and_provider_prefixes_only(model, expected):
assert bedrock_model_name_without_provider(model) == expected
def test_output_config_effort_normalization_uses_model_info_ceiling(monkeypatch):
import litellm.llms.bedrock.common_utils as mod

3
tests/unit/proxy/client/cli/test_agents.py
View file

@ -1037,10 +1037,11 @@ class TestCodexModelSync:
assert entry["supported_in_api"] is True
assert entry["shell_type"] == "unified_exec"
assert entry["supported_reasoning_levels"] == []
assert entry["truncation_policy"] == {"mode": "bytes", "limit": 10000}
assert entry["truncation_policy"] == {"mode": "tokens", "limit": 10000}
assert entry["experimental_supported_tools"] == []
assert entry["support_verbosity"] is False
assert entry["supports_reasoning_summaries"] is False
assert "supports_reasoning_summary_parameter" not in entry
assert entry["supports_parallel_tool_calls"] is False
for nullable in ("description", "availability_nux", "upgrade", "default_verbosity", "apply_patch_tool_type"):
assert nullable in entry and entry[nullable] is None

422
tests/unit/proxy/common_utils/test_codex_model_catalog.py
View file

@ -1,5 +1,6 @@
import json
from collections.abc import Mapping
from dataclasses import replace
from types import MappingProxyType
from typing import Final, Literal
@ -7,10 +8,10 @@ import pytest
import litellm
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.llm_cost_calc.utils import get_service_tier_cost_key
from litellm.proxy.common_utils.codex_model_catalog import (
CODEX_CATALOG_BYTE_LIMIT,
CodexCatalogRow,
CodexServiceTier,
CodexStockModel,
CodexStockUpgrade,
bundled_codex_models,
@ -18,13 +19,17 @@ from litellm.proxy.common_utils.codex_model_catalog import (
codex_model_list_body,
codex_models_response_json,
configured_service_tiers,
deployment_capabilities,
)
from litellm.proxy.common_utils.codex_model_translation import CodexDeploymentCapabilities, CodexServiceTier
_PROMPT = "base prompt"
_FAST = {"id": "priority", "name": "Fast", "description": "2x speed, increased usage"}
def _stock(slug, *, visibility="list", supported_in_api=True, upgrade=None, service_tiers=(_FAST,), default_tier=None):
def _stock(
slug, *, visibility="list", supported_in_api=True, upgrade=None, service_tiers=(_FAST,), default_tier=None
):
return CodexStockModel(
slug=slug,
display_name=slug.upper(),
@ -33,6 +38,7 @@ def _stock(slug, *, visibility="list", supported_in_api=True, upgrade=None, serv
supported_in_api=supported_in_api,
upgrade=CodexStockUpgrade(model=upgrade) if upgrade else None,
service_tiers=tuple(CodexServiceTier(**tier) for tier in service_tiers),
additional_speed_tiers=("fast",) if service_tiers else (),
default_service_tier=default_tier,
model_messages={"instructions_template": f"{slug} prompt"},
)
@ -84,29 +90,10 @@ def test_body_is_exactly_codex_models_response():
assert [entry["slug"] for entry in body["models"]] == ["gpt-6-astra"]
def test_unknown_model_entry_carries_every_field_codex_deserializes_without_a_default():
"""Codex 0.159.3 `codex-rs/protocol/src/openai_models.rs` ModelInfo, the fields with no
`#[serde(default)]`, read on 2026-10-01; `base_instructions` because its `ModelsResponse`
decoder rejects an entry with neither it nor `model_messages.instructions_template`."""
def test_unknown_model_entry_takes_its_slug_window_and_prompt_from_the_row():
"""The fields every Codex release decodes are pinned in test_codex_model_translation.py."""
(entry,) = _models(_row("gpt-6-astra"))
assert entry.keys() >= {
"slug",
"display_name",
"description",
"supported_reasoning_levels",
"shell_type",
"visibility",
"supported_in_api",
"priority",
"availability_nux",
"upgrade",
"support_verbosity",
"default_verbosity",
"apply_patch_tool_type",
"truncation_policy",
"experimental_supported_tools",
}
assert (entry["slug"], entry["display_name"], entry["visibility"], entry["supported_in_api"]) == (
"gpt-6-astra",
"gpt-6-astra",
@ -116,25 +103,14 @@ def test_unknown_model_entry_carries_every_field_codex_deserializes_without_a_de
assert (entry["context_window"], entry["base_instructions"], entry["service_tiers"]) == (272000, _PROMPT, [])
_NO_REASONING_BODY = (
'{"models":[{"slug":"gpt-6-astra","display_name":"gpt-6-astra","description":null,'
'"supported_reasoning_levels":[],"shell_type":"unified_exec","visibility":"list","supported_in_api":true,'
'"priority":0,"service_tiers":[],"availability_nux":null,"upgrade":null,"support_verbosity":false,'
'"supports_reasoning_summaries":false,"supports_parallel_tool_calls":false,"default_verbosity":null,'
'"apply_patch_tool_type":null,"truncation_policy":{"mode":"bytes","limit":10000},'
'"experimental_supported_tools":[],"context_window":272000,"base_instructions":"base prompt",'
'"supports_search_tool":false}]}'
)
@pytest.mark.parametrize("efforts", [None, ()], ids=["unknown", "non-reasoning"])
@pytest.mark.parametrize("client_version", ["0.165.0", "0.120.0", None])
def test_unknown_or_no_reasoning_efforts_keep_the_fallback_entry_byte_for_byte(efforts, client_version):
"""The body an unknown model got before reasoning efforts reached the catalog."""
assert (
_body(_row("gpt-6-astra", supported_reasoning_efforts=efforts), client_version=client_version).json
== _NO_REASONING_BODY
)
def test_unknown_or_no_reasoning_efforts_offer_no_reasoning_level(efforts, client_version):
"""Codex reads a missing default level as `none`, so an entry offering no level sends no effort."""
(entry,) = _models(_row("mystery", supported_reasoning_efforts=efforts), client_version=client_version)
assert entry["supported_reasoning_levels"] == []
assert "default_reasoning_level" not in entry
_ABSENT = object()
@ -204,6 +180,51 @@ def test_a_model_codex_knows_keeps_its_own_reasoning_levels(levels):
assert (entry["default_reasoning_level"], entry["supported_reasoning_levels"]) == ("medium", levels)
@pytest.mark.parametrize("minimum", ["0.153.0", [0, 153, 0]], ids=["string", "parts"])
@pytest.mark.parametrize(
("client_version", "display_name", "levels"),
[
("0.153.0", "GPT-6.1 Terra", ["low", "max"]),
("0.166.0", "GPT-6.1 Terra", ["low", "max"]),
(None, "GPT-6.1 Terra", ["low", "max"]),
("0.152.9", "gpt-6.1-terra", ["low", "max"]),
("0.137.0", "gpt-6.1-terra", ["low"]),
("not-a-version", "gpt-6.1-terra", ["low"]),
],
)
def test_a_codex_older_than_an_entrys_minimal_version_gets_an_entry_it_can_decode(
minimum, client_version, display_name, levels
):
"""Codex 0.137 rejects the whole catalog on an effort it has no name for, such as the `max` Codex's own
gpt-6.1-terra entry offers, so a release older than the entry's `minimal_client_version` gets the translated
entry, whose levels are trimmed to that release."""
stock = CodexStockModel.model_validate(
{
"slug": "gpt-6.1-terra",
"display_name": "GPT-6.1 Terra",
"priority": 3,
"visibility": "list",
"minimal_client_version": minimum,
"default_reasoning_level": "low",
"supported_reasoning_levels": [{"effort": "low"}, {"effort": "max"}],
}
)
(entry,) = json.loads(
codex_models_response_json(
(_row("gpt-6.1-terra", supported_reasoning_efforts=("low", "max")),),
stock={"gpt-6.1-terra": stock},
instructions=_PROMPT,
client_version=client_version,
).json
)["models"]
assert (entry["display_name"], [level["effort"] for level in entry["supported_reasoning_levels"]]) == (
display_name,
levels,
)
def test_known_model_keeps_codex_stock_entry_listed_under_the_served_id():
(entry,) = _models(_row("gpt-daybreak"))
@ -516,6 +537,27 @@ def test_configured_service_tiers_replace_stock_tiers_and_a_known_id_keeps_its_c
assert _tiers(entry) == expected
@pytest.mark.parametrize(
("rows", "alias_defaults"),
[
((_row("gpt-5.5", service_tiers=[]),), "codex"),
((_row("gpt-5.5", service_tiers=["ultrafast"]),), "codex"),
((_row("gpt-5.5", service_tiers=["priority"]),), "codex"),
((_row("gpt-5.5"),), "codex"),
((_row("team-55", upstream_model="gpt-5.5"),), "deployment"),
((_row("team-55", upstream_model="gpt-5.5", service_tiers=["priority"]),), "deployment"),
],
)
def test_codexs_fast_toggle_is_on_exactly_when_the_entry_offers_the_priority_tier(rows, alias_defaults):
"""Codex shows `/fast` for a speed tier `fast` as well as for a `priority` service tier, but only sends
the tier when `service_tiers` lists it (codex-rs `ModelPreset::supports_fast_mode` and
`get_service_tier`, 0.166.0), so a leftover speed tier is a `/fast` toggle that does nothing."""
(entry,) = _models(*rows, alias_defaults=alias_defaults)
offers_priority = "priority" in [tier["id"] for tier in entry["service_tiers"]]
assert ("fast" in entry["additional_speed_tiers"]) is offers_priority
@pytest.mark.parametrize(
"invalid", ["ultrafast", [""], [" "], [1], [{"name": "no id"}], [{"id": "x", "bogus": 1}], {"id": "x"}]
)
@ -724,12 +766,13 @@ def test_catalog_rows_read_the_router_by_each_entry_lookup_id():
rows = codex_catalog_rows(listing, (("gpt-5.5-team", "model_name_team-1_c0ffee"), ("plain", "plain")), router)
assert rows == (
assert tuple(replace(row, deployment_capabilities=()) for row in rows) == (
CodexCatalogRow(
id="gpt-5.5-team",
mode="chat",
max_input_tokens=7,
upstream_model="openai/gpt-5.5",
upstream_provider="openai",
display_name="Team 5.5",
service_tiers=(["ultrafast"], None),
supported_reasoning_efforts=router.get_model_group_info(
@ -741,11 +784,15 @@ def test_catalog_rows_read_the_router_by_each_entry_lookup_id():
CodexCatalogRow(
id="plain",
upstream_model="openai/some-unmapped-model",
upstream_provider="openai",
service_tiers=(None,),
search_tool_support=(None,),
configured_max_input_tokens=(None,),
),
)
catalog_window = GetModelCostMap.loaded_model_cost_map()["gpt-5.5"]["max_input_tokens"]
assert [deployment.max_input_tokens for deployment in rows[0].deployment_capabilities] == [catalog_window] * 2
assert rows[1].deployment_capabilities == (CodexDeploymentCapabilities(),)
def test_catalog_rows_read_an_alias_off_its_target_under_the_alias_name():
@ -765,11 +812,13 @@ def test_catalog_rows_read_an_alias_off_its_target_under_the_alias_name():
(row,) = codex_catalog_rows(listing, (("gpt-6", "gpt-6"),), router)
assert row == CodexCatalogRow(
assert len(row.deployment_capabilities) == 1
assert replace(row, deployment_capabilities=()) == CodexCatalogRow(
id="gpt-6",
mode="chat",
max_input_tokens=9,
upstream_model="openai/gpt-6-astra",
upstream_provider="openai",
service_tiers=(["ultrafast"],),
supported_reasoning_efforts=router.get_model_group_info("gpt-6-astra").supported_reasoning_efforts,
search_tool_support=(None,),
@ -944,3 +993,292 @@ def test_codex_tool_search_is_offered_off_the_route_and_model_info_and_only_when
"mixed": False,
}
assert models[0]["context_window"] == 1000000
_SOL_SLUG: Final = next(
slug for slug, model in bundled_codex_models().items() if model.default_service_tier is not None
)
@pytest.mark.parametrize(
"upstream",
[
"bedrock/us.openai.{slug}",
"bedrock/global.openai.{slug}",
"bedrock/converse/us.openai.{slug}",
"bedrock/openai.{slug}",
"bedrock_mantle/openai.{slug}",
"bedrock_mantle/us-east-1/openai.{slug}",
"openai/{slug}",
"{slug}",
],
)
@pytest.mark.parametrize("slug", sorted(bundled_codex_models()))
def test_a_model_codex_knows_keeps_codexs_entry_under_any_provider_prefix(upstream: str, slug: str) -> None:
stock: Final = bundled_codex_models()[slug].model_dump(mode="json")
(entry,) = json.loads(
codex_models_response_json(
(_row("team-model", upstream_model=upstream.format(slug=slug)),), byte_limit=None
).json
)["models"]
assert entry["model_messages"] == stock["model_messages"]
assert entry["supported_reasoning_levels"] == stock["supported_reasoning_levels"]
@pytest.mark.parametrize("model", ["us.openai.{slug}", "converse/global.openai.{slug}"])
def test_a_bedrock_deployment_naming_its_provider_apart_from_the_model_keeps_codexs_entry(model: str) -> None:
router: Final = litellm.Router(
model_list=[
{
"model_name": "team-model",
"litellm_params": {
"model": model.format(slug=_SOL_SLUG),
"custom_llm_provider": "bedrock",
"aws_region_name": "us-east-1",
},
}
]
)
entry: Final = _served(router, "team-model")["team-model"]
assert entry["model_messages"] == bundled_codex_models()[_SOL_SLUG].model_dump(mode="json")["model_messages"]
def test_an_upstream_with_no_provider_prefix_matches_only_by_its_own_name() -> None:
(unknown, known) = json.loads(
codex_models_response_json(
(
_row("first", upstream_model=f"us.openai.{_SOL_SLUG}"),
_row("second", upstream_model=_SOL_SLUG),
),
instructions=_PROMPT,
).json
)["models"]
assert "model_messages" not in unknown and unknown["base_instructions"] == _PROMPT
assert known["model_messages"] == bundled_codex_models()[_SOL_SLUG].model_dump(mode="json")["model_messages"]
@pytest.mark.parametrize(
"upstream",
["bedrock/us.openai.{slug}-mini", "bedrock/us.meta.{slug}x", "bedrock_mantle/openai.{slug}.1", "{slug}-0"],
)
def test_a_model_whose_name_only_starts_like_one_codex_knows_gets_a_translated_entry(upstream: str) -> None:
(entry,) = json.loads(
codex_models_response_json(
(_row("team-model", upstream_model=upstream.format(slug=_SOL_SLUG)),), instructions=_PROMPT
).json
)["models"]
assert "model_messages" not in entry and entry["base_instructions"] == _PROMPT
def _listing(*names: str, max_input_tokens: int | None = None) -> tuple[dict[str, object], ...]:
return tuple(
{
"id": name,
"object": "model",
"created": 0,
"owned_by": "x",
"mode": "chat",
"max_input_tokens": max_input_tokens,
}
for name in names
)
def _served(
router: litellm.Router, *names: str, team_id: str | None = None, **listing: int
) -> dict[str, Mapping[str, object]]:
body: Final = codex_model_list_body(
_listing(*names, **listing), tuple((name, name) for name in names), router, team_id, client_version="0.166.0"
)
return {entry["slug"]: entry for entry in json.loads(body)["models"]}
def _vllm(name: str, host: str, **model_info: object) -> dict[str, object]:
return {
"model_name": name,
"litellm_params": {"model": f"hosted_vllm/coder-{host}", "api_base": f"https://{host}.example/v1"},
"model_info": model_info,
}
def test_a_text_only_deployment_takes_images_off_the_name_and_a_silent_one_does_not() -> None:
router: Final = litellm.Router(
model_list=[
_vllm("mixed", "h0", supports_vision=True),
_vllm("mixed", "h1", supports_vision=False),
_vllm("vision-and-silent", "h2", supports_vision=True),
_vllm("vision-and-silent", "h3"),
]
)
served: Final = _served(router, "mixed", "vision-and-silent")
assert served["mixed"]["input_modalities"] == ["text"]
assert served["vision-and-silent"]["input_modalities"] == ["text", "image"]
def test_a_silent_deployment_does_not_vote_against_tool_calls_but_one_without_them_does() -> None:
router: Final = litellm.Router(
model_list=[
_vllm("tools-and-silent", "h4", supports_function_calling=True),
_vllm("tools-and-silent", "h5"),
_vllm("tools-and-none", "h6", supports_function_calling=True),
_vllm("tools-and-none", "h7", supports_function_calling=False),
]
)
served: Final = _served(router, "tools-and-silent", "tools-and-none")
assert served["tools-and-silent"]["apply_patch_tool_type"] == "freeform"
assert served["tools-and-none"]["apply_patch_tool_type"] is None
def test_a_mixed_group_takes_the_smallest_window_any_deployment_declares() -> None:
router: Final = litellm.Router(
model_list=[
_vllm("mixed", "h8", max_input_tokens=200_000),
_vllm("mixed", "h9", max_input_tokens=128_000, supports_vision=True),
_vllm("mixed", "h10"),
_vllm("undeclared", "h11"),
]
)
served: Final = _served(router, "mixed", "undeclared", max_input_tokens=200_000)
assert served["mixed"]["context_window"] == 128_000
assert served["undeclared"]["context_window"] == 200_000
def test_a_claude_like_deployment_gets_codexs_coding_setup() -> None:
router: Final = litellm.Router(
model_list=[
_vllm(
"claude-like",
"h-claude",
max_input_tokens=1_000_000,
supports_vision=True,
supports_function_calling=True,
supports_reasoning=True,
),
_vllm("plain", "h12"),
_vllm("no-reasoning", "h12b", supports_reasoning=False),
]
)
served: Final = _served(router, "claude-like", "plain", "no-reasoning")
claude_like: Final = served["claude-like"]
assert claude_like["apply_patch_tool_type"] == "freeform"
assert claude_like["truncation_policy"] == {"mode": "tokens", "limit": 10000}
assert claude_like["supports_reasoning_summary_parameter"] is True
assert claude_like["context_window"] == 1_000_000
assert served["plain"]["apply_patch_tool_type"] is None
assert "supports_reasoning_summary_parameter" not in served["plain"]
assert served["no-reasoning"]["supports_reasoning_summary_parameter"] is False
def test_capabilities_are_read_only_off_the_deployments_the_key_team_routes_to() -> None:
router: Final = litellm.Router(
model_list=[
_vllm("coder", "h13", supports_vision=True),
_vllm("coder", "h14", supports_vision=False, team_id="team-2"),
]
)
assert _served(router, "coder", team_id="team-1")["coder"]["input_modalities"] == ["text", "image"]
assert _served(router, "coder", team_id="team-2")["coder"]["input_modalities"] == ["text"]
def _tier_fields(entry: Mapping[str, object]) -> tuple[object, object, object]:
return (entry["service_tiers"], entry.get("additional_speed_tiers"), entry.get("default_service_tier"))
def test_codexs_own_tiers_stay_only_when_every_deployment_behind_the_name_prices_them() -> None:
stock: Final = bundled_codex_models()[_SOL_SLUG].model_dump(mode="json")
assert stock["service_tiers"] and stock["additional_speed_tiers"] and stock["default_service_tier"]
token_price: Final = {"input_cost_per_token": 1e-06}
priority_price: Final = {**token_price, "input_cost_per_token_priority": 1e-05}
def upstream(host: str) -> dict[str, str]:
return {"model": f"hosted_vllm/{_SOL_SLUG}", "api_base": f"https://{host}.example/v1"}
router: Final = litellm.Router(
model_list=[
{"model_name": "all-priced", "litellm_params": upstream("a"), "model_info": priority_price},
{"model_name": "all-priced", "litellm_params": upstream("b"), "model_info": priority_price},
{"model_name": "half-priced", "litellm_params": upstream("c"), "model_info": priority_price},
{"model_name": "half-priced", "litellm_params": upstream("d"), "model_info": token_price},
{"model_name": "priced-and-unknown", "litellm_params": upstream("f"), "model_info": priority_price},
{"model_name": "priced-and-unknown", "litellm_params": {"model": "azure/prod-sol-eastus"}},
{
"model_name": "configured",
"litellm_params": upstream("e"),
"model_info": {"service_tiers": ["priority"]},
},
]
)
served: Final = _served(router, "all-priced", "half-priced", "priced-and-unknown", "configured")
assert _tier_fields(served["all-priced"]) == _tier_fields(stock)
assert _tier_fields(served["priced-and-unknown"]) == _tier_fields(stock)
assert _tier_fields(served["half-priced"]) == ([], [], None)
assert served["half-priced"]["model_messages"] == stock["model_messages"]
assert [tier["id"] for tier in served["configured"]["service_tiers"]] == ["priority"]
def _tiered_stock(default_service_tier: str) -> Mapping[str, CodexStockModel]:
tiers: Final = (
CodexServiceTier(id="priority", name="Fast", description="faster"),
CodexServiceTier(id="flex", name="Flex", description="cheaper"),
)
return MappingProxyType(
{
"tiered": CodexStockModel(
slug="tiered",
display_name="Tiered",
priority=0,
visibility="list",
service_tiers=tiers,
default_service_tier=default_service_tier,
)
}
)
@pytest.mark.parametrize(("default_tier", "kept_default"), [("priority", "priority"), ("flex", None)])
def test_dropping_one_stock_tier_keeps_the_default_only_while_it_is_still_offered(default_tier, kept_default) -> None:
priority_and_flex: Final = CodexDeploymentCapabilities(
prices_tokens=True,
priced_service_tier_keys=frozenset(
get_service_tier_cost_key("input_cost_per_token", tier) for tier in ("priority", "flex")
),
)
priority_only: Final = CodexDeploymentCapabilities(
prices_tokens=True,
priced_service_tier_keys=frozenset({get_service_tier_cost_key("input_cost_per_token", "priority")}),
)
row: Final = _row("tiered", deployment_capabilities=(priority_and_flex, priority_only))
(entry,) = json.loads(
codex_models_response_json((row,), stock=_tiered_stock(default_tier), instructions=_PROMPT).json
)["models"]
assert [tier["id"] for tier in entry["service_tiers"]] == ["priority"]
assert entry["default_service_tier"] == kept_default
def test_model_info_that_does_not_validate_declares_no_capability_but_keeps_its_prices() -> None:
priced: Final = get_service_tier_cost_key("input_cost_per_token", "priority")
capabilities: Final = deployment_capabilities(
{"supports_vision": "sometimes", "supports_function_calling": False, priced: 1e-05}, "odd"
)
assert capabilities == CodexDeploymentCapabilities(priced_service_tier_keys=frozenset({priced}))

224
tests/unit/proxy/common_utils/test_codex_model_translation.py Normal file
View file

@ -0,0 +1,224 @@
import json
from typing import Final
import pytest
from litellm.litellm_core_utils.llm_cost_calc.utils import get_service_tier_cost_key
from litellm.proxy.common_utils.codex_model_translation import (
CodexCapabilities,
CodexDeploymentCapabilities,
CodexServiceTier,
codex_fallback_entry,
combined_capabilities,
no_priced_deployment_lacks_tier,
)
_PROMPT: Final = "base prompt"
_SILENT: Final = CodexDeploymentCapabilities()
_ABSENT: Final = object()
def _entry(capabilities: CodexCapabilities, **kwargs: object) -> dict[str, object]:
entry: Final = codex_fallback_entry(
slug="m", display_name="M", priority=3, capabilities=capabilities, instructions=_PROMPT, **kwargs
)
return json.loads(entry.model_dump_json())
_CLAUDE_LIKE: Final = CodexCapabilities(
context_window=1_000_000, supports_vision=True, supports_function_calling=True, supports_reasoning=True
)
def test_a_model_that_calls_tools_and_reasons_gets_codexs_own_coding_setup():
entry: Final = _entry(_CLAUDE_LIKE)
assert entry["apply_patch_tool_type"] == "freeform"
assert entry["truncation_policy"] == {"mode": "tokens", "limit": 10000}
assert entry["supports_reasoning_summary_parameter"] is True
assert entry["supports_reasoning_summaries"] is True
assert entry["input_modalities"] == ["text", "image"]
assert entry["context_window"] == 1_000_000
@pytest.mark.parametrize(
("flag", "field", "by_value"),
[
("supports_function_calling", "apply_patch_tool_type", {True: "freeform", False: None, None: None}),
("supports_reasoning", "supports_reasoning_summary_parameter", {True: True, False: False, None: _ABSENT}),
("supports_reasoning", "supports_reasoning_summaries", {True: True, False: False, None: False}),
("supports_parallel_function_calling", "supports_parallel_tool_calls", {True: True, False: False, None: False}),
],
)
@pytest.mark.parametrize("value", [True, False, None])
def test_each_codex_feature_follows_what_the_deployments_declare(flag, field, by_value, value):
"""A missing `supports_reasoning_summary_parameter` is Codex's own default (true, `default_true` in
`codex-rs/protocol/src/openai_models.rs` at rust-v0.166.0), so a model nobody describes keeps it."""
entry: Final = _entry(CodexCapabilities(**{flag: value}))
assert entry.get(field, _ABSENT) == by_value[value]
@pytest.mark.parametrize(
("vision", "modalities"), [(False, ["text"]), (True, ["text", "image"]), (None, ["text", "image"])]
)
def test_images_are_offered_unless_a_deployment_cannot_read_them(vision, modalities):
assert _entry(CodexCapabilities(supports_vision=vision))["input_modalities"] == modalities
def test_the_rest_of_the_entry_is_what_the_listing_decides():
tier: Final = CodexServiceTier(id="flex", name="Flex", description="cheaper")
entry: Final = _entry(
CodexCapabilities(),
service_tiers=(tier,),
supports_search_tool=True,
reasoning_efforts=("low", "high"),
client_version="0.166.0",
)
assert (entry["slug"], entry["display_name"], entry["priority"], entry["base_instructions"]) == (
"m",
"M",
3,
_PROMPT,
)
assert entry["service_tiers"] == [{"id": "flex", "name": "Flex", "description": "cheaper"}]
assert entry["supports_search_tool"] is True
assert [level["effort"] for level in entry["supported_reasoning_levels"]] == ["low", "high"]
assert entry["default_reasoning_level"] == "low"
assert entry["context_window"] is None
assert (entry["shell_type"], entry["visibility"], entry["supported_in_api"]) == ("unified_exec", "list", True)
assert (entry["support_verbosity"], entry["use_responses_lite"]) == (False, False)
assert "tool_mode" not in entry
_REQUIRED_SINCE_CODEX_0_140: Final = frozenset(
{
"slug",
"display_name",
"description",
"supported_reasoning_levels",
"shell_type",
"visibility",
"supported_in_api",
"priority",
"availability_nux",
"upgrade",
"base_instructions",
"supports_reasoning_summaries",
"support_verbosity",
"default_verbosity",
"apply_patch_tool_type",
"truncation_policy",
"supports_parallel_tool_calls",
"experimental_supported_tools",
}
)
_CODEX_ENUMS: Final = {
"shell_type": {"unified_exec", "disabled"},
"visibility": {"list", "hide", "none"},
"apply_patch_tool_type": {"freeform", None},
}
_CODEX_MODALITIES: Final = {"text", "image"}
_CODEX_TRUNCATION_MODES: Final = {"bytes", "tokens"}
@pytest.mark.parametrize("described", [True, False], ids=["described", "undescribed"])
def test_every_entry_carries_the_fields_and_enum_values_every_supported_codex_decodes(described):
"""`codex-rs/protocol/src/openai_models.rs` ModelInfo at rust-v0.140.0 and rust-v0.166.0, the fields with
no `#[serde(default)]` and the variants of InputModality, ConfigShellToolType, ModelVisibility,
ApplyPatchToolType and TruncationMode in both, read on 2026-11-02. Codex rejects the whole catalog when
one entry does not decode."""
entry: Final = _entry(
CodexCapabilities(
context_window=128_000,
supports_vision=False,
supports_function_calling=True,
supports_parallel_function_calling=True,
supports_reasoning=True,
)
if described
else CodexCapabilities()
)
assert entry.keys() >= _REQUIRED_SINCE_CODEX_0_140
assert all(entry[field] in values for field, values in _CODEX_ENUMS.items())
assert set(entry["input_modalities"]) <= _CODEX_MODALITIES and "text" in entry["input_modalities"]
assert entry["truncation_policy"]["mode"] in _CODEX_TRUNCATION_MODES
def test_a_deployment_that_says_nothing_does_not_vote():
declares: Final = CodexDeploymentCapabilities(
supports_vision=True,
supports_function_calling=True,
supports_parallel_function_calling=True,
supports_reasoning=True,
)
assert combined_capabilities((declares, _SILENT)) == CodexCapabilities(
supports_vision=True,
supports_function_calling=True,
supports_parallel_function_calling=True,
supports_reasoning=True,
)
assert combined_capabilities((_SILENT, _SILENT)) == CodexCapabilities()
assert combined_capabilities(()) == CodexCapabilities()
@pytest.mark.parametrize(
"flag",
["supports_vision", "supports_function_calling", "supports_parallel_function_calling", "supports_reasoning"],
)
def test_one_deployment_lacking_a_capability_takes_it_off_the_whole_name(flag):
has: Final = CodexDeploymentCapabilities(**{flag: True})
lacks: Final = CodexDeploymentCapabilities(**{flag: False})
for order in ((has, lacks, _SILENT), (_SILENT, lacks, has)):
assert getattr(combined_capabilities(order), flag) is False
@pytest.mark.parametrize(
("windows", "listed", "expected"),
[
((200_000, 128_000, None), 400_000, 128_000),
((None, 64_000), 400_000, 64_000),
((None, None), 400_000, 400_000),
((), 400_000, 400_000),
((0, -1, 32_000), None, 32_000),
((0,), None, None),
],
)
def test_the_window_is_the_smallest_any_deployment_declares(windows, listed, expected):
deployments: Final = tuple(CodexDeploymentCapabilities(max_input_tokens=window) for window in windows)
assert combined_capabilities(deployments, listed).context_window == expected
def _priced(*tier_ids: str) -> CodexDeploymentCapabilities:
return CodexDeploymentCapabilities(
prices_tokens=True,
priced_service_tier_keys=frozenset(
get_service_tier_cost_key("input_cost_per_token", tier) for tier in tier_ids
),
)
@pytest.mark.parametrize(
("deployments", "tier", "kept"),
[
((_priced("priority"), _priced("priority", "flex")), "priority", True),
((_priced("priority"), _priced("flex")), "priority", False),
((_priced("priority"), _priced()), "priority", False),
((_priced("priority"), _SILENT), "priority", True),
((_SILENT, _SILENT), "priority", True),
((_priced("priority"),), "fast", True),
((_priced("flex"),), "fast", False),
((_priced("priority", "flex"),), "made-up-tier", False),
((), "priority", True),
],
)
def test_a_tier_stays_unless_a_deployment_that_prices_tokens_has_no_price_for_it(deployments, tier, kept):
"""A deployment with no token price at all (a subscription upstream, a deployment name the cost map does
not know) says nothing about which tiers its upstream sells, so it does not vote."""
assert no_priced_deployment_lacks_tier(tier, deployments) is kept

2
tests/unit/proxy/proxy_server/test_routes_models.py
View file

@ -56,9 +56,11 @@ def patched_models(monkeypatch):
deployment.litellm_params.model = "gpt-4"
router.get_deployment_by_model_group_name = MagicMock(return_value=deployment)
router.get_routable_upstream_model = MagicMock(return_value="gpt-4")
router.get_routable_upstream_provider = MagicMock(return_value=None)
router.get_configured_display_name = MagicMock(return_value=None)
router.get_configured_service_tiers = MagicMock(return_value=())
router.get_configured_max_input_tokens = MagicMock(return_value=())
router.get_routable_deployment_model_info = MagicMock(return_value=())
router.get_model_group_info = MagicMock(return_value=None)
router.get_search_tool_support = MagicMock(return_value=())
route_entry_points_to_stubs(router)

4
tests/unit/proxy/router_entry_point_stubs.py
View file

@ -99,6 +99,8 @@ def route_entry_points_to_stubs(router: _RouterDouble) -> _RouterDouble:
return router.is_recognized_model(name)
if kind == "upstream":
return router.get_routable_upstream_model(name, **_given(team_id=team_id))
if kind == "upstream_provider":
return router.get_routable_upstream_provider(name, **_given(team_id=team_id))
if kind == "search_tool_support":
return router.get_search_tool_support(name, **_given(team_id=team_id))
if kind == "from_id":
@ -125,6 +127,8 @@ def route_entry_points_to_stubs(router: _RouterDouble) -> _RouterDouble:
return router.get_configured_service_tiers(name, **_given(team_id=team_id))
if field == "max_input_tokens":
return router.get_configured_max_input_tokens(name, **_given(team_id=team_id))
if field == "deployment_model_info":
return router.get_routable_deployment_model_info(name, **_given(team_id=team_id))
if field == "listing_info":
return router.get_model_listing_info(name)
if field == "mode":

21
tests/unit/test_router/test_router.py
View file

@ -7299,6 +7299,27 @@ def test_upstream_model_and_service_tiers_are_read_off_the_deployments_the_team_
assert router.get_configured_service_tiers(model_name, team_id) == service_tiers
@pytest.mark.parametrize(
("litellm_params", "provider"),
[
({"model": "us.openai.gpt-6.1-sol", "custom_llm_provider": "bedrock"}, "bedrock"),
({"model": "bedrock_mantle/openai.gpt-6.1-sol"}, "bedrock_mantle"),
({"model": "gpt-6.1-sol"}, None),
],
)
def test_the_upstream_provider_is_the_one_the_routable_deployment_declares(litellm_params, provider):
router = litellm.Router(
model_list=[
{"model_name": "sol", "litellm_params": litellm_params, "model_info": {"team_id": "team-1"}},
{"model_name": "sol", "litellm_params": {"model": "azure/sol-deployment"}, "model_info": {"team_id": "t2"}},
]
)
assert router.registry().resolve("sol", kind="upstream_provider", team_id="team-1") == provider
assert router.registry().resolve("sol", kind="upstream_provider", team_id="t2") == "azure"
assert router.registry().resolve("not-a-real-model", kind="upstream_provider") is None
def test_get_configured_display_name_treats_malformed_values_as_absent():
malformed = ["", " ", 12345, ["Kimi K3"], {"name": "Kimi K3"}, True]
router = litellm.Router(