feat(proxy): serve a Codex-native model catalog with per-model service_tiers from /v1/models (#44136)

* feat(proxy): serve a Codex-native model catalog with per-model service_tiers from /v1/models

GET /v1/models and /models answer Codex CLI's catalog fetch (the request
carrying its client_version query parameter) with Codex's own
{"models": [...]} shape: a model Codex knows keeps the metadata of its
bundled 0.159.3 catalog (vendored), any other model gets Codex's fallback
entry, and model_info.service_tiers becomes each entry's service tiers so
Codex offers them as slash commands that send service_tier upstream.
Without the parameter the OpenAI list shape is unchanged. The CLI's
litellm agents codex catalog shares the same builder.

* fix(proxy): offer a Codex service tier only when every deployment of the model lists it

* fix(codex-catalog): an invalid service_tiers value offers no tier for the model

* fix(codex-catalog): read service tiers off the deployments the key's team can route to

A tier is offered to Codex only when every deployment of the model name a
request from the key's team can route to lists it, so another team's
deployment of the name and a deployment an admin paused via model_info.blocked
no longer withhold or add tiers for requests that never reach them

The catalog's always-null fields are annotated NoneType so the module imports
under pydantic 2.12.0 on Python 3.14, the lowest pin the MCP resolve job
installs, which rejects a None annotation with a None default

* test(codex-catalog): drop the redundant module docstring and sort the imports

* test(integration): add the Codex catalog audit cells and the multi-worker convergence note

* test(integration): clean up every catalog test model and answer the refresh GET

* fix(proxy): keep tiered models under Codex's catalog cut and resolve alias tiers

Under Codex's 1 MiB catalog limit the entries offering a service tier are kept
ahead of those offering none, each group in model_list order, with every kept
entry at its listing position, so the model an operator configured tiers for
survives a wide key's long listing. A model_group_alias row reads its target's
deployments, so it carries the target's tiers and stock metadata under the
alias name.

* fix(proxy): pick Codex catalog metadata per team and skip entries too large for the cut

The upstream model that selects Codex's stock entry was read off the first deployment of a name
without checking the key's team, so a team whose requests route to a different deployment could be
handed another team's prompt, reasoning levels, and tiers. The upstream model and the tiers now come
from the same team-aware selection routing uses, and a caller with no team reads the deployments no
team owns

The byte cut kept a prefix of the tier-first order, so one entry larger than the whole limit emptied
the catalog. An entry too large for the bytes left is now passed over and the smaller ones after it
are still kept

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-03 20:42:55 +00:00 • committed by GitHub
parent fd14e51ecf
commit fe683ea139
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 4261 additions and 122 deletions

View file

@ -9,11 +9,19 @@ from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Final, Literal, TypeAlias
from typing import Final, TypeAlias
import click
import requests
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from pydantic import BaseModel, TypeAdapter, ValidationError
from litellm.proxy.common_utils.codex_model_catalog import (
CODEX_BASE_INSTRUCTIONS_PATH,
CodexCatalogRow,
CodexStockCatalog,
CodexStockModel,
codex_models_response_json,
)
from .auth import CliContextObj, context_secret_vault, get_stored_api_key, login
from .claude_settings import ClaudeSettingsError, install_statusline_script
@ -68,7 +76,6 @@ _HIDDEN_AGENTS: Final = frozenset({"pi"})
CODEX_PROXY_PROVIDER: Final = "litellm"
CODEX_HOME_ENV: Final = "CODEX_HOME"
CODEX_MODEL_CATALOG_FILENAME: Final = "litellm-models.json"
_CODEX_BASE_INSTRUCTIONS_PATH: Final = Path(__file__).with_name("codex_base_instructions.md")
_CODEX_PREFLIGHT_TIMEOUT_SECONDS: Final = 10.0
@ -390,96 +397,8 @@ def opencode_model_sync_env(
return MappingProxyType({OPENCODE_CONFIG_CONTENT_ENV: opencode_provider_config(base_url, listing)})
class _CodexTruncationPolicy(BaseModel):
mode: Literal["bytes"] = "bytes"
limit: int = 10_000
class _CodexModel(BaseModel):
"""One `ModelInfo` entry of a Codex model catalog for a model the installed 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.
"""
slug: str
display_name: str
description: None = None
supported_reasoning_levels: tuple[()] = ()
shell_type: Literal["unified_exec"] = "unified_exec"
visibility: Literal["list"] = "list"
supported_in_api: Literal[True] = True
priority: int
availability_nux: None = None
upgrade: None = None
support_verbosity: Literal[False] = False
supports_reasoning_summaries: Literal[False] = False
supports_parallel_tool_calls: Literal[False] = False
default_verbosity: None = None
apply_patch_tool_type: None = None
truncation_policy: _CodexTruncationPolicy = _CodexTruncationPolicy()
experimental_supported_tools: tuple[()] = ()
context_window: int | None
base_instructions: str
class _StockCodexUpgrade(BaseModel):
model_config = ConfigDict(extra="allow")
model: str
class _StockCodexModel(BaseModel):
"""One `ModelInfo` entry as the installed Codex prints it from `codex debug models`.
Only the fields the sync rewrites are named; everything else that release
knows about the model (its reasoning levels, prompt, tool support) rides
along untouched, whatever the release's schema.
"""
model_config = ConfigDict(extra="allow")
slug: str
priority: int
visibility: str
supported_in_api: bool = True
upgrade: _StockCodexUpgrade | None = None
class _StockCodexCatalog(BaseModel):
models: tuple[_StockCodexModel, ...]
class _CodexCatalog(BaseModel):
models: tuple[_CodexModel | _StockCodexModel, ...]
def _codex_catalog_entry(
priority: int,
listed: ListedModel,
stock: _StockCodexModel | None,
served: frozenset[str],
instructions: str,
) -> _CodexModel | _StockCodexModel:
if stock is None:
return _CodexModel(
slug=listed.id,
display_name=listed.id,
priority=priority,
context_window=listed.max_input_tokens,
base_instructions=instructions,
)
upgrade: Final = stock.upgrade if stock.upgrade is not None and stock.upgrade.model in served else None
return stock.model_copy(
update={"priority": priority, "visibility": "list", "supported_in_api": True, "upgrade": upgrade}
)
def codex_model_catalog(
models: Sequence[ListedModel], stock: Sequence[_StockCodexModel], instructions: str
models: Sequence[ListedModel], stock: Sequence[CodexStockModel], instructions: str
) -> str | None:
"""The `model_catalog_json` body listing the proxy's chat models, or None if there are none.
@ -490,19 +409,14 @@ def codex_model_catalog(
Codex hides it or keeps it off the API, and keeps Codex's upgrade nudge only
when the model it points at is served too. A model Codex does not know gets the fallback
entry, with the same base instructions Codex itself uses so the agent never
runs without a system prompt.
runs without a system prompt. The catalog is the one `/v1/models?client_version=...`
serves, built from the installed Codex's own entries instead of the proxy's vendored ones.
"""
chat_models: Final = _chat_models(models)
if not chat_models:
return None
served: Final = frozenset(m.id for m in chat_models)
known: Final = MappingProxyType({m.slug: m for m in stock})
catalog: Final = _CodexCatalog(
models=tuple(
_codex_catalog_entry(index, m, known.get(m.id), served, instructions) for index, m in enumerate(chat_models)
)
rows: Final = tuple(CodexCatalogRow(id=m.id, mode=m.mode, max_input_tokens=m.max_input_tokens) for m in models)
body: Final = codex_models_response_json(
rows, stock=MappingProxyType({m.slug: m for m in stock}), instructions=instructions, byte_limit=None
)
return catalog.model_dump_json()
return body.json if body.listed else None
def codex_model_catalog_path(env: Mapping[str, str], *, home: Callable[[], Path] = Path.home) -> Path:
@ -558,12 +472,12 @@ def _codex_debug_models(
def _stock_codex_models(
binary: str, env: Mapping[str, str], *, run: Callable[..., subprocess.CompletedProcess[str]]
) -> tuple[_StockCodexModel, ...] | ModelSyncSkipped:
) -> tuple[CodexStockModel, ...] | ModelSyncSkipped:
printed: Final = _codex_debug_models(binary, (), env, run=run)
if isinstance(printed, ModelSyncSkipped):
return printed
try:
return _StockCodexCatalog.model_validate_json(printed).models
return CodexStockCatalog.model_validate_json(printed).models
except ValidationError as e:
name: Final = os.path.basename(binary)
return ModelSyncSkipped(f"`{name} debug models` printed no model catalog: {e.errors()[0]['msg']}")
@ -578,7 +492,7 @@ def codex_model_sync_args(
get: Callable[..., requests.Response] = requests.get,
run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
home: Callable[[], Path] = Path.home,
instructions_path: Path = _CODEX_BASE_INSTRUCTIONS_PATH,
instructions_path: Path = CODEX_BASE_INSTRUCTIONS_PATH,
) -> ModelSyncArgs | ModelSyncSkipped:
"""`-c model_catalog_json=...` pointing Codex at the proxy's model list, or why it was skipped.

File diff suppressed because one or more lines are too long

View file

@ -0,0 +1,420 @@
"""Codex-native model catalog for the model listing endpoints.
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
``service_tiers``, which Codex turns into slash commands (``/ultrafast``) that send ``service_tier``
upstream.
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
``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 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
from pydantic import BaseModel, ConfigDict, StringConstraints, TypeAdapter, ValidationError
from litellm._logging import verbose_proxy_logger
if TYPE_CHECKING:
from litellm.router import Router
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_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 = "]}"
class CodexServiceTier(BaseModel):
"""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 _ConfiguredTier(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
id: _TierId
name: str | None = None
description: str | None = None
_CONFIGURED_TIERS: Final = TypeAdapter(tuple[_TierId | _ConfiguredTier, ...])
class CodexTruncationPolicy(BaseModel):
mode: Literal["bytes"] = "bytes"
limit: int = 10_000
class CodexFallbackModel(BaseModel):
"""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.
"""
slug: str
display_name: str
description: NoneType = None
supported_reasoning_levels: tuple[()] = ()
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
class CodexStockUpgrade(BaseModel):
model_config = ConfigDict(extra="allow")
model: str
class CodexStockModel(BaseModel):
"""One `ModelInfo` entry as Codex ships it or prints it from `codex debug models`.
Only the fields the listing rewrites are named; everything else that release
knows about the model (its reasoning levels, prompt, tool support) rides
along untouched, whatever the release's schema.
"""
model_config = ConfigDict(extra="allow")
slug: str
display_name: str
priority: int
visibility: str
supported_in_api: bool = True
upgrade: CodexStockUpgrade | None = None
service_tiers: tuple[CodexServiceTier, ...] = ()
default_service_tier: str | None = None
class CodexStockCatalog(BaseModel):
models: tuple[CodexStockModel, ...]
CodexCatalogEntry = CodexFallbackModel | CodexStockModel
@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, 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."""
id: str
mode: str | None = None
max_input_tokens: int | None = None
upstream_model: str | None = None
display_name: str | None = None
service_tiers: tuple[object, ...] = ()
@cache
def bundled_codex_models() -> Mapping[str, CodexStockModel]:
"""Codex's bundled catalog by slug, read once from the vendored file."""
catalog: Final = CodexStockCatalog.model_validate_json(CODEX_BUNDLED_MODELS_PATH.read_text(encoding="utf-8"))
return MappingProxyType({model.slug: model for model in catalog.models})
@cache
def codex_base_instructions() -> str:
return CODEX_BASE_INSTRUCTIONS_PATH.read_text(encoding="utf-8")
def _tier(item: str | _ConfiguredTier, known: Mapping[str, CodexServiceTier]) -> CodexServiceTier:
if isinstance(item, str) and item in known:
return known[item]
configured: Final = _ConfiguredTier(id=item) if isinstance(item, str) else item
return CodexServiceTier(
id=configured.id,
name=configured.name if configured.name is not None else configured.id.capitalize(),
description=(
configured.description
if configured.description is not None
else f"Sends service_tier={configured.id} upstream"
),
)
def _first_by_id(tiers: Sequence[CodexServiceTier]) -> tuple[CodexServiceTier, ...]:
return tuple(tier for index, tier in enumerate(tiers) if tier.id not in {seen.id for seen in tiers[:index]})
_NO_KNOWN_TIERS: Final[Mapping[str, CodexServiceTier]] = MappingProxyType({})
def _deployment_tiers(
raw: object, model_id: str, known: Mapping[str, CodexServiceTier]
) -> tuple[CodexServiceTier, ...]:
try:
items: Final = _CONFIGURED_TIERS.validate_python(raw)
except ValidationError as e:
verbose_proxy_logger.warning(
"model_info.service_tiers for %s offers no tier, expected a list of tier ids or {id, name, description} objects: %s",
model_id,
e.errors()[0]["msg"],
)
return ()
return _first_by_id(tuple(_tier(item, known) for item in items))
def configured_service_tiers(
per_deployment: Sequence[object], model_id: str, known: Mapping[str, CodexServiceTier] = _NO_KNOWN_TIERS
) -> tuple[CodexServiceTier, ...] | None:
"""The tiers `model_info.service_tiers` configures for `model_id`, given each of its deployments'
raw value (None where unset), or None when no deployment sets one.
Accepts tier id strings and `{id, name, description}` objects. A string naming one of the
model's `known` tiers (the ones Codex ships for it) keeps that tier's name and description, so
`"priority"` stays Codex's `/fast`; any other string `x` is the command `/x` described as
"Sends service_tier=x upstream". A duplicate id keeps its first entry. A request to `model_id`
can route to any of its deployments, so a tier is offered only when every deployment lists it,
in the first deployment's order; a deployment that sets none leaves nothing to offer. Each of
those cases is logged, and an invalid value on any deployment is logged and offers no tier for
the whole model, stock tiers included, so one typo never fails the listing and never offers a
tier nobody declared.
"""
declared: Final = tuple(raw for raw in per_deployment if raw is not None)
if not declared:
return None
parsed: Final = tuple(_deployment_tiers(raw, model_id, known) for raw in declared)
if len(parsed) < len(per_deployment):
verbose_proxy_logger.warning(
"model_info.service_tiers for %s is set on %d of its %d deployments, so no tier is offered: "
"a tier is offered only when every deployment lists it",
model_id,
len(parsed),
len(per_deployment),
)
return ()
tier_ids: Final = tuple(frozenset(tier.id for tier in tiers) for tiers in parsed)
shared: Final = tier_ids[0].intersection(*tier_ids[1:])
offered: Final = tuple(tier for tier in parsed[0] if tier.id in shared)
if len(offered) < len(parsed[0]):
verbose_proxy_logger.warning(
"model_info.service_tiers for %s differs between its deployments, offering only the tiers every one lists: %s",
model_id,
", ".join(tier.id for tier in offered) or "none",
)
return offered
def _codex_can_drive(row: CodexCatalogRow) -> bool:
return "*" not in row.id and (row.mode is None or row.mode in CODEX_CHAT_MODES)
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 _entry(
index: int,
row: CodexCatalogRow,
stock: Mapping[str, CodexStockModel],
served: frozenset[str],
instructions: str,
) -> CodexCatalogEntry:
stock_model: Final = _stock_for(row, stock)
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)
if stock_model is None:
return CodexFallbackModel(
slug=row.id,
display_name=row.display_name if row.display_name is not None else row.id,
priority=index,
service_tiers=configured if configured is not None else (),
context_window=row.max_input_tokens,
base_instructions=instructions,
)
upgrade: Final = (
stock_model.upgrade if stock_model.upgrade is not None and stock_model.upgrade.model in served else None
)
stock_name: Final = stock_model.display_name if row.id == stock_model.slug else row.id
listing: Final = {
"slug": row.id,
"display_name": row.display_name if row.display_name is not None else stock_name,
"priority": index,
"visibility": "list",
"supported_in_api": True,
"upgrade": upgrade,
}
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
),
}
)
def _serialized(entry: CodexCatalogEntry) -> str:
"""A stock entry keeps exactly the fields its source printed (an older Codex may know no
`service_tiers`), a fallback entry spells out every field Codex requires."""
return entry.model_dump_json(exclude_unset=isinstance(entry, CodexStockModel))
def _bytes_left_after(left: int, size: int) -> int:
return left - size if size <= left else left
def _fits(serialized: Sequence[str], byte_limit: int | None) -> tuple[bool, ...]:
"""Whether each entry, taken in order, fits in what is left of a body of at most `byte_limit`
bytes: an entry larger than the bytes left is passed over and a smaller one after it is still
taken, each entry counted with its separating comma and the last one's comma given back."""
if byte_limit is None:
return (True,) * len(serialized)
budget: Final = byte_limit - len(_BODY_PREFIX) - len(_BODY_SUFFIX) + 1
left: Final = tuple(
accumulate((len(entry.encode()) + 1 for entry in serialized), _bytes_left_after, initial=budget)
)
return tuple(after < before for before, after in pairwise(left))
def _kept_positions(
entries: Sequence[CodexCatalogEntry], serialized: Sequence[str], byte_limit: int | None
) -> frozenset[int]:
"""The listing positions that survive the byte limit: an entry offering a service tier is taken
ahead of one offering none, each group in listing order, since a model without a tier is driven
by name just as well off Codex's bundled catalog; an entry too large for the bytes left never
costs the smaller ones after it their place."""
survival: Final = sorted(range(len(entries)), key=lambda position: (not entries[position].service_tiers, position))
fits: Final = _fits(tuple(serialized[position] for position in survival), byte_limit)
return frozenset(position for position, fit in zip(survival, fits, strict=True) if fit)
@dataclass(frozen=True, slots=True)
class CodexCatalogBody:
"""A `ModelsResponse` body, the ids it lists, and the chat-capable ids the byte limit left out."""
json: str
listed: tuple[str, ...]
left_out: tuple[str, ...]
def codex_models_response_json(
rows: Sequence[CodexCatalogRow],
*,
stock: Mapping[str, CodexStockModel] | None = None,
instructions: str | None = None,
byte_limit: int | None = CODEX_CATALOG_BYTE_LIMIT,
) -> CodexCatalogBody:
"""Codex's `ModelsResponse` body for `rows`, in listing order, cut to Codex's byte limit.
Only chat-capable rows are listed (Codex cannot drive an embedding or image model, and a
wildcard id is no model), their order is their `priority`. Codex drops a catalog over its
byte limit silently and keeps its bundled one, so the body holds only the entries that fit,
the ones offering a service tier taken first and the rest in listing order, an entry too large
for the bytes left passed over, every kept entry at its listing position; `byte_limit=None`
keeps every entry.
"""
known: Final = bundled_codex_models() if stock is None else stock
prompt: Final = codex_base_instructions() if instructions is None else instructions
drivable: Final = tuple(row for row in rows if _codex_can_drive(row))
served: Final = frozenset(row.id for row in drivable)
entries: Final = tuple(_entry(index, row, known, served, prompt) for index, row in enumerate(drivable))
serialized: Final = tuple(_serialized(entry) for entry in entries)
kept: Final = _kept_positions(entries, serialized, byte_limit)
return CodexCatalogBody(
json=f"{_BODY_PREFIX}{','.join(serialized[position] for position in sorted(kept))}{_BODY_SUFFIX}",
listed=tuple(row.id for position, row in enumerate(drivable) if position in kept),
left_out=tuple(row.id for position, row in enumerate(drivable) if position not in kept),
)
def _catalog_row(
row: ModelInfoResponse, lookup_id: str, llm_router: Router | None, team_id: str | None
) -> CodexCatalogRow:
return CodexCatalogRow(
id=row["id"],
mode=row.get("mode"),
max_input_tokens=row.get("max_input_tokens"),
upstream_model=llm_router.get_routable_upstream_model(lookup_id, team_id) if llm_router is not None else None,
display_name=llm_router.get_configured_display_name(lookup_id) if llm_router is not None else None,
service_tiers=llm_router.get_configured_service_tiers(lookup_id, team_id) if llm_router is not None else (),
)
def codex_catalog_rows(
rows: Sequence[ModelInfoResponse],
entries: Sequence[tuple[str, str]],
llm_router: Router | None,
team_id: str | None = None,
) -> tuple[CodexCatalogRow, ...]:
"""`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 and the tiers are read only off the deployments its
requests can route to."""
lookup_ids: Final = MappingProxyType(dict(entries))
return tuple(_catalog_row(row, lookup_ids.get(row["id"], row["id"]), llm_router, team_id) for row in rows)
def codex_model_list_body(
rows: Sequence[ModelInfoResponse],
entries: Sequence[tuple[str, str]],
llm_router: Router | None,
team_id: str | None = None,
) -> str:
"""The `/v1/models?client_version=...` body, logging the models Codex's byte limit left out."""
body: Final = codex_models_response_json(codex_catalog_rows(rows, entries, llm_router, team_id))
if body.left_out:
verbose_proxy_logger.warning(
"Codex model catalog cut at %d bytes, left out: %s. Models offering a service tier are kept first, "
"then model_list order; list the models Codex users need first in model_list",
CODEX_CATALOG_BYTE_LIMIT,
", ".join(body.left_out),
)
return body.json

View file

@ -410,6 +410,7 @@ from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import (
AuthCacheInvalidationSubscriber,
)
from litellm.proxy.common_utils.callback_utils import initialize_callbacks_on_proxy
from litellm.proxy.common_utils.codex_model_catalog import codex_model_list_body
from litellm.proxy.common_utils.config_includes import resolve_include_file_path, resolve_includes
from litellm.proxy.common_utils.config_sync_pubsub import ConfigSyncSubscriber
from litellm.proxy.common_utils.debug_utils import init_verbose_loggers
@ -11593,6 +11594,7 @@ async def model_list(
fallback_type: str | None = None,
scope: str | None = None,
healthy_only: bool | None = False,
client_version: str | None = None,
):
"""
Use `/model/info` - to get detailed model information, example - pricing, mode, etc.
@ -11600,6 +11602,11 @@ async def model_list(
This is just for compatibility with openai projects like aider.
Query Parameters:
- client_version: Sent by Codex CLI (`?client_version=0.159.3`) when it fetches a
provider's model catalog. When present, the response is Codex's own
catalog shape (`{"models": [...]}`) built from the same listing, with
each model's `model_info.service_tiers` as its service tiers; absent,
the OpenAI shape below is returned
- include_metadata: Include additional metadata in the response with fallback information
- fallback_type: Type of fallbacks to include ("general", "context_window", "content_policy")
Defaults to "general" when include_metadata=true
@ -11742,8 +11749,16 @@ async def model_list(
model_info["id"] = response_id
model_data.append(model_info)
admin_listing: Final = cast(Sequence[ModelInfoResponse], model_data) # cast-ok: rows built above
if client_version is not None:
return Response(
content=codex_model_list_body(
admin_listing, admin_entries, llm_router, team_id or user_api_key_dict.team_id
),
media_type="application/json",
)
if wants_anthropic_format:
admin_listing: Final = cast(Sequence[ModelInfoResponse], model_data) # cast-ok: rows built above
return create_anthropic_model_list_response(
admin_listing,
display_names=configured_display_names(admin_entries, llm_router),
@ -11802,8 +11817,14 @@ async def model_list(
model_info["id"] = response_id
model_data.append(model_info)
listing: Final = cast(Sequence[ModelInfoResponse], model_data) # cast-ok: rows built above
if client_version is not None:
return Response(
content=codex_model_list_body(listing, entries, llm_router, team_id or user_api_key_dict.team_id),
media_type="application/json",
)
if wants_anthropic_format:
listing: Final = cast(Sequence[ModelInfoResponse], model_data) # cast-ok: rows built above
return create_anthropic_model_list_response(
listing,
display_names=configured_display_names(entries, llm_router),

View file

@ -476,6 +476,7 @@ def _stream_chunks_have_generated_content(chunks: Sequence[ModelResponseStream])
_NO_SESSION_KWARGS: Final[Mapping[str, Mapping[str, object]]] = MappingProxyType({})
_SESSION_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_MODEL_INFO_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_SILENT_MODEL_ADAPTER: Final = TypeAdapter(str | list[str])
_ROUTING_KWARGS_ADAPTER: Final[TypeAdapter[Mapping[str, object] | None]] = TypeAdapter(Mapping[str, object] | None)
_DEPLOYMENT_SELECTED_EVENT: Final = "litellm.request.deployment_selected"
@ -500,6 +501,10 @@ def _deployment_pick_attributes(model: str, request_kwargs: Mapping[str, object]
)
def _configured_model_info(deployment: Mapping[str, object]) -> Mapping[str, object]:
return _MODEL_INFO_ADAPTER.validate_python(deployment.get("model_info") or {})
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 ()
@ -10563,6 +10568,63 @@ class Router:
return display_name
return None
def routable_model_group(self, model_name: str) -> str:
"""
The model group a request to model_name routes to: its target when
model_name is a `model_group_alias`, else model_name itself.
"""
target: Final = self._get_model_from_alias(model_name)
return target if target is not None else model_name
def _routable_deployments(self, model_name: str, team_id: str | None) -> tuple[DeploymentTypedDict, ...]:
"""
The deployments a request from team_id to model_name can route to, in
model_list order and via the same O(1) index lookup as
get_configured_display_name, selected the way routing selects them: a
`model_group_alias` reads its target's deployments, another team's
deployment of the name is left out, a caller with no team reads the
deployments no team owns (every deployment of the name when a team owns
each one, which is what an admin's request routes to), and one an admin
paused via `LiteLLM_ProxyModelTable.blocked` is left out.
Returns an empty tuple for wildcard-expanded or unknown names.
"""
named: Final = self._get_all_deployments(model_name=self.routable_model_group(model_name), team_id=team_id)
usable: Final = tuple(
deployment for deployment in named if self._deployment_usable_by_team(deployment, team_id)
)
return tuple(
deployment
for deployment in (usable or named)
if _configured_model_info(deployment).get("blocked") is not True
)
def get_configured_service_tiers(self, model_name: str, team_id: str | None = None) -> tuple[object, ...]:
"""
Return the service_tiers value each deployment a request from team_id
to model_name can route to (see _routable_deployments) configures in
its model_info, unvalidated and in model_list order, None for a
deployment that sets none; the caller validates the shape it expects
and decides how the deployments combine.
Returns an empty tuple for wildcard-expanded or unknown names.
"""
return tuple(
_configured_model_info(deployment).get("service_tiers")
for deployment in self._routable_deployments(model_name, team_id)
)
def get_routable_upstream_model(self, model_name: str, team_id: str | None = None) -> str | None:
"""
Return the `litellm_params.model` of the first deployment a request
from team_id to model_name can route to (see _routable_deployments).
Returns None for wildcard-expanded or unknown names, and when the team
can route to no deployment of the name.
"""
deployment: Final = next(iter(self._routable_deployments(model_name, team_id)), None)
return deployment["litellm_params"].get("model") if deployment is not None else None
def get_credential_deployment(self, model_id: str, team_id: str | None = None) -> Deployment | None:
"""
The deployment a passthrough endpoint (files, batches, etc.) resolves for a

View file

@ -314,7 +314,8 @@ include = [
"litellm/router_strategy/complexity_router/artifacts/*.json",
"litellm/router_strategy/complexity_router/fuse_presets.json",
"litellm/proxy/model_insights_tasks.json",
"litellm/proxy/client/cli/commands/codex_base_instructions.md",
"litellm/proxy/common_utils/codex_base_instructions.md",
"litellm/proxy/common_utils/codex_bundled_models_0.159.3.json",
"litellm/proxy/lens/prompts/*.md",
]
exclude = [

View file

@ -54,13 +54,9 @@ def get_all_functions_called_in_tests(base_dir):
if file == "test_router_validate_fallbacks.py":
print(f"tree: {tree}")
for node in ast.walk(tree):
if isinstance(node, ast.Call) and isinstance(
node.func, ast.Name
):
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
called_functions.add(node.func.id)
elif isinstance(node, ast.Call) and isinstance(
node.func, ast.Attribute
):
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
called_functions.add(node.func.attr)
return called_functions
@ -91,6 +87,8 @@ ignored_function_names = [
"_get_claude_code_session_router_binding", # Tested through the two-worker session routing test in test_router.py
"_apply_updated_routing_strategy_args", # Tested via update_settings in test_lowest_latency.py (file lacks "router" in name)
"arm_routing_read_prefetch", # Tested in tests/unit/caching/test_request_redis_batch_pre_call.py (file lacks "router" in name)
"_configured_model_info", # Tested through get_configured_service_tiers in test_router.py
"_routable_deployments", # Tested through get_configured_service_tiers and get_routable_upstream_model in test_router.py
"_async_get_available_deployment", # Body of the `route {model}` phase wrapper, exercised through async_get_available_deployment in test_router.py
"_async_get_available_deployment_for_pass_through", # Same, through async_get_available_deployment_for_pass_through in test_router.py
"_embedding",
@ -109,9 +107,7 @@ def main():
# "../../litellm/router_utils/pattern_match_deployments.py",
# "../../litellm/router_utils/batch_utils.py",
# ] ## LOCAL TESTING
tests_dir = (
"./tests/" # Update this path if your tests directory is located elsewhere
)
tests_dir = "./tests/" # Update this path if your tests directory is located elsewhere
# tests_dir = "../../tests/" # LOCAL TESTING
router_functions = []
@ -119,9 +115,7 @@ def main():
router_functions.extend(get_functions_from_router(file))
print("router_functions: ", router_functions)
called_functions_in_tests = get_all_functions_called_in_tests(tests_dir)
untested_functions = [
fn for fn in router_functions if fn not in called_functions_in_tests
]
untested_functions = [fn for fn in router_functions if fn not in called_functions_in_tests]
if untested_functions:
all_untested_functions = []

View file

@ -4,7 +4,7 @@ These tests exercise a running gateway, PostgreSQL and Redis with an owned local
The `cost` group is driven by `cost_tracking_cases.json`, which contains the cost map, literal requests, literal provider responses and expected accounting values. Each case has a name, contract ID, cost-map model, optional deployment overrides, request body, tagged response and exact or recount expectations. Request bodies use `$MODEL` for the registered proxy model, while responses use `$REQUEST_ID` for the per-run scenario ID. To add a case, add a cost-map entry when the model is new, add the request body and exact provider response data, and add hand-computed expected values. The upstream serves each stored response for any path under `/<scenario_id>`, while the test-owned cost map is served over loopback through `LITELLM_MODEL_COST_MAP_URL`
Use `tests/integration/run.py management`, `accounting`, `database`, `providers`, `extensions`, `mcp`, `sdk` or `cost` to run a selected group. The group to directory mapping is the `GROUPS` literal at the top of `run.py`; a new directory needs a `GROUPS` entry and an `OWNED_DIRECTORIES` entry in `_support/manifest.py`. Set `INTEGRATION_WORKERS` above 1 to run a group under pytest-xdist; the `mcp` job does this in CI, so MCP tests must own their resources per scenario. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
Use `tests/integration/run.py management`, `accounting`, `database`, `providers`, `extensions`, `mcp`, `sdk` or `cost` to run a selected group. The group to directory mapping is the `GROUPS` literal at the top of `run.py`; a new directory needs a `GROUPS` entry and an `OWNED_DIRECTORIES` entry in `_support/manifest.py`. Set `INTEGRATION_WORKERS` above 1 to run a group under pytest-xdist; the `mcp` job does this in CI, so MCP tests must own their resources per scenario. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. When that deployment runs more than one proxy worker, set `INTEGRATION_PROXY_WORKERS` to the count so a test that writes a model and then calls it waits out the config reload interval, the only cross-worker convergence bound the wire exposes. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
Management also requires `INTEGRATION_PEER_URL`, `REDIS_HOST` and `REDIS_PORT`. CircleCI starts two directly addressed proxy processes sharing only that job's stores. The test-only CLI wrapper supplies enterprise route entitlement, following the existing behavior suite's convention. It does not qualify license validation; run it with one worker and no reload

View file

@ -0,0 +1,826 @@
import asyncio
import json
import os
import time
import uuid
from collections.abc import Callable, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from hashlib import sha256
from itertools import chain
from pathlib import Path
from typing import Final, TypeVar
import anthropic
import httpx
import openai
from pydantic import JsonValue
from litellm.constants import PROXY_CONFIG_RELOAD_INTERVAL_SECONDS
from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value
from tests.integration._support.database import read_rows
from tests.integration._support.wire import Reply, Request, Wire, wire_server
T = TypeVar("T")
CLIENT_VERSION: Final = "0.159.3"
CONSISTENT_READS: Final = 6
CONVERGENCE_SECONDS: Final = 60
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"
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"}
USAGE: Final = {"prompt_tokens": 11, "completion_tokens": 4, "total_tokens": 15}
def _generic_tier(identity: str) -> dict[str, JsonValue]:
return {"id": identity, "name": identity.capitalize(), "description": f"Sends service_tier={identity} upstream"}
def _bundled_stock() -> dict[str, dict[str, JsonValue]]:
catalog: Final = JSON_OBJECT.validate_json(BUNDLED_STOCK_PATH.read_bytes())
models: Final = catalog["models"]
assert isinstance(models, list), catalog
return {string_value(object_value(model)["slug"]): object_value(model) for model in models}
def _catalog_response(
gateway: Gateway,
*,
key: str | None = None,
params: Mapping[str, str] | None = None,
headers: Mapping[str, str] | None = None,
path: str = "/v1/models",
) -> httpx.Response:
return gateway.request(
"GET",
path,
key=key,
params={"client_version": CLIENT_VERSION, **(params or {})},
headers={**FRESH_CONNECTION, **(headers or {})},
)
def _entries(response: httpx.Response) -> list[dict[str, JsonValue]]:
assert response.status_code == 200, response.text
body: Final = JSON_OBJECT.validate_json(response.content)
assert set(body) == {"models"}, response.text
models: Final = body["models"]
assert isinstance(models, list), response.text
return [object_value(entry) for entry in models]
def _catalog(
gateway: Gateway, *, key: str | None = None, params: Mapping[str, str] | None = None
) -> list[dict[str, JsonValue]]:
return _entries(_catalog_response(gateway, key=key, params=params))
def _by_slug(entries: Sequence[Mapping[str, JsonValue]]) -> dict[str, dict[str, JsonValue]]:
return {string_value(entry["slug"]): dict(entry) for entry in entries}
def _tier_ids(entry: Mapping[str, JsonValue]) -> tuple[str, ...]:
tiers: Final = entry["service_tiers"]
assert isinstance(tiers, list), entry
return tuple(string_value(object_value(tier)["id"]) for tier in tiers)
def _tiers_in(entries: Sequence[Mapping[str, JsonValue]]) -> dict[str, tuple[str, ...]]:
return {slug: _tier_ids(entry) for slug, entry in _by_slug(entries).items()}
def _tiers_by_slug(
gateway: Gateway, *, key: str | None = None, params: Mapping[str, str] | None = None
) -> dict[str, tuple[str, ...]]:
return _tiers_in(_catalog(gateway, key=key, params=params))
def _converged_catalog(
gateway: Gateway,
satisfied: Callable[[dict[str, dict[str, JsonValue]]], bool],
*,
key: str | None = None,
params: Mapping[str, str] | None = None,
) -> dict[str, dict[str, JsonValue]]:
return eventually(
lambda: _by_slug(_catalog(gateway, key=key, params=params)), satisfied, seconds=CONVERGENCE_SECONDS
)
def _settled(read: Callable[[], T], satisfied: Callable[[T], bool], *, seconds: float = CONVERGENCE_SECONDS) -> T:
reads: Final = eventually(
lambda: tuple(read() for _ in range(CONSISTENT_READS)),
lambda values: all(satisfied(value) for value in values),
seconds=seconds,
)
return reads[-1]
def _settled_on_every_worker(read: Callable[[], T], satisfied: Callable[[T], bool], *, written_at: float) -> T:
eventually(
lambda: (time.monotonic() - written_at, read()),
lambda stamped: stamped[0] >= WORKER_SYNC_SECONDS and satisfied(stamped[1]),
seconds=WORKER_SYNC_SECONDS + CONVERGENCE_SECONDS,
)
return _settled(read, satisfied)
def _plain_ids(gateway: Gateway, *, key: str | None = None, params: Mapping[str, str] | None = None) -> list[str]:
response: Final = gateway.request("GET", "/v1/models", key=key, params=params, headers=FRESH_CONNECTION)
assert response.status_code == 200, response.text
body: Final = JSON_OBJECT.validate_json(response.content)
assert set(body) == {"data", "object"} and body["object"] == "list", response.text
data: Final = body["data"]
assert isinstance(data, list), response.text
return [string_value(object_value(entry)["id"]) for entry in data]
def _model_info_rows(gateway: Gateway, model_name: str) -> list[dict[str, JsonValue]]:
entries: Final = gateway.get("/model/info")["data"]
assert isinstance(entries, list), entries
return [object_value(entry) for entry in entries if object_value(entry)["model_name"] == model_name]
def _delete_model_if_present(gateway: Gateway, identity: str) -> None:
if read_rows('SELECT model_id FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)):
gateway.post("/model/delete", {"id": identity})
assert read_rows('SELECT model_id FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)) == []
def _new_model(
scenario: Scenario,
name: str,
*,
model: str = "openai/gpt-4o-mini",
model_info: Mapping[str, JsonValue] | None = None,
api_base: str | None = None,
) -> str:
created: Final = scenario.gateway.post(
"/model/new",
{
"model_name": name,
"litellm_params": {
"model": model,
"api_key": "integration-provider-key",
"api_base": api_base or f"{scenario.gateway.upstream_url}/v1",
},
"model_info": dict(model_info) if model_info is not None else {},
},
)
identity: Final = string_value(object_value(created["model_info"])["id"])
scenario.cleanups.callback(_delete_model_if_present, scenario.gateway, identity)
return identity
def _model_name(prefix: str) -> str:
return f"{prefix}-{uuid.uuid4().hex}"
def test_client_version_answers_codex_catalog_shape(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model(model_info={"max_input_tokens": 2048})
catalog: Final = _converged_catalog(
gateway, lambda value: model in value and value[model]["context_window"] == 2048
)
slugs: Final = list(catalog)
entry: Final = catalog[model]
assert entry["display_name"] == model, entry
assert entry["priority"] == slugs.index(model), entry
assert entry["visibility"] == "list" and entry["supported_in_api"] is True, entry
assert entry["service_tiers"] == [], entry
assert entry["context_window"] == 2048, entry
assert entry["base_instructions"] == BASE_INSTRUCTIONS_PATH.read_text(encoding="utf-8"), entry["slug"]
def test_models_route_matches_v1_models(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model()
key: Final = scenario.key(models=[model])
routes: Final = eventually(
lambda: (_entries(_catalog_response(gateway, key=key, path="/models")), _catalog(gateway, key=key)),
lambda pair: pair[0] == pair[1],
seconds=CONVERGENCE_SECONDS,
)
assert [entry["slug"] for entry in routes[0]] == [model], routes
def test_plain_listing_keeps_openai_shape(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model(model_info={"service_tiers": ["priority"]})
key: Final = scenario.key(models=[model])
ids: Final = eventually(
lambda: _plain_ids(gateway, key=key), lambda value: value == [model], seconds=CONVERGENCE_SECONDS
)
assert ids == [model], ids
response: Final = gateway.request("GET", "/v1/models", key=key, headers=FRESH_CONNECTION)
assert "service_tiers" not in response.text and "base_instructions" not in response.text, response.text
def test_client_version_wins_over_anthropic_headers(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model()
key: Final = scenario.key(models=[model])
codex: Final = eventually(
lambda: _catalog_response(gateway, key=key, headers={**ANTHROPIC_HEADERS, "x-api-key": key}),
lambda response: [entry["slug"] for entry in _entries(response)] == [model],
seconds=CONVERGENCE_SECONDS,
)
assert [entry["slug"] for entry in _entries(codex)] == [model], codex.text
anthropic_listing: Final = gateway.request(
"GET", "/v1/models", key=key, headers={**ANTHROPIC_HEADERS, "x-api-key": key, **FRESH_CONNECTION}
)
assert anthropic_listing.status_code == 200, anthropic_listing.text
body: Final = JSON_OBJECT.validate_json(anthropic_listing.content)
assert set(body) == {"data", "first_id", "has_more", "last_id"}, anthropic_listing.text
def test_cursor_models_route_ignores_client_version(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model()
key: Final = scenario.key(models=[model])
response: Final = eventually(
lambda: _catalog_response(gateway, key=key, path="/cursor/v1/models"),
lambda value: value.status_code == 200 and model in value.text,
seconds=CONVERGENCE_SECONDS,
)
assert response.status_code == 200, response.text
body: Final = JSON_OBJECT.validate_json(response.content)
assert set(body) == {"data", "object"}, response.text
data: Final = body["data"]
assert isinstance(data, list) and [object_value(entry)["id"] for entry in data] == [model], response.text
def test_client_version_value_shapes(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model(model_info={"service_tiers": ["priority"]})
key: Final = scenario.key(models=[model])
for value in ("", "1", "a" * 5000):
response: Final = gateway.request(
"GET", "/v1/models", key=key, params={"client_version": value}, headers=FRESH_CONNECTION
)
assert [entry["slug"] for entry in _entries(response)] == [model], (value[:20], response.text)
repeated: Final = gateway.client.get(
"/v1/models?client_version=1&client_version=2",
headers={"Authorization": f"Bearer {key}", **FRESH_CONNECTION},
)
assert [entry["slug"] for entry in _entries(repeated)] == [model], repeated.text
eventually(
lambda: (_catalog(gateway, key=key), _catalog(gateway, key=key)),
lambda pair: pair[0] == pair[1] and _tiers_in(pair[0]) == {model: ("priority",)},
seconds=CONVERGENCE_SECONDS,
)
unauthenticated: Final = gateway.client.get(
"/v1/models", params={"client_version": CLIENT_VERSION}, headers=FRESH_CONNECTION
)
assert unauthenticated.status_code == 401, unauthenticated.text
def test_configured_tiers_on_unknown_model(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
ultrafast: Final = {"id": "ultrafast", "name": "Ultra", "description": "fast lane"}
model: Final = scenario.model(
model_info={
"service_tiers": ["priority", ultrafast],
"display_name": "Probe Unknown",
"max_input_tokens": 1234,
}
)
entry: Final = _converged_catalog(
gateway, lambda value: model in value and value[model]["context_window"] == 1234
)[model]
assert entry["service_tiers"] == [_generic_tier("priority"), ultrafast], entry
assert entry["display_name"] == "Probe Unknown", entry
assert entry["context_window"] == 1234, entry
assert entry["upgrade"] is None, entry
def test_stock_model_keeps_codex_entry(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
name: Final = _model_name("codex-stock")
_new_model(scenario, name, model="openai/gpt-5.5")
catalog: Final = _converged_catalog(gateway, lambda value: name in value)
slugs: Final = list(catalog)
stock: Final = _bundled_stock()["gpt-5.5"]
assert stock["upgrade"] is not None and stock["service_tiers"] != [], stock
expected: Final = {
**stock,
"slug": name,
"display_name": name,
"priority": slugs.index(name),
"visibility": "list",
"supported_in_api": True,
"upgrade": None,
}
assert catalog[name] == expected, catalog[name]
def test_stock_model_configured_tiers_override(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
stock: Final = _bundled_stock()["gpt-6-sol"]
stock_tiers: Final = stock["service_tiers"]
assert isinstance(stock_tiers, list) and len(stock_tiers) == 1, stock
stock_tier: Final = object_value(stock_tiers[0])
assert stock["default_service_tier"] == stock_tier["id"], stock
unknown_tier: Final = _model_name("codex-unknown-tier")
known_tier: Final = _model_name("codex-known-tier")
_new_model(scenario, unknown_tier, model="openai/gpt-6-sol", model_info={"service_tiers": ["ultrafast"]})
_new_model(scenario, known_tier, model="openai/gpt-6-sol", model_info={"service_tiers": [stock_tier["id"]]})
catalog: Final = _converged_catalog(gateway, lambda value: unknown_tier in value and known_tier in value)
assert catalog[unknown_tier]["service_tiers"] == [_generic_tier("ultrafast")], catalog[unknown_tier]
assert catalog[unknown_tier]["default_service_tier"] is None, catalog[unknown_tier]
assert catalog[known_tier]["service_tiers"] == [stock_tier], catalog[known_tier]
assert catalog[known_tier]["default_service_tier"] == stock_tier["id"], catalog[known_tier]
assert catalog[known_tier]["display_name"] == known_tier, catalog[known_tier]
def test_stock_model_empty_tiers_disable(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
name: Final = _model_name("codex-no-tiers")
_new_model(scenario, name, model="openai/gpt-6-sol", model_info={"service_tiers": []})
entry: Final = _converged_catalog(gateway, lambda value: name in value)[name]
assert entry["service_tiers"] == [] and entry["default_service_tier"] is None, entry
def test_two_deployments_offer_only_shared_tiers(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
shared: Final = _model_name("codex-shared")
half_set: Final = _model_name("codex-half-set")
_new_model(scenario, shared, model_info={"service_tiers": ["a", "b"]})
_new_model(scenario, shared, model_info={"service_tiers": ["b", "c"]})
_new_model(scenario, half_set, model_info={"service_tiers": ["a"]})
_new_model(scenario, half_set)
tiers: Final = _settled(
lambda: _tiers_by_slug(gateway),
lambda value: value.get(shared) == ("b",) and value.get(half_set) == (),
)
assert tiers[shared] == ("b",) and tiers[half_set] == (), tiers
def test_invalid_tiers_offer_nothing_and_keep_listing(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
as_string: Final = scenario.model(model_info={"service_tiers": "priority"})
stray_key: Final = scenario.model(
model_info={"service_tiers": [{"id": "x", "name": "X", "description": "d", "stray": 1}]}
)
empty_id: Final = scenario.model(model_info={"service_tiers": [{"id": "", "name": "", "description": ""}]})
control: Final = scenario.model(model_info={"service_tiers": ["priority"]})
written: Final = (as_string, stray_key, empty_id, control)
catalog: Final = _converged_catalog(gateway, lambda value: all(slug in value for slug in written))
assert catalog[as_string]["service_tiers"] == [], catalog[as_string]
assert catalog[stray_key]["service_tiers"] == [], catalog[stray_key]
assert catalog[empty_id]["service_tiers"] == [], catalog[empty_id]
assert catalog[control]["service_tiers"] == [_generic_tier("priority")], catalog[control]
def test_blocked_deployment_tiers_are_ignored(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
name: Final = _model_name("codex-blockable")
_new_model(scenario, name, model_info={"service_tiers": ["a", "b"]})
narrowing: Final = _new_model(scenario, name, model_info={"service_tiers": ["b", "c"]})
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value == ("b",))
blocked: Final = gateway.request("PATCH", f"/model/{narrowing}/update", {"blocked": True})
assert blocked.status_code == 200, blocked.text
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value == ("a", "b"))
unblocked: Final = gateway.request("PATCH", f"/model/{narrowing}/update", {"blocked": False})
assert unblocked.status_code == 200, unblocked.text
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value == ("b",))
def test_restricted_key_lists_only_its_models(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
tiered: Final = scenario.model(model_info={"service_tiers": ["priority"]})
other: Final = scenario.model(model_info={"service_tiers": ["flex"]})
key: Final = scenario.key(models=[tiered])
tiers: Final = eventually(
lambda: _tiers_by_slug(gateway, key=key),
lambda value: value == {tiered: ("priority",)},
seconds=CONVERGENCE_SECONDS,
)
assert tiers == {tiered: ("priority",)}, tiers
admin_tiers: Final = eventually(
lambda: _tiers_by_slug(gateway), lambda value: value.get(other) == ("flex",), seconds=CONVERGENCE_SECONDS
)
assert admin_tiers[other] == ("flex",), admin_tiers
def test_team_scoped_model_tiers(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
owning_team: Final = scenario.team()
other_team: Final = scenario.team()
public_name: Final = _model_name("codex-team-model")
_new_model(scenario, public_name, model_info={"team_id": owning_team, "service_tiers": ["priority"]})
owning_key: Final = scenario.key(team_id=owning_team)
other_key: Final = scenario.key(team_id=other_team)
owning_tiers: Final = _settled(
lambda: _tiers_by_slug(gateway, key=owning_key), lambda value: value.get(public_name) == ("priority",)
)
assert owning_tiers[public_name] == ("priority",), owning_tiers
admin_tiers: Final = _settled(
lambda: _tiers_by_slug(gateway, params={"team_id": owning_team}),
lambda value: value.get(public_name) == ("priority",),
)
assert admin_tiers[public_name] == ("priority",), admin_tiers
assert public_name not in _tiers_by_slug(gateway, key=other_key), other_key[:8]
assert public_name not in _tiers_by_slug(gateway)
def test_wildcard_deployment_never_listed(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model()
_new_model(scenario, "openai/*", model="openai/*", model_info={"service_tiers": ["priority"]})
flags: Final = {"return_wildcard_routes": "true"}
plain: Final = eventually(
lambda: _plain_ids(gateway, params=flags), lambda ids: "openai/*" in ids and model in ids
)
assert "openai/*" in plain, plain
slugs: Final = eventually(
lambda: [string_value(entry["slug"]) for entry in _catalog(gateway, params=flags)],
lambda values: model in values,
)
assert all("*" not in slug for slug in slugs), slugs
def test_a_tiered_model_listed_last_survives_the_byte_cut(gateway: Gateway) -> None:
"""Codex 0.159.3 rejects a `model_catalog_url` body over `MAX_MODEL_CATALOG_BYTES` (1 MiB,
`codex-rs/model-provider/src/models_endpoint.rs`, read on 2026-10-03), so the proxy keeps the
entries offering a tier first and then listing order; fallback entries carry Codex's base
prompt, so some fifty of them overrun the limit."""
with gateway.scenario() as scenario:
plain: Final = tuple(scenario.model() for _ in range(52))
tiered: Final = scenario.model(model_info={"service_tiers": ["ultrafast"]})
key: Final = scenario.key(models=[*plain, tiered])
listing: Final = eventually(
lambda: _plain_ids(gateway, key=key), lambda ids: set(ids) == {*plain, tiered}, seconds=CONVERGENCE_SECONDS
)
assert listing.index(tiered) == len(plain), listing
response: Final = eventually(
lambda: _catalog_response(gateway, key=key),
lambda value: value.status_code == 200 and tiered in {entry["slug"] for entry in _entries(value)},
seconds=CONVERGENCE_SECONDS,
)
entries: Final = _entries(response)
slugs: Final = [string_value(entry["slug"]) for entry in entries]
assert len(response.content) <= 1024 * 1024, len(response.content)
assert 0 < len(slugs) < len(listing), (len(slugs), len(listing))
assert slugs == [slug for slug in listing if slug in slugs], slugs
assert slugs[-1] == tiered and _tier_ids(entries[-1]) == ("ultrafast",), entries[-1]
assert slugs[:-1] == list(plain[: len(slugs) - 1]), slugs
assert [entry["priority"] for entry in entries] == [listing.index(slug) for slug in slugs], slugs
def test_an_oversized_tiered_model_is_left_out_alone(gateway: Gateway) -> None:
"""A tier description longer than Codex's whole byte limit makes an entry that can never fit;
it is taken first as a tiered entry, so the cut passes over it and still keeps the models after it."""
with gateway.scenario() as scenario:
plain: Final = scenario.model()
oversized_tier: Final = {"id": "huge", "name": "Huge", "description": "x" * (1024 * 1024)}
oversized: Final = scenario.model(model_info={"service_tiers": [oversized_tier]})
tiered: Final = scenario.model(model_info={"service_tiers": ["ultrafast"]})
key: Final = scenario.key(models=[plain, oversized, tiered])
listing: Final = eventually(
lambda: _plain_ids(gateway, key=key),
lambda ids: set(ids) == {plain, oversized, tiered},
seconds=CONVERGENCE_SECONDS,
)
assert listing == [plain, oversized, tiered], listing
def kept() -> tuple[int, tuple[tuple[JsonValue, JsonValue, tuple[str, ...]], ...]]:
response: Final = _catalog_response(gateway, key=key)
return len(response.content), tuple(
(entry["slug"], entry["priority"], _tier_ids(entry)) for entry in _entries(response)
)
expected: Final = ((plain, 0, ()), (tiered, 2, ("ultrafast",)))
size, entries = _settled(kept, lambda value: value[1] == expected)
assert entries == expected, entries
assert size <= 1024 * 1024, size
def test_model_info_echoes_configured_tiers(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
configured: Final = ["priority", {"id": "ultrafast", "name": "Ultra", "description": "fast lane"}]
model: Final = scenario.model(model_info={"service_tiers": configured})
matching: Final = eventually(
lambda: _model_info_rows(gateway, model), lambda rows: len(rows) == 1, seconds=CONVERGENCE_SECONDS
)
assert len(matching) == 1, model
assert object_value(matching[0]["model_info"])["service_tiers"] == configured, matching[0]
def test_tier_update_propagates(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
name: Final = _model_name("codex-updatable")
identity: Final = _new_model(scenario, name, model_info={"service_tiers": ["priority"]})
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value == ("priority",))
updated: Final = gateway.request(
"PATCH", f"/model/{identity}/update", {"model_info": {"service_tiers": ["flex", "ultrafast"]}}
)
assert updated.status_code == 200, updated.text
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value == ("flex", "ultrafast"))
def test_model_delete_drops_the_slug(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
name: Final = _model_name("codex-deletable")
identity: Final = _new_model(scenario, name, model_info={"service_tiers": ["priority"]})
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value == ("priority",))
gateway.post("/model/delete", {"id": identity})
_settled(lambda: _tiers_by_slug(gateway).get(name), lambda value: value is None)
def _sse(events: Sequence[object]) -> tuple[bytes, ...]:
return tuple(b"data: " + json.dumps(event).encode() + b"\n\n" for event in events) + (b"data: [DONE]\n\n",)
def _chat_reply(stream: bool) -> Reply:
identity: Final = "chatcmpl-" + uuid.uuid4().hex
if not stream:
return Reply(
body=json.dumps(
{
"id": identity,
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{"index": 0, "message": {"role": "assistant", "content": "tier probe"}, "finish_reason": "stop"}
],
"usage": USAGE,
}
).encode()
)
head: Final = {"id": identity, "object": "chat.completion.chunk", "created": 1, "model": "gpt-4o-mini"}
return Reply(
content_type="text/event-stream",
chunks=_sse(
(
{**head, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "tier "}}]},
{**head, "choices": [{"index": 0, "delta": {"content": "probe"}}]},
{**head, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
{**head, "choices": [], "usage": USAGE},
)
),
)
def _responses_reply() -> Reply:
identity: Final = uuid.uuid4().hex
return Reply(
body=json.dumps(
{
"id": "resp_" + identity,
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-4o-mini",
"output": [
{
"type": "message",
"id": "msg_" + identity,
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "tier probe", "annotations": []}],
}
],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
"usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15},
}
).encode()
)
def _upstream(request: Request) -> Reply:
if request.method == "GET" and request.target.endswith("/models"):
return Reply(body=json.dumps({"object": "list", "data": []}).encode())
if request.target.endswith("/responses"):
return _responses_reply()
return _chat_reply(json.loads(request.body).get("stream") is True)
def _v1(gateway: Gateway) -> str:
return str(gateway.client.base_url).rstrip("/") + "/v1"
def _llm_requests(wire: Wire) -> tuple[Request, ...]:
return tuple(request for request in wire.drain() if request.method == "POST")
def _request_bodies(requests: Sequence[Request]) -> list[dict[str, JsonValue]]:
return [JSON_OBJECT.validate_json(request.body) for request in requests]
def _texts(value: JsonValue) -> tuple[str, ...]:
if isinstance(value, str):
return (value,)
if isinstance(value, list):
return tuple(chain.from_iterable(_texts(item) for item in value))
if isinstance(value, dict):
return tuple(chain.from_iterable(_texts(value.get(key)) for key in ("content", "text", "input", "messages")))
return ()
def _prompt_of(body: Mapping[str, JsonValue]) -> str:
return " ".join(_texts(body.get("input", body.get("messages"))))
def _sent_prompt(body: Mapping[str, JsonValue], sent: Sequence[str]) -> str:
text: Final = _prompt_of(body)
matches: Final = [prompt for prompt in sent if prompt in text]
assert len(matches) == 1, (matches, text)
return matches[0]
def _spend_ids_for_key(key: str) -> list[str]:
rows: Final = read_rows(
'SELECT request_id FROM "LiteLLM_SpendLogs" WHERE api_key=%s', (sha256(key.encode()).hexdigest(),)
)
return [string_value(row["request_id"]) for row in rows]
def test_advertised_tier_reaches_the_upstream(gateway: Gateway) -> None:
with wire_server(_upstream) as wire, gateway.scenario() as scenario:
written_at: Final = time.monotonic()
model: Final = scenario.model(api_base=f"{wire.url}/v1", model_info={"service_tiers": ["priority"]})
key: Final = scenario.key(models=[model])
marker: Final = uuid.uuid4().hex
advertised: Final = _settled_on_every_worker(
lambda: _tiers_by_slug(gateway, key=key),
lambda tiers: tiers == {model: ("priority",)},
written_at=written_at,
)
assert advertised == {model: ("priority",)}, advertised
sync_client: Final = openai.OpenAI(
api_key=key, base_url=_v1(gateway), max_retries=0, http_client=httpx.Client(timeout=15, trust_env=False)
)
responses_id: Final = sync_client.responses.create(
model=model, input=f"{marker} responses probe", service_tier="priority"
).id
chat_id: Final = sync_client.chat.completions.create(
model=model, messages=[{"role": "user", "content": f"{marker} chat probe"}], service_tier="priority"
).id
async def streamed() -> str:
client: Final = openai.AsyncOpenAI(
api_key=key,
base_url=_v1(gateway),
max_retries=0,
http_client=httpx.AsyncClient(timeout=15, trust_env=False),
)
stream: Final = await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": f"{marker} stream probe"}],
service_tier="priority",
stream=True,
)
chunks: Final = [chunk async for chunk in stream]
assert chunks, "no chunks streamed"
return chunks[0].id
stream_id: Final = asyncio.run(streamed())
messages_id: Final = (
anthropic.Anthropic(
api_key=key,
base_url=str(gateway.client.base_url).rstrip("/"),
max_retries=0,
http_client=httpx.Client(timeout=15, trust_env=False),
)
.messages.create(
model=model, max_tokens=64, messages=[{"role": "user", "content": f"{marker} messages probe"}]
)
.id
)
received: Final = _llm_requests(wire)
assert len(received) == 4, [request.target for request in received]
targets: Final = sorted(request.target for request in received)
assert targets == ["/v1/chat/completions"] * 2 + ["/v1/responses"] * 2, targets
openai_shaped: Final = [
body for body in _request_bodies(received) if f"{marker} messages probe" not in _prompt_of(body)
]
assert len(openai_shaped) == 3, _request_bodies(received)
assert all(body["service_tier"] == "priority" for body in openai_shaped), openai_shaped
ids: Final = sorted((responses_id, chat_id, stream_id, messages_id))
assert len(set(ids)) == 4, ids
landed: Final = eventually(lambda: _spend_ids_for_key(key), lambda values: len(values) == 4, seconds=70)
assert sorted(landed) == ids, (landed, ids)
BURST_CATALOG: Final = 12
BURST_CHAT: Final = 8
BURST_STREAM: Final = 8
BURST_RESPONSES: Final = 4
BURST_MESSAGES: Final = 4
BURST_SIZES: Final = (
("chat", BURST_CHAT),
("stream", BURST_STREAM),
("responses", BURST_RESPONSES),
("messages", BURST_MESSAGES),
)
def _llm_tasks() -> tuple[tuple[str, int], ...]:
return tuple(
(kind, index) for kind, size in BURST_SIZES for index in range(size)
) # comprehension-ok: a flat task list of two small axes
def _stream_id(response: httpx.Response) -> str:
assert response.status_code == 200, response.text
events: Final = [
object_value(json.loads(line.removeprefix("data:")))
for line in response.iter_lines()
if line.startswith("data:") and line.removeprefix("data:").strip() != "[DONE]"
]
assert events, response.text
return string_value(events[0]["id"])
def _json_id(response: httpx.Response) -> str:
assert response.status_code == 200, response.text
return string_value(JSON_OBJECT.validate_json(response.content)["id"])
def test_catalog_burst_with_model_churn(gateway: Gateway) -> None:
with wire_server(_upstream) as wire, gateway.scenario() as scenario:
written_at: Final = time.monotonic()
model: Final = scenario.model(api_base=f"{wire.url}/v1", model_info={"service_tiers": ["priority"]})
key: Final = scenario.key(models=[model])
marker: Final = uuid.uuid4().hex
_settled_on_every_worker(
lambda: _tiers_by_slug(gateway, key=key),
lambda tiers: tiers == {model: ("priority",)},
written_at=written_at,
)
def prompt(kind: str, index: int) -> str:
return f"{marker} {kind} {index}"
def catalog(index: int) -> tuple[str, str]:
tiers: Final = _tiers_by_slug(gateway, key=key)
assert tiers == {model: ("priority",)}, (index, tiers)
return ("catalog", "")
def chat(index: int) -> tuple[str, str]:
body: Final = {"model": model, "messages": [{"role": "user", "content": prompt("chat", index)}]}
return ("chat", _json_id(gateway.request("POST", "/v1/chat/completions", body, key=key)))
def stream(index: int) -> tuple[str, str]:
body: Final = {
"model": model,
"messages": [{"role": "user", "content": prompt("stream", index)}],
"stream": True,
}
return ("stream", _stream_id(gateway.request("POST", "/v1/chat/completions", body, key=key)))
def responses(index: int) -> tuple[str, str]:
body: Final = {"model": model, "input": prompt("responses", index), "service_tier": "priority"}
return ("responses", _json_id(gateway.request("POST", "/v1/responses", body, key=key)))
def messages(index: int) -> tuple[str, str]:
body: Final = {
"model": model,
"max_tokens": 64,
"messages": [{"role": "user", "content": prompt("messages", index)}],
}
return ("messages", _json_id(gateway.request("POST", "/v1/messages", body, key=key)))
def churn(index: int) -> tuple[str, str]:
churned: Final = _new_model(
scenario, _model_name(f"codex-churn-{index}"), model_info={"service_tiers": ["flex"]}
)
gateway.post("/model/delete", {"id": churned})
return ("churn", churned)
tasks: Final[tuple[tuple[Callable[[int], tuple[str, str]], int], ...]] = (
*((catalog, index) for index in range(BURST_CATALOG)),
*((chat, index) for index in range(BURST_CHAT)),
*((stream, index) for index in range(BURST_STREAM)),
*((responses, index) for index in range(BURST_RESPONSES)),
*((messages, index) for index in range(BURST_MESSAGES)),
(churn, 0),
(churn, 1),
)
with ThreadPoolExecutor(max_workers=len(tasks)) as pool:
outcomes: Final = tuple(pool.map(lambda task: task[0](task[1]), tasks))
llm_kinds: Final = frozenset({"chat", "stream", "responses", "messages"})
llm_ids: Final = sorted(identity for kind, identity in outcomes if kind in llm_kinds)
assert len(llm_ids) == BURST_CHAT + BURST_STREAM + BURST_RESPONSES + BURST_MESSAGES, outcomes
assert len(set(llm_ids)) == len(llm_ids), llm_ids
received: Final = _request_bodies(_llm_requests(wire))
assert len(received) == len(llm_ids), len(received)
prompts_sent: Final = sorted(prompt(kind, index) for kind, index in _llm_tasks())
prompts_seen: Final = sorted(_sent_prompt(body, prompts_sent) for body in received)
assert prompts_seen == prompts_sent, (prompts_seen, prompts_sent)
landed: Final = eventually(
lambda: _spend_ids_for_key(key), lambda values: len(values) == len(llm_ids), seconds=70
)
assert sorted(landed) == llm_ids, (landed, llm_ids)
final_tiers: Final = _settled(
lambda: _tiers_by_slug(gateway, key=key), lambda value: value == {model: ("priority",)}
)
assert final_tiers == {model: ("priority",)}, final_tiers

View file

@ -0,0 +1,430 @@
import json
import os
import signal
import stat
import subprocess
import sys
import uuid
from collections.abc import Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Final
import httpx
import psutil
import pytest
from pydantic import JsonValue
from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value
from tests.integration._support.process import group_members, owned_proxy, owned_proxy_process
CLIENT_VERSION: Final = "0.159.3"
FRESH_CONNECTION: Final = {"Connection": "close"}
CONSISTENT_READS: Final = 8
CONVERGENCE_SECONDS: Final = 60
BURST: Final = 24
CATALOG_FILENAME: Final = "litellm-models.json"
SHIM_STOCK_ENV: Final = "CODEX_SHIM_STOCK"
SHIM_LAUNCH_ENV: Final = "CODEX_SHIM_LAUNCH"
SHIM_SCRIPT: Final = """\
import json
import os
import sys
argv = sys.argv[1:]
if argv[-2:] == ["debug", "models"]:
override = next((arg for arg in argv if arg.startswith("model_catalog_json=")), None)
if override is None:
sys.stdout.write(open(os.environ[%(stock)r], encoding="utf-8").read())
else:
catalog = json.load(open(json.loads(override.partition("=")[2]), encoding="utf-8"))
if not catalog.get("models"):
sys.stderr.write("empty model catalog\\n")
sys.exit(2)
sys.stdout.write(json.dumps(catalog))
sys.exit(0)
recorded = {name: os.environ.get(name) for name in ("OPENAI_BASE_URL", "OPENAI_API_KEY", "LITELLM_PROXY_API_KEY")}
with open(os.environ[%(launch)r], "w", encoding="utf-8") as handle:
json.dump({"argv": argv, "env": recorded}, handle)
""" % {"stock": SHIM_STOCK_ENV, "launch": SHIM_LAUNCH_ENV}
def _generic_tier(identity: str) -> dict[str, JsonValue]:
return {"id": identity, "name": identity.capitalize(), "description": f"Sends service_tier={identity} upstream"}
def _catalog_response(proxy: Gateway, *, key: str | None = None) -> httpx.Response:
return proxy.request(
"GET", "/v1/models", key=key, params={"client_version": CLIENT_VERSION}, headers=FRESH_CONNECTION
)
def _entries(response: httpx.Response) -> list[dict[str, JsonValue]]:
assert response.status_code == 200, response.text
body: Final = JSON_OBJECT.validate_json(response.content)
assert set(body) == {"models"}, response.text
models: Final = body["models"]
assert isinstance(models, list), response.text
return [object_value(entry) for entry in models]
def _by_slug(entries: Sequence[Mapping[str, JsonValue]]) -> dict[str, dict[str, JsonValue]]:
return {string_value(entry["slug"]): dict(entry) for entry in entries}
def _write_config(directory: Path, upstream_url: str, tiered: str, plain: str, alias: str) -> Path:
config: Final = directory / f"codex_catalog_{uuid.uuid4().hex}.yaml"
params: Final = {
"model": "openai/gpt-4o-mini",
"api_base": f"{upstream_url}/v1",
"api_key": "integration-provider-key",
}
config.write_text(
json.dumps(
{
"model_list": [
{
"model_name": tiered,
"litellm_params": params,
"model_info": {
"service_tiers": ["priority", {"id": "flex", "name": "Flex", "description": "cheap lane"}],
"display_name": "YAML Tiered",
"max_input_tokens": 4321,
},
},
{"model_name": plain, "litellm_params": params},
],
"router_settings": {"model_group_alias": {alias: tiered}},
"general_settings": {
"master_key": "os.environ/LITELLM_MASTER_KEY",
"database_url": "os.environ/DATABASE_URL",
},
}
)
)
return config
@pytest.mark.timeout(240)
def test_yaml_tiers_and_alias_in_owned_proxy(gateway: Gateway, tmp_path: Path) -> None:
tiered: Final = f"codex-yaml-tiered-{uuid.uuid4().hex}"
plain: Final = f"codex-yaml-plain-{uuid.uuid4().hex}"
alias: Final = f"codex-yaml-alias-{uuid.uuid4().hex}"
config: Final = _write_config(tmp_path, gateway.upstream_url, tiered, plain, alias)
with owned_proxy(gateway, tmp_path, {"STORE_MODEL_IN_DB": "False"}, config=config, workers=2) as candidate:
readings: Final = tuple(_entries(_catalog_response(candidate)) for _ in range(CONSISTENT_READS))
assert all(reading == readings[0] for reading in readings), readings
catalog: Final = _by_slug(readings[0])
assert sorted(catalog) == sorted((tiered, plain, alias)), sorted(catalog)
assert catalog[tiered]["service_tiers"] == [
_generic_tier("priority"),
{"id": "flex", "name": "Flex", "description": "cheap lane"},
], catalog[tiered]
assert catalog[tiered]["display_name"] == "YAML Tiered", catalog[tiered]
assert catalog[tiered]["context_window"] == 4321, catalog[tiered]
assert catalog[plain]["service_tiers"] == [] and catalog[plain]["display_name"] == plain, catalog[plain]
assert catalog[alias]["service_tiers"] == catalog[tiered]["service_tiers"], catalog[alias]
assert catalog[alias]["display_name"] == alias, catalog[alias]
assert catalog[alias]["context_window"] == 4321, catalog[alias]
assert string_value(catalog[alias]["base_instructions"]), alias
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"
STOCK_UPSTREAM: Final = "gpt-5.5"
def _bundled_stock_entry(slug: str) -> dict[str, JsonValue]:
models: Final = JSON_OBJECT.validate_json(BUNDLED_STOCK_PATH.read_bytes())["models"]
assert isinstance(models, list), slug
(entry,) = (object_value(model) for model in models if object_value(model)["slug"] == slug)
return entry
def _write_team_owned_config(
directory: Path, upstream_url: str, stock_team: str, plain_team: str, owned: str, mixed: str, alias: str
) -> Path:
config: Final = directory / f"codex_catalog_teams_{uuid.uuid4().hex}.yaml"
plain_params: Final = {
"model": "openai/gpt-4o-mini",
"api_base": f"{upstream_url}/v1",
"api_key": "integration-provider-key",
}
stock_params: Final = {**plain_params, "model": f"openai/{STOCK_UPSTREAM}"}
config.write_text(
json.dumps(
{
"model_list": [
{"model_name": owned, "litellm_params": stock_params, "model_info": {"team_id": stock_team}},
{
"model_name": owned,
"litellm_params": plain_params,
"model_info": {"team_id": plain_team, "service_tiers": ["flex"]},
},
{"model_name": mixed, "litellm_params": stock_params, "model_info": {"team_id": stock_team}},
{"model_name": mixed, "litellm_params": plain_params, "model_info": {"service_tiers": ["flex"]}},
],
"router_settings": {"model_group_alias": {alias: owned}},
"general_settings": {
"master_key": "os.environ/LITELLM_MASTER_KEY",
"database_url": "os.environ/DATABASE_URL",
},
}
)
)
return config
@pytest.mark.timeout(240)
def test_team_owned_deployments_of_one_name_serve_each_team_its_own_upstream(gateway: Gateway, tmp_path: Path) -> None:
"""Two teams own a deployment of one model name (and of a `model_group_alias` of it): Codex's stock
entry for the first deployment's upstream goes only to the team whose requests reach it, and a
caller outside the owning team reads the deployment no team owns."""
owned: Final = f"codex-team-owned-{uuid.uuid4().hex}"
mixed: Final = f"codex-team-mixed-{uuid.uuid4().hex}"
alias: Final = f"codex-team-alias-{uuid.uuid4().hex}"
stock: Final = _bundled_stock_entry(STOCK_UPSTREAM)
assert stock["supported_reasoning_levels"] != [] and stock["service_tiers"] != [], stock
with gateway.scenario() as scenario:
stock_team: Final = scenario.team()
plain_team: Final = scenario.team()
stock_key: Final = scenario.key(team_id=stock_team, models=[owned, alias, mixed])
plain_key: Final = scenario.key(team_id=plain_team, models=[owned, alias, mixed])
teamless_key: Final = scenario.key(models=[mixed])
config: Final = _write_team_owned_config(
tmp_path, gateway.upstream_url, stock_team, plain_team, owned, mixed, alias
)
with owned_proxy(gateway, tmp_path, {"STORE_MODEL_IN_DB": "False"}, config=config, workers=2) as candidate:
def stable_catalog(key: str) -> dict[str, dict[str, JsonValue]]:
readings: Final = tuple(
_entries(_catalog_response(candidate, key=key)) for _ in range(CONSISTENT_READS)
)
assert all(reading == readings[0] for reading in readings), readings
return _by_slug(readings[0])
for_stock_team: Final = stable_catalog(stock_key)
for_plain_team: Final = stable_catalog(plain_key)
for_teamless: Final = stable_catalog(teamless_key)
assert sorted(for_stock_team) == sorted(for_plain_team) == sorted((owned, alias, mixed)), sorted(for_stock_team)
assert sorted(for_teamless) == [mixed], sorted(for_teamless)
for slug in (owned, alias):
assert for_stock_team[slug]["supported_reasoning_levels"] == stock["supported_reasoning_levels"], slug
assert for_stock_team[slug]["model_messages"] == stock["model_messages"], slug
assert for_stock_team[slug]["service_tiers"] == stock["service_tiers"], slug
assert for_plain_team[slug]["supported_reasoning_levels"] == [], for_plain_team[slug]
assert "model_messages" not in for_plain_team[slug], slug
assert string_value(for_plain_team[slug]["base_instructions"]), slug
assert for_plain_team[slug]["service_tiers"] == [_generic_tier("flex")], for_plain_team[slug]
assert for_stock_team[mixed]["supported_reasoning_levels"] == stock["supported_reasoning_levels"], mixed
assert for_stock_team[mixed]["service_tiers"] == [], for_stock_team[mixed]
for outsider in (for_plain_team, for_teamless):
assert outsider[mixed]["supported_reasoning_levels"] == [], outsider[mixed]
assert "model_messages" not in outsider[mixed], mixed
assert outsider[mixed]["service_tiers"] == [_generic_tier("flex")], outsider[mixed]
def _worker_pids(root_pid: int) -> frozenset[int]:
def is_worker(process: psutil.Process) -> bool:
try:
return "spawn_main" in " ".join(process.cmdline())
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
return False
return frozenset(process.pid for process in group_members(root_pid) if is_worker(process))
def _catalog_outcome(proxy: Gateway, slugs: Sequence[JsonValue]) -> str:
try:
response: Final = _catalog_response(proxy)
except httpx.TransportError as error:
return f"transport:{type(error).__name__}"
assert response.status_code == 200, response.text
assert sorted(entry["slug"] for entry in _entries(response)) == list(slugs), response.text
return "ok"
@pytest.mark.timeout(300)
def test_catalog_survives_a_worker_kill(gateway: Gateway, tmp_path: Path) -> None:
tiered: Final = f"codex-yaml-tiered-{uuid.uuid4().hex}"
plain: Final = f"codex-yaml-plain-{uuid.uuid4().hex}"
alias: Final = f"codex-yaml-alias-{uuid.uuid4().hex}"
config: Final = _write_config(tmp_path, gateway.upstream_url, tiered, plain, alias)
with owned_proxy_process(gateway, tmp_path, {"STORE_MODEL_IN_DB": "False"}, config=config, workers=2) as owned:
candidate: Final = owned.gateway
workers: Final = eventually(lambda: _worker_pids(owned.process.pid), lambda pids: len(pids) == 2, seconds=30)
victim: Final = min(workers)
slugs: Final = sorted(entry["slug"] for entry in _entries(_catalog_response(candidate)))
assert slugs == sorted((tiered, plain, alias)), slugs
def attempt(index: int) -> str:
if index == 2:
os.kill(victim, signal.SIGKILL)
return _catalog_outcome(candidate, slugs)
with ThreadPoolExecutor(max_workers=BURST) as pool:
outcomes: Final = tuple(pool.map(attempt, range(BURST)))
assert outcomes.count("ok") >= 1, outcomes
assert all(outcome == "ok" or outcome.startswith("transport:") for outcome in outcomes), outcomes
respawned: Final = eventually(
lambda: _worker_pids(owned.process.pid),
lambda pids: len(pids) == 2 and victim not in pids,
seconds=60,
)
assert victim not in respawned, respawned
def settled_burst() -> tuple[str, ...]:
with ThreadPoolExecutor(max_workers=BURST) as pool:
return tuple(pool.map(lambda _: _catalog_outcome(candidate, slugs), range(BURST)))
final: Final = eventually(settled_burst, lambda values: all(value == "ok" for value in values), seconds=40)
assert final == ("ok",) * BURST, final
def _install_shim(directory: Path, stock: Mapping[str, JsonValue]) -> tuple[Path, dict[str, str]]:
shim_dir: Final = directory / "bin"
shim_dir.mkdir()
script: Final = directory / "codex_shim.py"
script.write_text(SHIM_SCRIPT, encoding="utf-8")
shim: Final = shim_dir / "codex"
shim.write_text(f'#!/bin/sh\nexec "{sys.executable}" "{script}" "$@"\n', encoding="utf-8")
shim.chmod(shim.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
stock_path: Final = directory / "stock.json"
stock_path.write_text(json.dumps(stock), encoding="utf-8")
launch_path: Final = directory / "launch.json"
home: Final = directory / "home"
home.mkdir()
codex_home: Final = directory / "codex-home"
environment: Final = {
**os.environ,
"HOME": str(home),
"CODEX_HOME": str(codex_home),
"PATH": f"{shim_dir}{os.pathsep}{os.environ['PATH']}",
SHIM_STOCK_ENV: str(stock_path),
SHIM_LAUNCH_ENV: str(launch_path),
}
return launch_path, environment
def _run_lite_codex(proxy: Gateway, key: str, environment: Mapping[str, str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, "-I", "-c", "from litellm.proxy.client.cli import cli; cli()", "codex", "--shim-marker"],
env={**environment, "LITELLM_PROXY_URL": str(proxy.client.base_url).rstrip("/"), "LITELLM_PROXY_API_KEY": key},
stdin=subprocess.DEVNULL,
capture_output=True,
text=True,
timeout=90,
check=False,
)
def _new_model(scenario: Scenario, name: str, model_info: Mapping[str, JsonValue]) -> None:
created: Final = scenario.gateway.post(
"/model/new",
{
"model_name": name,
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "integration-provider-key",
"api_base": f"{scenario.gateway.upstream_url}/v1",
},
"model_info": dict(model_info),
},
)
scenario.cleanups.callback(scenario.delete_model, string_value(object_value(created["model_info"])["id"]))
def _launch_record(path: Path) -> dict[str, JsonValue]:
assert path.exists(), "the shim was never launched"
return JSON_OBJECT.validate_json(path.read_bytes())
def _catalog_override(argv: Sequence[JsonValue]) -> str | None:
overrides: Final = [
string_value(arg) for arg in argv if isinstance(arg, str) and arg.startswith("model_catalog_json=")
]
assert len(overrides) <= 1, argv
return json.loads(overrides[0].partition("=")[2]) if overrides else None
def test_lite_codex_writes_the_shared_catalog(gateway: Gateway, tmp_path: Path) -> None:
"""`lite codex` writes the installed Codex's own entry for a proxy model whose name is a stock slug and
the fallback entry for any other; the proxy's configured tiers never reach the file (Decision 17)."""
stock_slug: Final = f"codex-stock-{uuid.uuid4().hex}"
stock_tiers: Final = [{"id": "priority", "name": "Fast", "description": "1.5x speed"}]
stock: Final = {
"models": [
{
"slug": stock_slug,
"display_name": "Stock Model",
"priority": 7,
"visibility": "hide",
"supported_in_api": False,
"service_tiers": stock_tiers,
"default_service_tier": None,
"supported_reasoning_levels": [{"effort": "high", "description": "thinks longer"}],
}
]
}
launch_path, environment = _install_shim(tmp_path, stock)
with gateway.scenario() as scenario:
_new_model(scenario, stock_slug, {"service_tiers": ["ultrafast"]})
fallback: Final = scenario.model(model_info={"service_tiers": ["priority"]})
key: Final = scenario.key(models=[stock_slug, fallback])
completed: Final = _run_lite_codex(gateway, key, environment)
assert completed.returncode == 0, (completed.stdout, completed.stderr)
assert "not syncing" not in completed.stderr, completed.stderr
record: Final = _launch_record(launch_path)
argv: Final = record["argv"]
assert isinstance(argv, list) and argv[-1] == "--shim-marker", argv
catalog_path: Final = Path(environment["CODEX_HOME"]) / CATALOG_FILENAME
assert _catalog_override(argv) == str(catalog_path), argv
env: Final = object_value(record["env"])
assert env["OPENAI_BASE_URL"] == str(gateway.client.base_url).rstrip("/") + "/v1", env
assert env["OPENAI_API_KEY"] == key, env.keys()
written: Final = JSON_OBJECT.validate_json(catalog_path.read_bytes())
assert set(written) == {"models"}, written.keys()
models: Final = written["models"]
assert isinstance(models, list), written
file_entries: Final = _by_slug([object_value(entry) for entry in models])
served: Final = eventually(
lambda: _by_slug(_entries(_catalog_response(gateway, key=key))),
lambda value: (
stock_slug in value
and fallback in value
and value[stock_slug]["service_tiers"] == [_generic_tier("ultrafast")]
and value[fallback]["service_tiers"] == [_generic_tier("priority")]
),
seconds=CONVERGENCE_SECONDS,
)
assert list(file_entries) == list(served), (list(file_entries), list(served))
assert file_entries[stock_slug]["service_tiers"] == stock_tiers, file_entries[stock_slug]
assert file_entries[stock_slug]["visibility"] == "list", file_entries[stock_slug]
assert file_entries[stock_slug]["supported_in_api"] is True, file_entries[stock_slug]
assert file_entries[stock_slug]["priority"] == list(file_entries).index(stock_slug), file_entries[stock_slug]
assert file_entries[stock_slug]["supported_reasoning_levels"] == [
{"effort": "high", "description": "thinks longer"}
]
assert file_entries[fallback]["service_tiers"] == [], file_entries[fallback]
assert file_entries[fallback]["display_name"] == fallback, file_entries[fallback]
assert string_value(file_entries[fallback]["base_instructions"]), fallback
assert served[stock_slug]["service_tiers"] == [_generic_tier("ultrafast")], served[stock_slug]
assert served[fallback]["service_tiers"] == [_generic_tier("priority")], served[fallback]
def test_lite_codex_skips_a_stock_dump_without_display_name(gateway: Gateway, tmp_path: Path) -> None:
"""A `codex debug models` dump that predates `display_name` is reported, not written, and Codex still
launches without a catalog override."""
stock: Final = {"models": [{"slug": "gpt-5.5", "priority": 0, "visibility": "list"}]}
launch_path, environment = _install_shim(tmp_path, stock)
with gateway.scenario() as scenario:
model: Final = scenario.model(model_info={"service_tiers": ["priority"]})
key: Final = scenario.key(models=[model])
completed: Final = _run_lite_codex(gateway, key, environment)
assert completed.returncode == 0, (completed.stdout, completed.stderr)
assert "printed no model catalog" in completed.stderr, completed.stderr
record: Final = _launch_record(launch_path)
argv: Final = record["argv"]
assert isinstance(argv, list) and argv[-1] == "--shim-marker", argv
assert _catalog_override(argv) is None, argv
assert not (Path(environment["CODEX_HOME"]) / CATALOG_FILENAME).exists(), environment["CODEX_HOME"]

View file

@ -0,0 +1,535 @@
import json
from types import MappingProxyType
import pytest
import litellm
from litellm.proxy.common_utils.codex_model_catalog import (
CODEX_CATALOG_BYTE_LIMIT,
CodexCatalogRow,
CodexServiceTier,
CodexStockModel,
CodexStockUpgrade,
bundled_codex_models,
codex_catalog_rows,
codex_model_list_body,
codex_models_response_json,
configured_service_tiers,
)
_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):
return CodexStockModel(
slug=slug,
display_name=slug.upper(),
priority=99,
visibility=visibility,
supported_in_api=supported_in_api,
upgrade=CodexStockUpgrade(model=upgrade) if upgrade else None,
service_tiers=tuple(CodexServiceTier(**tier) for tier in service_tiers),
default_service_tier=default_tier,
model_messages={"instructions_template": f"{slug} prompt"},
)
_STOCK = MappingProxyType(
{
model.slug: model
for model in (
_stock("gpt-5.5", upgrade="gpt-6-sol"),
_stock("gpt-6-sol", default_tier="priority"),
_stock("gpt-daybreak", visibility="hide", supported_in_api=False),
)
}
)
def _row(model_id, **overrides):
"""A one-deployment row; `service_tiers` is that deployment's raw value, `deployments` a per-deployment tuple."""
per_deployment = (
{"service_tiers": (overrides["service_tiers"],)}
if "service_tiers" in overrides
else {"service_tiers": overrides["deployments"]}
if "deployments" in overrides
else {}
)
listing = {key: value for key, value in overrides.items() if key not in ("service_tiers", "deployments")}
return CodexCatalogRow(
**{"id": model_id, "mode": "responses", "max_input_tokens": 272000, **listing, **per_deployment}
)
def _body(*rows, **kwargs):
return codex_models_response_json(rows, stock=_STOCK, instructions=_PROMPT, **kwargs)
def _models(*rows, **kwargs):
return json.loads(_body(*rows, **kwargs).json)["models"]
def _tiers(entry):
return [(tier["id"], tier["name"], tier["description"]) for tier in entry["service_tiers"]]
def test_body_is_exactly_codex_models_response():
body = json.loads(_body(_row("gpt-6-astra")).json)
assert list(body) == ["models"]
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`."""
(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",
"list",
True,
)
assert (entry["context_window"], entry["base_instructions"], entry["service_tiers"]) == (272000, _PROMPT, [])
def test_known_model_keeps_codex_stock_entry_listed_under_the_served_id():
(entry,) = _models(_row("gpt-daybreak"))
assert entry["model_messages"] == {"instructions_template": "gpt-daybreak prompt"}
assert "base_instructions" not in entry
assert (entry["slug"], entry["display_name"], entry["visibility"], entry["supported_in_api"]) == (
"gpt-daybreak",
"GPT-DAYBREAK",
"list",
True,
)
assert _tiers(entry) == [("priority", "Fast", "2x speed, increased usage")]
@pytest.mark.parametrize("upstream", ["gpt-5.5", "openai/gpt-5.5"])
def test_alias_of_a_known_upstream_model_takes_stock_metadata_under_its_own_name(upstream):
(entry,) = _models(_row("team-55", upstream_model=upstream))
assert entry["model_messages"] == {"instructions_template": "gpt-5.5 prompt"}
assert (entry["slug"], entry["display_name"]) == ("team-55", "team-55")
@pytest.mark.parametrize(
("row", "expected"),
[
(_row("gpt-5.5", display_name="Our 5.5"), "Our 5.5"),
(_row("gpt-5.5"), "GPT-5.5"),
(_row("fast-55", upstream_model="gpt-5.5"), "fast-55"),
(_row("mystery", display_name="Mystery"), "Mystery"),
(_row("mystery"), "mystery"),
],
)
def test_display_name_is_configured_then_stock_for_its_own_slug_then_the_id(row, expected):
(entry,) = _models(row)
assert entry["display_name"] == expected
def test_priority_is_listing_order():
entries = _models(_row("gpt-6-sol"), _row("mystery"), _row("gpt-5.5"))
assert [(entry["slug"], entry["priority"]) for entry in entries] == [
("gpt-6-sol", 0),
("mystery", 1),
("gpt-5.5", 2),
]
def test_upgrade_nudge_survives_only_when_its_target_is_served():
with_target, without_target = (
_models(_row("gpt-5.5"), _row("gpt-6-sol"))[0],
_models(_row("gpt-5.5"))[0],
)
assert with_target["upgrade"]["model"] == "gpt-6-sol"
assert without_target["upgrade"] is None
@pytest.mark.parametrize(
("row", "expected"),
[
(
_row("gpt-6-astra", service_tiers=["ultrafast"]),
[("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream")],
),
(
_row("gpt-6-astra", service_tiers=[{"id": "ultrafast", "name": "Ultra fast", "description": "Fastest"}]),
[("ultrafast", "Ultra fast", "Fastest")],
),
(
_row("gpt-6-astra", service_tiers=[{"id": "ultrafast"}]),
[("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream")],
),
(
_row("gpt-5.5", service_tiers=["ultrafast"]),
[("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream")],
),
(
_row("gpt-5.5", service_tiers=["priority", "ultrafast"]),
[
("priority", "Fast", "2x speed, increased usage"),
("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream"),
],
),
(_row("gpt-5.5", service_tiers=[]), []),
(
_row("gpt-5.5", service_tiers=["ultrafast", "priority", "ultrafast", {"id": "priority"}]),
[
("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream"),
("priority", "Fast", "2x speed, increased usage"),
],
),
],
)
def test_configured_service_tiers_replace_stock_tiers_and_a_known_id_keeps_its_codex_name(row, expected):
(entry,) = _models(row)
assert _tiers(entry) == expected
@pytest.mark.parametrize(
"invalid", ["ultrafast", [""], [" "], [1], [{"name": "no id"}], [{"id": "x", "bogus": 1}], {"id": "x"}]
)
def test_invalid_service_tiers_offer_no_tier_even_on_a_model_codex_ships_tiers_for(invalid):
"""Keeping the stock tiers would offer Codex's `/fast` on a model whose operator declared something else."""
stock_entry, fallback_entry = _models(
_row("gpt-5.5", service_tiers=invalid), _row("mystery", service_tiers=invalid)
)
assert _tiers(stock_entry) == []
assert fallback_entry["service_tiers"] == []
@pytest.mark.parametrize(
("deployments", "expected"),
[
((["ultrafast"], ["ultrafast"]), [("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream")]),
(
(["priority", "ultrafast"], ["ultrafast", "flex", "priority"]),
[
("priority", "Fast", "2x speed, increased usage"),
("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream"),
],
),
(
(["priority", "ultrafast"], ["ultrafast"]),
[("ultrafast", "Ultrafast", "Sends service_tier=ultrafast upstream")],
),
((["ultrafast"], ["priority"]), []),
((["ultrafast"], None), []),
((None, ["ultrafast"], None), []),
],
)
def test_a_tier_is_offered_only_when_every_deployment_of_the_model_lists_it(deployments, expected):
"""Any deployment can serve the request, so the first deployment's order is kept but a tier one of them
lacks, or a deployment that lists none, offers nothing Codex could send to the wrong deployment."""
(entry,) = _models(_row("gpt-5.5", deployments=deployments))
assert _tiers(entry) == expected
def test_deployments_that_all_leave_service_tiers_unset_keep_the_stock_tiers():
stock_entry, fallback_entry = _models(
_row("gpt-5.5", deployments=(None, None)), _row("mystery", deployments=(None, None))
)
assert _tiers(stock_entry) == [("priority", "Fast", "2x speed, increased usage")]
assert fallback_entry["service_tiers"] == []
@pytest.mark.parametrize(
"deployments", [(["ultrafast"], "not-a-list"), ("not-a-list", ["ultrafast"]), (["priority"], "not-a-list")]
)
def test_an_invalid_value_on_one_deployment_offers_no_tier_for_the_whole_model(deployments):
"""The deployment with the typo lists nothing Codex could send, and the stock `/fast` stays off too."""
(entry,) = _models(_row("gpt-5.5", deployments=deployments))
assert _tiers(entry) == []
@pytest.mark.parametrize(
("configured", "expected"),
[(None, "priority"), (["priority", "ultrafast"], "priority"), (["ultrafast"], None), ([], None)],
)
def test_default_service_tier_survives_only_while_still_offered(configured, expected):
(entry,) = _models(_row("gpt-6-sol", service_tiers=configured))
assert entry["default_service_tier"] == expected
def test_non_chat_and_wildcard_rows_are_left_out_of_the_catalog():
body = _body(
_row("gpt-6-astra"),
CodexCatalogRow(id="text-embedding-3-small", mode="embedding"),
CodexCatalogRow(id="openai/*"),
CodexCatalogRow(id="no-mode-model"),
_row("gpt-5.5", mode="chat"),
)
assert body.listed == ("gpt-6-astra", "no-mode-model", "gpt-5.5")
assert body.left_out == ()
def test_body_leaves_out_the_entry_that_would_cross_the_byte_limit():
rows = tuple(_row(f"model-{index}") for index in range(6))
unlimited = _body(*rows, byte_limit=None)
limit = len(unlimited.json.encode()) - 1
body = _body(*rows, byte_limit=limit)
one_more = _body(*rows[: len(body.listed) + 1], byte_limit=None)
assert len(body.json.encode()) <= limit < len(one_more.json.encode())
assert body.listed + body.left_out == tuple(row.id for row in rows)
assert body.left_out == ("model-5",)
assert [entry["slug"] for entry in json.loads(body.json)["models"]] == list(body.listed)
def test_entries_offering_a_tier_survive_the_byte_limit_ahead_of_the_rest_and_keep_their_listing_place():
rows = (
_row("plain-first"),
_row("gpt-5.5"),
_row("plain-middle"),
_row("tiered-last", service_tiers=["ultrafast"]),
)
unlimited = _models(*rows, byte_limit=None)
two_entries_only = len(_body(rows[1], rows[3], byte_limit=None).json.encode())
body = _body(*rows, byte_limit=two_entries_only)
models = json.loads(body.json)["models"]
assert body.listed == ("gpt-5.5", "tiered-last") and body.left_out == ("plain-first", "plain-middle")
assert [entry["slug"] for entry in models] == ["gpt-5.5", "tiered-last"]
assert [entry["priority"] for entry in models] == [1, 3]
assert models == [unlimited[1], unlimited[3]]
assert len(body.json.encode()) <= two_entries_only
def test_an_entry_too_large_for_the_bytes_left_is_passed_over_and_the_smaller_ones_after_it_are_kept():
"""A tier description longer than the whole byte limit sorts first (tiered) and can never fit,
and a long display name fits an empty body but not what is left after the entries ahead of it."""
kept_rows = (
_row("plain-first"),
_row("tiered-small", service_tiers=["ultrafast"]),
_row("plain-last"),
)
limit = len(_body(*kept_rows, byte_limit=None).json.encode())
oversized_tier = {"id": "huge", "name": "Huge", "description": "x" * limit}
rows = (
kept_rows[0],
_row("tiered-oversized", service_tiers=[oversized_tier]),
kept_rows[1],
_row("plain-wide", display_name="w" * 200),
kept_rows[2],
)
unlimited = _models(*rows, byte_limit=None)
assert len(_body(rows[3], byte_limit=None).json.encode()) < limit
body = _body(*rows, byte_limit=limit)
models = json.loads(body.json)["models"]
assert body.listed == ("plain-first", "tiered-small", "plain-last")
assert body.left_out == ("tiered-oversized", "plain-wide")
assert models == [unlimited[0], unlimited[2], unlimited[4]]
assert [entry["priority"] for entry in models] == [0, 2, 4]
assert len(body.json.encode()) <= limit
def test_unlimited_body_keeps_every_entry():
rows = tuple(_row(f"model-{index}") for index in range(3))
body = _body(*rows, byte_limit=None)
assert (body.listed, body.left_out) == (tuple(row.id for row in rows), ())
assert CODEX_CATALOG_BYTE_LIMIT == 1024 * 1024
def test_vendored_codex_catalog_entries_decode_on_codex():
"""Every stock entry served keeps the `model_messages` Codex's `ModelsResponse` decoder
needs in place of `base_instructions`, and the catalog is read from the vendored file."""
stock = bundled_codex_models()
rows = tuple(CodexCatalogRow(id=slug, mode="responses") for slug in stock)
body = codex_models_response_json(rows, instructions=_PROMPT, byte_limit=None)
entries = json.loads(body.json)["models"]
assert len(entries) == len(stock) > 0
assert all(entry["model_messages"]["instructions_template"] for entry in entries)
assert all(entry["visibility"] == "list" and entry["supported_in_api"] is True for entry in entries)
assert all("base_instructions" not in entry for entry in entries)
def test_configured_service_tiers_without_known_tiers_names_strings_after_their_id():
assert configured_service_tiers((["ultrafast"],), "m") == (
CodexServiceTier(id="ultrafast", name="Ultrafast", description="Sends service_tier=ultrafast upstream"),
)
assert configured_service_tiers((None,), "m") is None
assert configured_service_tiers((), "m") is None
def test_catalog_rows_read_the_router_by_each_entry_lookup_id():
router = litellm.Router(
model_list=[
{
"model_name": "model_name_team-1_c0ffee",
"litellm_params": {"model": "openai/gpt-5.5"},
"model_info": {"display_name": "Team 5.5", "service_tiers": ["ultrafast"]},
},
{
"model_name": "model_name_team-1_c0ffee",
"litellm_params": {"model": "openai/gpt-5.5", "api_base": "https://second.example"},
},
{"model_name": "plain", "litellm_params": {"model": "openai/some-unmapped-model"}},
]
)
listing = (
{
"id": "gpt-5.5-team",
"object": "model",
"created": 0,
"owned_by": "openai",
"mode": "chat",
"max_input_tokens": 7,
},
{"id": "plain", "object": "model", "created": 0, "owned_by": "openai"},
)
rows = codex_catalog_rows(listing, (("gpt-5.5-team", "model_name_team-1_c0ffee"), ("plain", "plain")), router)
assert rows == (
CodexCatalogRow(
id="gpt-5.5-team",
mode="chat",
max_input_tokens=7,
upstream_model="openai/gpt-5.5",
display_name="Team 5.5",
service_tiers=(["ultrafast"], None),
),
CodexCatalogRow(id="plain", upstream_model="openai/some-unmapped-model", service_tiers=(None,)),
)
def test_catalog_rows_read_an_alias_off_its_target_under_the_alias_name():
router = litellm.Router(
model_list=[
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra"},
"model_info": {"display_name": "Astra", "service_tiers": ["ultrafast"]},
}
],
model_group_alias={"gpt-6": "gpt-6-astra"},
)
listing = (
{"id": "gpt-6", "object": "model", "created": 0, "owned_by": "openai", "mode": "chat", "max_input_tokens": 9},
)
(row,) = codex_catalog_rows(listing, (("gpt-6", "gpt-6"),), router)
assert row == CodexCatalogRow(
id="gpt-6", mode="chat", max_input_tokens=9, upstream_model="openai/gpt-6-astra", service_tiers=(["ultrafast"],)
)
(entry,) = json.loads(codex_model_list_body(listing, (("gpt-6", "gpt-6"),), router))["models"]
assert (entry["slug"], entry["display_name"], [tier["id"] for tier in entry["service_tiers"]]) == (
"gpt-6",
"gpt-6",
["ultrafast"],
)
assert entry["model_messages"] == bundled_codex_models()["gpt-6-astra"].model_dump(mode="json")["model_messages"]
def test_model_list_body_offers_a_tier_only_off_the_deployments_the_key_team_can_route_to():
router = litellm.Router(
model_list=[
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra"},
"model_info": {"service_tiers": ["ultrafast"]},
},
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra", "api_base": "https://team-2.example"},
"model_info": {"team_id": "team-2"},
},
]
)
listing = ({"id": "gpt-6-astra", "object": "model", "created": 0, "owned_by": "openai", "mode": "chat"},)
entries = (("gpt-6-astra", "gpt-6-astra"),)
def offered(team_id):
(entry,) = json.loads(codex_model_list_body(listing, entries, router, team_id))["models"]
return [tier["id"] for tier in entry["service_tiers"]]
assert offered("team-1") == ["ultrafast"]
assert offered("team-2") == []
assert offered(None) == ["ultrafast"]
def test_model_list_body_takes_stock_metadata_off_the_deployment_the_key_team_routes_to():
"""Two teams own a deployment of one name: Codex's stock gpt-5.5 entry goes only to the team whose
requests reach gpt-5.5, under the name and under a `model_group_alias` of it."""
router = litellm.Router(
model_list=[
{
"model_name": "coding-model",
"litellm_params": {"model": "openai/gpt-5.5"},
"model_info": {"team_id": "team-1"},
},
{
"model_name": "coding-model",
"litellm_params": {"model": "openai/some-unmapped-model"},
"model_info": {"team_id": "team-2", "service_tiers": ["flex"]},
},
],
model_group_alias={"coding": "coding-model"},
)
listing = tuple(
{"id": name, "object": "model", "created": 0, "owned_by": "openai", "mode": "chat"}
for name in ("coding-model", "coding")
)
entries = (("coding-model", "coding-model"), ("coding", "coding"))
stock = bundled_codex_models()["gpt-5.5"].model_dump(mode="json")
def served(team_id):
return json.loads(codex_model_list_body(listing, entries, router, team_id))["models"]
for entry in served("team-1"):
assert entry["model_messages"] == stock["model_messages"]
assert entry["supported_reasoning_levels"] == stock["supported_reasoning_levels"] != []
assert entry["service_tiers"] == stock["service_tiers"] != []
for entry in served("team-2"):
assert "model_messages" not in entry and entry["base_instructions"]
assert entry["supported_reasoning_levels"] == []
assert [tier["id"] for tier in entry["service_tiers"]] == ["flex"]
assert [entry["slug"] for entry in served("team-2")] == ["coding-model", "coding"]
def test_catalog_rows_without_a_router_carry_only_the_listing():
listing = ({"id": "plain", "object": "model", "created": 0, "owned_by": "openai", "mode": "chat"},)
assert codex_catalog_rows(listing, (("plain", "plain"),), None) == (CodexCatalogRow(id="plain", mode="chat"),)

View file

@ -46,7 +46,9 @@ def patched_models(monkeypatch):
deployment = MagicMock()
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_configured_display_name = MagicMock(return_value=None)
router.get_configured_service_tiers = MagicMock(return_value=())
monkeypatch.setattr(proxy_server, "llm_router", router)
monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
@ -397,3 +399,118 @@ def test_anthropic_format_keeps_served_ids_for_other_anthropic_clients(client, a
assert response.status_code == 200
assert [m["id"] for m in response.json()["data"]] == ["gpt-4", "claude-sonnet"]
@pytest.mark.parametrize("path", ["/v1/models", "/models"])
@pytest.mark.parametrize("params", [{}, {"scope": "expand"}])
def test_codex_format_when_client_version_present(client, auth_as, patched_models, path, params):
"""Codex CLI fetches a provider's catalog as ``GET /v1/models?client_version=<its version>`` and
decodes Codex's own ``{"models": [...]}`` shape; the same request without the parameter keeps the
OpenAI shape byte for byte."""
with auth_as():
codex_response = client.get(path, params={**params, "client_version": "0.159.3"})
openai_response = client.get(path, params=params)
assert codex_response.status_code == 200
assert codex_response.headers["content-type"] == "application/json"
body = codex_response.json()
assert list(body) == ["models"]
assert [(m["slug"], m["display_name"], m["priority"]) for m in body["models"]] == [
("gpt-4", "gpt-4", 0),
("claude-sonnet", "claude-sonnet", 1),
]
assert all(m["base_instructions"] and m["visibility"] == "list" for m in body["models"])
assert openai_response.status_code == 200
assert normalize(openai_response.json()) == {
"data": [
{"id": "<VOLATILE>", "object": "model", "created": "<VOLATILE>", "owned_by": "openai"},
{"id": "<VOLATILE>", "object": "model", "created": "<VOLATILE>", "owned_by": "openai"},
],
"object": "list",
}
@pytest.mark.parametrize("path", ["/v1/models", "/models"])
def test_codex_format_wins_over_the_anthropic_header(client, auth_as, patched_models, path):
with auth_as():
response = client.get(path, params={"client_version": "0.159.3"}, headers={"anthropic-version": "2023-06-01"})
assert response.status_code == 200
assert list(response.json()) == ["models"]
@pytest.mark.parametrize("path", ["/v1/models", "/models"])
def test_codex_format_carries_configured_service_tiers(client, auth_as, patched_models, path):
"""A deployment's ``model_info.service_tiers`` becomes the entry's ``service_tiers``, which Codex
offers as slash commands; a model without one offers none, and the OpenAI shape gains no field."""
patched_models.get_configured_service_tiers = MagicMock(
side_effect=lambda model_name, team_id=None: (["ultrafast"],) if model_name == "gpt-4" else (None,)
)
with auth_as():
codex_response = client.get(path, params={"client_version": "0.159.3"})
openai_response = client.get(path)
gpt_4, claude = codex_response.json()["models"]
assert gpt_4["service_tiers"] == [
{"id": "ultrafast", "name": "Ultrafast", "description": "Sends service_tier=ultrafast upstream"}
]
assert claude["service_tiers"] == []
assert all("service_tiers" not in m for m in openai_response.json()["data"])
@pytest.mark.parametrize("params", [{}, {"scope": "expand"}])
def test_codex_service_tiers_are_read_for_the_key_team(client, auth_as, patched_models, params):
"""A tier and the upstream model that picks Codex's stock entry are read off the deployments the
key's team can route to, so both listing paths hand the router the key's team, and no team for a
key without one."""
patched_models.get_configured_service_tiers = MagicMock(
side_effect=lambda model_name, team_id=None: (["ultrafast"],) if team_id == "team-1" else (None,)
)
patched_models.get_routable_upstream_model = MagicMock(
side_effect=lambda model_name, team_id=None: "openai/gpt-5.5" if team_id == "team-1" else "gpt-4"
)
with auth_as(team_id="team-1"):
team_response = client.get("/v1/models", params={**params, "client_version": "0.159.3"})
with auth_as():
teamless_response = client.get("/v1/models", params={**params, "client_version": "0.159.3"})
assert [[t["id"] for t in m["service_tiers"]] for m in team_response.json()["models"]] == [["ultrafast"]] * 2
assert [m["service_tiers"] for m in teamless_response.json()["models"]] == [[], []]
assert all(m["supported_reasoning_levels"] for m in team_response.json()["models"])
assert [m["supported_reasoning_levels"] for m in teamless_response.json()["models"]] == [[], []]
@pytest.mark.parametrize("params", [{}, {"scope": "expand"}])
def test_codex_service_tiers_resolved_via_internal_team_key(client, auth_as, patched_models, monkeypatch, params):
"""A team-scoped row's tiers are looked up by the internal routing key while the entry is keyed by
the public name Codex sends back as the model."""
from litellm.proxy import utils as proxy_utils
from litellm.proxy.auth import model_checks
internal_name = "model_name_team-1_c0ffee"
patched_models.get_model_list = MagicMock(
return_value=[
{"model_name": internal_name, "model_info": {"team_id": "team-1", "team_public_model_name": "gpt-4-team"}}
]
)
patched_models.get_model_names = MagicMock(return_value=[internal_name])
patched_models.get_configured_service_tiers = MagicMock(
side_effect=lambda model_name, team_id=None: (["ultrafast"],) if model_name == internal_name else ()
)
async def _fake_get_available_models_for_user(**kwargs):
return [internal_name]
monkeypatch.setattr(proxy_utils, "get_available_models_for_user", _fake_get_available_models_for_user)
monkeypatch.setattr(model_checks, "get_complete_model_list", lambda **kwargs: [internal_name])
with auth_as():
response = client.get("/v1/models", params={**params, "client_version": "0.159.3"})
assert response.status_code == 200
(entry,) = response.json()["models"]
assert (entry["slug"], [tier["id"] for tier in entry["service_tiers"]]) == ("gpt-4-team", ["ultrafast"])

View file

@ -10276,6 +10276,200 @@ def test_get_configured_display_name_skips_wildcard_pattern_matching():
)
@pytest.mark.parametrize(
"configured",
[["ultrafast"], ["priority", {"id": "ultrafast", "name": "Ultrafast", "description": "Fastest"}], [], "not-a-list"],
)
def test_get_configured_service_tiers_returns_the_deployment_model_info_value_as_set(configured):
router = litellm.Router(
model_list=[
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra"},
"model_info": {"service_tiers": configured},
}
]
)
assert router.get_configured_service_tiers("gpt-6-astra") == (configured,)
def test_get_configured_service_tiers_returns_one_value_per_deployment_in_model_list_order():
router = litellm.Router(
model_list=[
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra"},
"model_info": {"service_tiers": ["ultrafast"]},
},
{"model_name": "gpt-6-astra", "litellm_params": {"model": "openai/gpt-6-astra", "api_base": "https://a.example"}},
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra", "api_base": "https://b.example"},
"model_info": {"service_tiers": ["priority", "ultrafast"]},
},
]
)
assert router.get_configured_service_tiers("gpt-6-astra") == (["ultrafast"], None, ["priority", "ultrafast"])
def test_get_configured_service_tiers_returns_none_for_an_unset_deployment_and_nothing_for_an_unknown_name():
router = litellm.Router(
model_list=[
{
"model_name": "no-tiers-model",
"litellm_params": {"model": "openai/some-unmapped-model"},
}
]
)
assert router.get_configured_service_tiers("no-tiers-model") == (None,)
assert router.get_configured_service_tiers("not-a-real-model") == ()
def test_get_configured_service_tiers_does_not_apply_a_wildcard_deployment_to_matched_names():
router = litellm.Router(
model_list=[
{
"model_name": "openai/*",
"litellm_params": {"model": "openai/*"},
"model_info": {"service_tiers": ["ultrafast"]},
}
]
)
assert router.get_configured_service_tiers("openai/gpt-6-astra") == ()
def test_get_configured_service_tiers_reads_only_the_deployments_a_request_can_route_to():
router = litellm.Router(
model_list=[
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra"},
"model_info": {"service_tiers": ["ultrafast"]},
},
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra", "api_base": "https://paused.example"},
"model_info": {"blocked": True},
},
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra", "api_base": "https://team-2.example"},
"model_info": {"team_id": "team-2"},
},
]
)
assert router.get_configured_service_tiers("gpt-6-astra", team_id="team-1") == (["ultrafast"],)
assert router.get_configured_service_tiers("gpt-6-astra", team_id="team-2") == (["ultrafast"], None)
assert router.get_configured_service_tiers("gpt-6-astra") == (["ultrafast"],)
@pytest.mark.parametrize(
"alias_value, expected_group",
[
("gpt-6-astra", "gpt-6-astra"),
({"model": "gpt-6-astra", "hidden": True}, "gpt-6-astra"),
({"model": "", "hidden": False}, "gpt-6"),
],
ids=["string-alias", "item-alias", "malformed-alias-is-itself"],
)
def test_routable_model_group_is_the_alias_target_else_the_name_itself(alias_value, expected_group):
router = litellm.Router(
model_list=[{"model_name": "gpt-6-astra", "litellm_params": {"model": "openai/gpt-6-astra"}}],
model_group_alias={"gpt-6": alias_value},
)
assert router.routable_model_group("gpt-6") == expected_group
assert router.routable_model_group("gpt-6-astra") == "gpt-6-astra"
assert router.routable_model_group("not-a-real-model") == "not-a-real-model"
def test_get_configured_service_tiers_reads_an_alias_off_its_target_deployments():
router = litellm.Router(
model_list=[
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra"},
"model_info": {"service_tiers": ["ultrafast"]},
},
{
"model_name": "gpt-6-astra",
"litellm_params": {"model": "openai/gpt-6-astra", "api_base": "https://team-2.example"},
"model_info": {"team_id": "team-2", "service_tiers": ["priority"]},
},
],
model_group_alias={"gpt-6": "gpt-6-astra", "gpt-6-quiet": {"model": "gpt-6-astra", "hidden": True}},
)
assert router.get_configured_service_tiers("gpt-6") == router.get_configured_service_tiers("gpt-6-astra")
assert router.get_configured_service_tiers("gpt-6", team_id="team-1") == (["ultrafast"],)
assert router.get_configured_service_tiers("gpt-6", team_id="team-2") == (["ultrafast"], ["priority"])
assert router.get_configured_service_tiers("gpt-6-quiet", team_id="team-1") == (["ultrafast"],)
def _router_with_team_owned_deployments():
return litellm.Router(
model_list=[
{
"model_name": "owned-by-teams",
"litellm_params": {"model": "openai/gpt-5.5"},
"model_info": {"team_id": "team-1", "service_tiers": ["priority"]},
},
{
"model_name": "owned-by-teams",
"litellm_params": {"model": "openai/paused-model"},
"model_info": {"team_id": "team-2", "blocked": True, "service_tiers": ["paused"]},
},
{
"model_name": "owned-by-teams",
"litellm_params": {"model": "openai/team-2-model"},
"model_info": {"team_id": "team-2", "service_tiers": ["flex"]},
},
{
"model_name": "owned-and-shared",
"litellm_params": {"model": "openai/team-1-model"},
"model_info": {"team_id": "team-1", "service_tiers": ["priority"]},
},
{
"model_name": "owned-and-shared",
"litellm_params": {"model": "openai/shared-model"},
"model_info": {"service_tiers": ["flex"]},
},
{"model_name": "openai/*", "litellm_params": {"model": "openai/*"}},
],
model_group_alias={"nickname": "owned-by-teams"},
)
@pytest.mark.parametrize(
"model_name, team_id, upstream_model, service_tiers",
[
("owned-by-teams", "team-1", "openai/gpt-5.5", (["priority"],)),
("owned-by-teams", "team-2", "openai/team-2-model", (["flex"],)),
("nickname", "team-1", "openai/gpt-5.5", (["priority"],)),
("nickname", "team-2", "openai/team-2-model", (["flex"],)),
("owned-by-teams", "team-3", None, ()),
("owned-by-teams", None, "openai/gpt-5.5", (["priority"], ["flex"])),
("owned-and-shared", "team-1", "openai/team-1-model", (["priority"], ["flex"])),
("owned-and-shared", "team-2", "openai/shared-model", (["flex"],)),
("owned-and-shared", None, "openai/shared-model", (["flex"],)),
("openai/gpt-5.5", "team-1", None, ()),
("not-a-real-model", None, None, ()),
],
)
def test_upstream_model_and_service_tiers_are_read_off_the_deployments_the_team_can_route_to(
model_name, team_id, upstream_model, service_tiers
):
router = _router_with_team_owned_deployments()
assert router.get_routable_upstream_model(model_name, team_id) == upstream_model
assert router.get_configured_service_tiers(model_name, team_id) == service_tiers
def test_get_configured_display_name_treats_malformed_values_as_absent():
malformed = ["", " ", 12345, ["Kimi K3"], {"name": "Kimi K3"}, True]
router = litellm.Router(

View file

@ -10480,6 +10480,11 @@ export interface paths {
* This is just for compatibility with openai projects like aider.
*
* Query Parameters:
* - client_version: Sent by Codex CLI (`?client_version=0.159.3`) when it fetches a
* provider's model catalog. When present, the response is Codex's own
* catalog shape (`{"models": [...]}`) built from the same listing, with
* each model's `model_info.service_tiers` as its service tiers; absent,
* the OpenAI shape below is returned
* - include_metadata: Include additional metadata in the response with fallback information
* - fallback_type: Type of fallbacks to include ("general", "context_window", "content_policy")
* Defaults to "general" when include_metadata=true
@ -21236,6 +21241,11 @@ export interface paths {
* This is just for compatibility with openai projects like aider.
*
* Query Parameters:
* - client_version: Sent by Codex CLI (`?client_version=0.159.3`) when it fetches a
* provider's model catalog. When present, the response is Codex's own
* catalog shape (`{"models": [...]}`) built from the same listing, with
* each model's `model_info.service_tiers` as its service tiers; absent,
* the OpenAI shape below is returned
* - include_metadata: Include additional metadata in the response with fallback information
* - fallback_type: Type of fallbacks to include ("general", "context_window", "content_policy")
* Defaults to "general" when include_metadata=true
@ -64662,6 +64672,7 @@ export interface operations {
fallback_type?: string | null;
scope?: string | null;
healthy_only?: boolean | null;
client_version?: string | null;
};
header?: never;
path?: never;
@ -79056,6 +79067,7 @@ export interface operations {
fallback_type?: string | null;
scope?: string | null;
healthy_only?: boolean | null;
client_version?: string | null;
};
header?: never;
path?: never;