fix(proxy): serve Codex its own Bedrock catalog entries for Bedrock deployments (#50743)

* fix(proxy): serve Codex its own Bedrock catalog entries for Bedrock deployments

Codex builds separate catalogs for Bedrock's runtime and Mantle endpoints that turn off what Bedrock does not support yet. A model whose deployments run on Bedrock now takes its entry from those catalogs, vendored from Codex 0.166.0, instead of from Codex's OpenAI catalog.

* fix(proxy): keep Bedrock catalog entries from Codex releases that cannot decode them

codex debug models leaves out fields such as supports_parallel_tool_calls that Codex releases before 0.148.0 require, so a Bedrock entry now needs at least 0.148.0 as well as its source entry's minimum.

litellm/proxy/common_utils/codex_model_catalog.py

@@ -6,7 +6,8 @@ feature on) and decodes the body as its own ``ModelsResponse``, ``{"models": [Mo
 never the OpenAI list shape. Codex then ignores its own bundled catalog, so this body is all it knows
 about the provider's models. It is built from the rows ``/v1/models`` already lists: a model Codex
 knows, by its id or by the upstream model its deployments call once the provider's own prefixes are
-stripped, keeps Codex's own entry for it, minus any service tier some deployment behind it prices tokens
+stripped, keeps Codex's own entry for it, out of the catalog Codex itself uses for the provider those
+deployments call (its Bedrock catalogs for ``bedrock`` and ``bedrock_mantle``, its OpenAI one otherwise), minus any service tier some deployment behind it prices tokens
 but not that tier for; any other model gets an entry translated from the capabilities its deployments' model info
 declares (see ``codex_model_translation``). ``model_info.service_tiers`` becomes the entry's
 ``service_tiers``, which Codex turns into slash commands (``/ultrafast``) that send ``service_tier``
@@ -20,6 +21,15 @@ https://github.com/openai/codex/blob/rust-v0.166.0/codex-rs/models-manager/model
 as ``codex_bundled_models_0.166.0.json``, and ``codex_base_instructions.md`` is
 ``codex-rs/models-manager/prompt.md`` at the same tag. Refresh both by downloading those two paths at
 a newer tag.
+
+Codex builds its Bedrock catalogs in code rather than shipping them as files, so the ones for Bedrock's
+runtime and Mantle endpoints are vendored from Codex 0.166.0's own output, with the ``base_instructions``
+copy of each entry's ``model_messages.instructions_template`` that it adds for older clients dropped:
+``codex -c 'model_provider="amazon-bedrock-runtime"' debug models | jq 'del(.models[].base_instructions)'``
+as ``codex_bedrock_runtime_models_0.166.0.json``, and ``model_provider="amazon-bedrock"`` the same way as
+``codex_bedrock_mantle_models_0.166.0.json``. Refresh them at the same release as the bundled catalog,
+which they take each entry's ``minimal_client_version`` from, and recheck
+``CODEX_BEDROCK_MODELS_MINIMAL_CLIENT_VERSION`` against the fields the new output leaves out.
 """
 
 from __future__ import annotations
@@ -58,6 +68,14 @@ if TYPE_CHECKING:
 
 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.166.0.json")
+CodexBedrockEndpoint: TypeAlias = Literal["bedrock", "bedrock_mantle"]
+CODEX_BEDROCK_MODELS_MINIMAL_CLIENT_VERSION: Final = "0.148.0"
+CODEX_BEDROCK_MODELS_PATHS: Final[Mapping[CodexBedrockEndpoint, Path]] = MappingProxyType(
+    {
+        "bedrock": Path(__file__).with_name("codex_bedrock_runtime_models_0.166.0.json"),
+        "bedrock_mantle": Path(__file__).with_name("codex_bedrock_mantle_models_0.166.0.json"),
+    }
+)
 CODEX_CATALOG_BYTE_LIMIT: Final = 1024 * 1024
 CODEX_CHAT_MODES: Final = frozenset({"chat", "responses"})
 
@@ -129,8 +147,8 @@ _NATIVE_ALIAS_DEFAULTS: Final[CodexAliasDefaults] = "deployment"
 @dataclass(frozen=True, slots=True)
 class CodexCatalogRow:
     """What the catalog needs to know about one listed model: its public id, the listing's mode and
-    input limit, the upstream model behind it and the provider its deployment declares, the reasoning
-    effort its deployments pin, and the
+    input limit, the upstream model behind it and the provider its deployment declares, every routable
+    deployment's `upstreams` (model and declared provider), the reasoning effort its deployments pin, and the
     `model_info` values the operator set; `service_tiers` holds one raw `model_info.service_tiers`
     value per deployment behind the id, None where a deployment sets none, and
     `configured_max_input_tokens` one `model_info.max_input_tokens` per deployment the same way.
@@ -142,6 +160,7 @@ class CodexCatalogRow:
     max_input_tokens: int | None = None
     upstream_model: str | None = None
     upstream_provider: str | None = None
+    upstreams: tuple[tuple[str, str | None], ...] = ()
     reasoning_effort: str | None = None
     display_name: str | None = None
     service_tiers: tuple[object, ...] = ()
@@ -158,6 +177,39 @@ def bundled_codex_models() -> Mapping[str, CodexStockModel]:
     return MappingProxyType({model.slug: model for model in catalog.models})
 
 
+def _with_minimal_client_version(model: CodexStockModel, built_from: str | tuple[int, ...] | None) -> CodexStockModel:
+    source_minimum: Final = built_from if not isinstance(built_from, tuple) else ".".join(map(str, built_from))
+    minimum: Final = max(
+        (version for version in (CODEX_BEDROCK_MODELS_MINIMAL_CLIENT_VERSION, source_minimum) if version is not None),
+        key=Version,
+    )
+    return model.model_copy(update={"minimal_client_version": minimum})
+
+
+@cache
+def bundled_bedrock_codex_models(endpoint: CodexBedrockEndpoint) -> Mapping[str, CodexStockModel]:
+    """Codex's catalog for a Bedrock endpoint by slug, read once from the vendored file. `codex debug models`
+    prints no `minimal_client_version`, so each entry takes the later of the bundled entry's Codex built it from
+    and `CODEX_BEDROCK_MODELS_MINIMAL_CLIENT_VERSION`, the first release that decodes an entry without the fields
+    `debug models` leaves out, such as `supports_parallel_tool_calls`."""
+    from litellm.llms.bedrock.common_utils import bedrock_model_name_without_provider
+
+    catalog: Final = CodexStockCatalog.model_validate_json(
+        CODEX_BEDROCK_MODELS_PATHS[endpoint].read_text(encoding="utf-8")
+    )
+    built_from: Final = MappingProxyType(
+        {slug: model.minimal_client_version for slug, model in bundled_codex_models().items()}
+    )
+    return MappingProxyType(
+        {
+            model.slug: _with_minimal_client_version(
+                model, built_from.get(bedrock_model_name_without_provider(model.slug))
+            )
+            for model in catalog.models
+        }
+    )
+
+
 @cache
 def codex_base_instructions() -> str:
     return CODEX_BASE_INSTRUCTIONS_PATH.read_text(encoding="utf-8")
@@ -274,15 +326,67 @@ def _provider_model_name(upstream_model: str, upstream_provider: str | None) ->
     return model if name_for is None else name_for(model)
 
 
-def _stock_for(row: CodexCatalogRow, stock: Mapping[str, CodexStockModel]) -> CodexStockModel | None:
+@dataclass(frozen=True, slots=True)
+class _StockMatch:
+    """Codex's entry for a row, and whether the row lists it under another name than Codex's (an alias)."""
+
+    model: CodexStockModel
+    alias: bool
+
+
+def _on_endpoint(
+    declared: Sequence[tuple[str | None, str]], endpoint: CodexBedrockEndpoint
+) -> tuple[CodexBedrockEndpoint, str] | None:
+    return next(((endpoint, model) for provider, model in declared if provider == endpoint), None)
+
+
+def _bedrock_upstream(row: CodexCatalogRow) -> tuple[CodexBedrockEndpoint, str] | None:
+    """The Bedrock endpoint a request to the row can reach whose Codex catalog promises the least, with the model
+    a deployment on it calls. A request can route to any deployment, and Codex's Bedrock entries leave out what
+    Bedrock does not support yet, so one Bedrock deployment puts the whole row on its catalog, the runtime one
+    (no tool search) ahead of Mantle's."""
+    from litellm.litellm_core_utils.get_llm_provider_logic import declared_provider
+
+    first: Final = () if row.upstream_model is None else ((row.upstream_model, row.upstream_provider),)
+    declared: Final = tuple((declared_provider(model, provider), model) for model, provider in (*first, *row.upstreams))
+    on_bedrock: Final = (_on_endpoint(declared, endpoint) for endpoint in CODEX_BEDROCK_MODELS_PATHS)
+    return next((found for found in on_bedrock if found is not None), None)
+
+
+def _bedrock_stock_for(row_id: str, endpoint: CodexBedrockEndpoint, upstream_model: str) -> _StockMatch | None:
+    """Codex's entry for the row's id or the Bedrock model as written, else for the same model under any of
+    Bedrock's routing and region prefixes (`eu.openai.gpt-6.1-sol` takes `global.openai.gpt-6.1-sol`'s)."""
+    from litellm.llms.bedrock.common_utils import bedrock_model_name_without_provider
+
+    stock: Final = bundled_bedrock_codex_models(endpoint)
+    names: Final = (row_id, upstream_model.removeprefix(f"{endpoint}/"))
+    exact: Final = next((stock[name] for name in names if name in stock), None)
+    model_names: Final = tuple(bedrock_model_name_without_provider(name) for name in names)
+    same_model: Final = (
+        next((model for slug, model in stock.items() if bedrock_model_name_without_provider(slug) == name), None)
+        for name in model_names
+    )
+    found: Final = exact if exact is not None else next((model for model in same_model if model is not None), None)
+    if found is None:
+        return None
+    return _StockMatch(found, model_names[0] != bedrock_model_name_without_provider(found.slug))
+
+
+def _stock_for(row: CodexCatalogRow, stock: Mapping[str, CodexStockModel]) -> _StockMatch | None:
     """Codex's entry for the row's id, else for its upstream model as written, else for that model's name
-    once its provider's own prefixes are stripped (`bedrock/us.openai.gpt-6.1-sol` is `gpt-6.1-sol`)."""
+    once its provider's own prefixes are stripped (`openai/gpt-6.1-sol` is `gpt-6.1-sol`), out of Codex's
+    Bedrock catalog when a deployment behind the row is on Bedrock and `stock` otherwise. A Bedrock model that
+    catalog does not know has no entry, since `stock`'s would promise what Bedrock does not support."""
+    bedrock: Final = _bedrock_upstream(row)
+    if bedrock is not None:
+        return _bedrock_stock_for(row.id, *bedrock)
     names: Final = (
         (row.id,)
         if row.upstream_model is None
         else (row.id, row.upstream_model, _provider_model_name(row.upstream_model, row.upstream_provider))
     )
-    return next((stock[name] for name in names if name in stock), None)
+    found: Final = next((stock[name] for name in names if name in stock), None)
+    return None if found is None else _StockMatch(found, row.id != found.slug)
 
 
 def _readable_by(stock_model: CodexStockModel, client_version: str | None) -> bool:
@@ -391,7 +495,7 @@ def _entry(
     client_version: str | None,
 ) -> CodexCatalogEntry:
     matched: Final = _stock_for(row, stock)
-    stock_model: Final = matched if matched is not None and _readable_by(matched, client_version) else None
+    stock_model: Final = matched.model if matched is not None and _readable_by(matched.model, client_version) else None
     stock_tiers: Final = MappingProxyType(
         {tier.id: tier for tier in (stock_model.service_tiers if stock_model is not None else ())}
     )
@@ -412,7 +516,7 @@ def _entry(
     upgrade: Final = (
         stock_model.upgrade if stock_model.upgrade is not None and stock_model.upgrade.model in served else None
     )
-    is_alias: Final = row.id != stock_model.slug
+    is_alias: Final = matched is not None and matched.alias
     deployment_decides: Final = is_alias and alias_defaults == "deployment"
     stock_name: Final = row.id if is_alias else stock_model.display_name
     listing: Final = {
@@ -659,6 +763,7 @@ def _catalog_row(
         upstream_provider=(
             registry.resolve(lookup_id, kind="upstream_provider", team_id=team_id) if registry is not None else None
         ),
+        upstreams=registry.resolve(lookup_id, kind="upstreams", team_id=team_id) if registry is not None else (),
         display_name=registry.group(label_id, field="display_name") if registry is not None else None,
         reasoning_effort=(
             _shared_pinned_effort(registry.group(lookup_id, field="pinned_reasoning_efforts", team_id=team_id))

litellm/router_api.py

@@ -153,6 +153,7 @@ ResolveKind: TypeAlias = Literal[
     "team_model_alternatives",
     "routing_group_alternatives",
     "search_tool_support",
+    "upstreams",
 ]
 GroupField: TypeAlias = Literal[
     "info",
@@ -780,6 +781,17 @@ class RouterRegistry:
         served_model: str | None = None,
     ) -> tuple[bool | None, ...]: ...
 
+    @overload
+    def resolve(
+        self,
+        name: str,
+        *,
+        kind: Literal["upstreams"],
+        team_id: str | None = None,
+        custom_llm_provider: str | None = None,
+        served_model: str | None = None,
+    ) -> tuple[tuple[str, str | None], ...]: ...
+
     @overload
     def resolve(
         self,
@@ -853,6 +865,8 @@ class RouterRegistry:
         - ``routing_group_alternatives``: (was ``routing_group_has_alternatives``)
         - ``search_tool_support``: per routable deployment, whether it answers Codex's client
           ``tool_search`` (was ``get_search_tool_support``)
+        - ``upstreams``: per routable deployment, its upstream model and the provider it declares, read
+          the way ``upstream`` and ``upstream_provider`` read the first one
         """
         r: Final = self._router
         if kind == "from_id":
@@ -874,7 +888,7 @@ class RouterRegistry:
             return name if served_model is None else r._core_resolve_served_model_name(name, served_model)
         if kind == "recognized":
             return r._core_is_recognized_model(name)
-        if kind in ("upstream", "upstream_provider", "search_tool_support"):
+        if kind in ("upstream", "upstream_provider", "search_tool_support", "upstreams"):
             return self._resolve_routable(name, kind, team_id)
         if kind == "group_of_id":
             return r._core_get_model_group(id=name)
@@ -887,14 +901,16 @@ class RouterRegistry:
     def _resolve_routable(
         self,
         name: str,
-        kind: Literal["upstream", "upstream_provider", "search_tool_support"],
+        kind: Literal["upstream", "upstream_provider", "search_tool_support", "upstreams"],
         team_id: str | None,
-    ) -> str | tuple[bool | None, ...] | None:
+    ) -> str | tuple[bool | None, ...] | tuple[tuple[str, str | None], ...] | None:
         r: Final = self._router
         if kind == "upstream":
             return r._core_get_routable_upstream_model(name, team_id=team_id)
         if kind == "upstream_provider":
             return r._core_get_routable_upstream_provider(name, team_id=team_id)
+        if kind == "upstreams":
+            return r._core_get_routable_upstreams(name, team_id=team_id)
         return r._core_get_search_tool_support(name, team_id=team_id)
 
     def aliases(

litellm/router_rust/surface/registry.py

@@ -353,6 +353,14 @@ def _bedrock_profile(params: Mapping[str, object], model: str) -> bool:
     return params.get("custom_llm_provider") == "bedrock" or model.startswith("bedrock/")
 
 
+def _declared_upstream(params: Mapping[str, object]) -> tuple[str, str | None] | None:
+    model: Final = params.get("model")
+    provider: Final = params.get("custom_llm_provider")
+    if not isinstance(model, str):
+        return None
+    return model, declared_provider(model, provider if isinstance(provider, str) else None)
+
+
 def _search_tool_support(deployment: Mapping[str, object]) -> bool | None:
     """A deployment's answer to Codex's client ``tool_search``."""
     from litellm.responses.litellm_completion_transformation.tool_search import chat_config_builds_anthropic_request
@@ -2107,6 +2115,15 @@ class HostRegistry(HostModelInfo):
             model if isinstance(model, str) else None, provider if isinstance(provider, str) else None
         )
 
+    def _core_get_routable_upstreams(
+        self, model_name: str, team_id: str | None = None
+    ) -> tuple[tuple[str, str | None], ...]:
+        """Per routable deployment that names a model: that model and the provider the deployment declares."""
+        upstreams: Final = (
+            _declared_upstream(_params_of(deployment)) for deployment in self._routable_deployments(model_name, team_id)
+        )
+        return tuple(upstream for upstream in upstreams if upstream is not None)
+
     def _core_get_routable_deployment_model_info(
         self, model_name: str, team_id: str | None = None
     ) -> "tuple[ModelMapInfo | None, ...]":

packaging/litellm-core/pyproject.toml

@@ -62,6 +62,8 @@ include = [
     "litellm/proxy/model_insights_tasks.json",
     "litellm/proxy/common_utils/codex_base_instructions.md",
     "litellm/proxy/common_utils/codex_bundled_models_0.166.0.json",
+    "litellm/proxy/common_utils/codex_bedrock_mantle_models_0.166.0.json",
+    "litellm/proxy/common_utils/codex_bedrock_runtime_models_0.166.0.json",
 ]
 exclude = [
     "litellm/proxy/_experimental/out",

pyproject.toml

@@ -346,6 +346,8 @@ include = [
     "litellm/proxy/public_endpoints/whats_new.json",
     "litellm/proxy/common_utils/codex_base_instructions.md",
     "litellm/proxy/common_utils/codex_bundled_models_0.166.0.json",
+    "litellm/proxy/common_utils/codex_bedrock_mantle_models_0.166.0.json",
+    "litellm/proxy/common_utils/codex_bedrock_runtime_models_0.166.0.json",
     { path = "model_prices_and_context_window.json", format = "sdist" },
 ]
 exclude = [

tests/unit/proxy/common_utils/test_codex_model_catalog.py

@@ -1,21 +1,25 @@
 import json
 import logging
-from collections.abc import Mapping
+from collections.abc import Iterator, Mapping
 from dataclasses import replace
 from types import MappingProxyType
 from typing import Final, Literal
 
 import pytest
+from packaging.version import Version
 
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
 from litellm.litellm_core_utils.llm_cost_calc.utils import get_service_tier_cost_key
+from litellm.llms.bedrock.common_utils import bedrock_model_name_without_provider
 from litellm.proxy.common_utils.codex_model_catalog import (
+    CODEX_BEDROCK_MODELS_MINIMAL_CLIENT_VERSION,
     CODEX_CATALOG_BYTE_LIMIT,
     CodexCatalogRow,
     CodexStockModel,
     CodexStockUpgrade,
+    bundled_bedrock_codex_models,
     bundled_codex_models,
     codex_catalog_etag,
     codex_catalog_rows,
@@ -32,9 +36,7 @@ _PROMPT = "base prompt"
 _FAST = {"id": "priority", "name": "Fast", "description": "2x speed, increased usage"}
 
 
-def _stock(
-    slug, *, visibility="list", supported_in_api=True, upgrade=None, service_tiers=(_FAST,), default_tier=None
-):
+def _stock(slug, *, visibility="list", supported_in_api=True, upgrade=None, service_tiers=(_FAST,), default_tier=None):
     return CodexStockModel(
         slug=slug,
         display_name=slug.upper(),
@@ -633,14 +635,17 @@ def test_default_service_tier_survives_only_while_still_offered(configured, expe
     assert entry["default_service_tier"] == expected
 
 
-
 @pytest.mark.parametrize(("alias_defaults", "expected"), [("codex", "priority"), ("deployment", None)])
 def test_alias_with_offered_tiers_takes_codexs_default_tier_only_under_codex_defaults(alias_defaults, expected):
     (entry,) = _models(
         _row("team-sol", upstream_model="gpt-6-sol", service_tiers=["priority"]), alias_defaults=alias_defaults
     )
 
-    assert (_tiers(entry), entry["default_service_tier"]) == ([("priority", "Fast", "2x speed, increased usage")], expected)
+    assert (_tiers(entry), entry["default_service_tier"]) == (
+        [("priority", "Fast", "2x speed, increased usage")],
+        expected,
+    )
+
 
 def test_non_chat_and_wildcard_rows_are_left_out_of_the_catalog():
     body = _body(
@@ -836,6 +841,7 @@ def test_catalog_rows_read_the_router_by_each_entry_lookup_id():
             max_input_tokens=7,
             upstream_model="openai/gpt-5.5",
             upstream_provider="openai",
+            upstreams=(("openai/gpt-5.5", "openai"), ("openai/gpt-5.5", "openai")),
             display_name="Team 5.5",
             service_tiers=(["ultrafast"], None),
             supported_reasoning_efforts=router.get_model_group_info(
@@ -848,6 +854,7 @@ def test_catalog_rows_read_the_router_by_each_entry_lookup_id():
             id="plain",
             upstream_model="openai/some-unmapped-model",
             upstream_provider="openai",
+            upstreams=(("openai/some-unmapped-model", "openai"),),
             service_tiers=(None,),
             search_tool_support=(None,),
             configured_max_input_tokens=(None,),
@@ -882,6 +889,7 @@ def test_catalog_rows_read_an_alias_off_its_target_under_the_alias_name():
         max_input_tokens=9,
         upstream_model="openai/gpt-6-astra",
         upstream_provider="openai",
+        upstreams=(("openai/gpt-6-astra", "openai"),),
         service_tiers=(["ultrafast"],),
         supported_reasoning_efforts=router.get_model_group_info("gpt-6-astra").supported_reasoning_efforts,
         search_tool_support=(None,),
@@ -1123,19 +1131,7 @@ _SOL_SLUG: Final = next(
 )
 
 
-@pytest.mark.parametrize(
-    "upstream",
-    [
-        "bedrock/us.openai.{slug}",
-        "bedrock/global.openai.{slug}",
-        "bedrock/converse/us.openai.{slug}",
-        "bedrock/openai.{slug}",
-        "bedrock_mantle/openai.{slug}",
-        "bedrock_mantle/us-east-1/openai.{slug}",
-        "openai/{slug}",
-        "{slug}",
-    ],
-)
+@pytest.mark.parametrize("upstream", ["openai/{slug}", "{slug}"])
 @pytest.mark.parametrize("slug", sorted(bundled_codex_models()))
 def test_a_model_codex_knows_keeps_codexs_entry_under_any_provider_prefix(upstream: str, slug: str) -> None:
     stock: Final = bundled_codex_models()[slug].model_dump(mode="json")
@@ -1150,24 +1146,227 @@ def test_a_model_codex_knows_keeps_codexs_entry_under_any_provider_prefix(upstre
     assert entry["supported_reasoning_levels"] == stock["supported_reasoning_levels"]
 
 
-@pytest.mark.parametrize("model", ["us.openai.{slug}", "converse/global.openai.{slug}"])
-def test_a_bedrock_deployment_naming_its_provider_apart_from_the_model_keeps_codexs_entry(model: str) -> None:
+_BEDROCK_UPSTREAMS: Final = MappingProxyType(
+    {
+        "bedrock": (
+            "bedrock/us.openai.{name}",
+            "bedrock/global.openai.{name}",
+            "bedrock/eu.openai.{name}",
+            "bedrock/converse/us.openai.{name}",
+            "bedrock/openai.{name}",
+        ),
+        "bedrock_mantle": ("bedrock_mantle/openai.{name}", "bedrock_mantle/us-east-1/openai.{name}"),
+    }
+)
+_LISTING_FIELDS: Final = frozenset({"slug", "display_name", "priority"})
+_TIER_FIELDS: Final = frozenset({"service_tiers", "additional_speed_tiers", "default_service_tier"})
+
+
+def _bedrock_names(endpoint: Literal["bedrock", "bedrock_mantle"]) -> tuple[str, ...]:
+    return tuple(
+        dict.fromkeys(bedrock_model_name_without_provider(slug) for slug in bundled_bedrock_codex_models(endpoint))
+    )
+
+
+def _bedrock_stock(endpoint: Literal["bedrock", "bedrock_mantle"], name: str) -> dict[str, object]:
+    model: Final = next(
+        model
+        for slug, model in bundled_bedrock_codex_models(endpoint).items()
+        if bedrock_model_name_without_provider(slug) == name
+    )
+    return model.model_dump(mode="json", exclude_unset=True)
+
+
+def _codex_fields(
+    entry: Mapping[str, object], stock: Mapping[str, object], skipped: frozenset[str]
+) -> dict[str, object]:
+    return {field: entry.get(field, _ABSENT) for field in stock if field not in skipped}
+
+
+def _without(stock: Mapping[str, object], skipped: frozenset[str]) -> dict[str, object]:
+    return {field: value for field, value in stock.items() if field not in skipped}
+
+
+def _only_entry(*rows: CodexCatalogRow, client_version: str | None = None) -> dict[str, object]:
+    (entry,) = json.loads(
+        codex_models_response_json(rows, instructions=_PROMPT, byte_limit=None, client_version=client_version).json
+    )["models"]
+    return entry
+
+
+def _bedrock_cases() -> Iterator[tuple[str, str, str]]:
+    for endpoint, upstreams in _BEDROCK_UPSTREAMS.items():
+        for upstream in upstreams:
+            for name in _bedrock_names(endpoint):
+                yield endpoint, upstream, name
+
+
+@pytest.mark.parametrize(("endpoint", "upstream", "name"), tuple(_bedrock_cases()))
+def test_a_bedrock_deployment_takes_codexs_entry_for_its_bedrock_endpoint_not_openais(
+    endpoint: Literal["bedrock", "bedrock_mantle"], upstream: str, name: str
+) -> None:
+    stock: Final = _bedrock_stock(endpoint, name)
+
+    entry: Final = _only_entry(_row("team-model", upstream_model=upstream.format(name=name)))
+    openai_entry: Final = _only_entry(_row("team-model", upstream_model=f"openai/{name}"))
+
+    assert _codex_fields(entry, stock, _LISTING_FIELDS) == _without(stock, _LISTING_FIELDS)
+    assert _codex_fields(openai_entry, stock, _LISTING_FIELDS) != _without(stock, _LISTING_FIELDS)
+
+
+@pytest.mark.parametrize("model", ["us.openai.{name}", "converse/global.openai.{name}"])
+def test_a_bedrock_deployment_naming_its_provider_apart_from_the_model_takes_codexs_bedrock_entry(model: str) -> None:
+    name: Final = _bedrock_names("bedrock")[0]
     router: Final = litellm.Router(
         model_list=[
             {
                 "model_name": "team-model",
                 "litellm_params": {
-                    "model": model.format(slug=_SOL_SLUG),
+                    "model": model.format(name=name),
                     "custom_llm_provider": "bedrock",
                     "aws_region_name": "us-east-1",
                 },
             }
         ]
     )
+    stock: Final = _bedrock_stock("bedrock", name)
 
     entry: Final = _served(router, "team-model")["team-model"]
 
-    assert entry["model_messages"] == bundled_codex_models()[_SOL_SLUG].model_dump(mode="json")["model_messages"]
+    assert _codex_fields(entry, stock, _LISTING_FIELDS | _TIER_FIELDS) == _without(
+        stock, _LISTING_FIELDS | _TIER_FIELDS
+    )
+
+
+@pytest.mark.parametrize(
+    "upstreams",
+    [
+        ("openai/{name}", "bedrock/us.openai.{name}"),
+        ("bedrock/us.openai.{name}", "openai/{name}"),
+        ("bedrock_mantle/openai.{name}", "bedrock/us.openai.{name}"),
+        ("openai/{name}", "bedrock_mantle/openai.{name}", "bedrock/us.openai.{name}"),
+    ],
+)
+def test_a_model_name_any_of_whose_deployments_is_on_bedrock_runtime_takes_codexs_runtime_entry(
+    upstreams: tuple[str, ...],
+) -> None:
+    name: Final = next(name for name in _bedrock_names("bedrock") if name in bundled_codex_models())
+    router: Final = litellm.Router(
+        model_list=[
+            {"model_name": name, "litellm_params": {"model": upstream.format(name=name)}} for upstream in upstreams
+        ]
+    )
+    stock: Final = _bedrock_stock("bedrock", name)
+
+    entry: Final = _served(router, name)[name]
+
+    assert _codex_fields(entry, stock, _LISTING_FIELDS | _TIER_FIELDS) == _without(
+        stock, _LISTING_FIELDS | _TIER_FIELDS
+    )
+
+
+@pytest.mark.parametrize(
+    "upstreams", [("openai/{name}", "bedrock_mantle/openai.{name}"), ("bedrock_mantle/openai.{name}",)]
+)
+def test_a_model_name_on_bedrock_mantle_but_not_runtime_takes_codexs_mantle_entry(upstreams: tuple[str, ...]) -> None:
+    name: Final = next(name for name in _bedrock_names("bedrock_mantle") if name in bundled_codex_models())
+    router: Final = litellm.Router(
+        model_list=[
+            {"model_name": name, "litellm_params": {"model": upstream.format(name=name)}} for upstream in upstreams
+        ]
+    )
+    stock: Final = _bedrock_stock("bedrock_mantle", name)
+
+    entry: Final = _served(router, name)[name]
+
+    assert _codex_fields(entry, stock, _LISTING_FIELDS | _TIER_FIELDS) == _without(
+        stock, _LISTING_FIELDS | _TIER_FIELDS
+    )
+
+
+def _left_out_of_bedrock() -> Iterator[tuple[str, str]]:
+    for endpoint, upstream in (
+        ("bedrock", "bedrock/us.openai.{slug}"),
+        ("bedrock_mantle", "bedrock_mantle/openai.{slug}"),
+    ):
+        for slug in sorted(frozenset(bundled_codex_models()) - frozenset(_bedrock_names(endpoint))):
+            yield upstream, slug
+
+
+@pytest.mark.parametrize(("upstream", "slug"), tuple(_left_out_of_bedrock()))
+def test_a_bedrock_deployment_of_a_model_its_bedrock_catalog_leaves_out_gets_a_translated_entry(
+    upstream: str, slug: str
+) -> None:
+    entry: Final = _only_entry(_row("team-model", upstream_model=upstream.format(slug=slug)))
+
+    assert "model_messages" not in entry and entry["base_instructions"] == _PROMPT
+
+
+def _bedrock_endpoint_names() -> Iterator[tuple[str, str]]:
+    for endpoint in _BEDROCK_UPSTREAMS:
+        for name in _bedrock_names(endpoint):
+            yield endpoint, name
+
+
+def _just_below(version: str) -> str:
+    release: Final = Version(version).release
+    major, minor, micro = (*release, 0, 0)[:3]
+    return f"{major}.{minor}.{micro - 1}" if micro else f"{major}.{minor - 1}.999"
+
+
+@pytest.mark.parametrize(("endpoint", "name"), tuple(_bedrock_endpoint_names()))
+def test_no_bedrock_entry_reaches_a_codex_whose_decoder_needs_fields_debug_models_leaves_out(
+    endpoint: Literal["bedrock", "bedrock_mantle"], name: str
+) -> None:
+    """Codex rust-v0.147.0's ModelInfo requires `supports_parallel_tool_calls`, which 0.166.0's
+    `codex debug models` does not print (checked in openai/codex on 2026-11-09)."""
+    entry: Final = _only_entry(
+        _row("team-model", upstream_model=_BEDROCK_UPSTREAMS[endpoint][0].format(name=name)), client_version="0.147.0"
+    )
+
+    assert "model_messages" not in entry and entry["base_instructions"] == _PROMPT
+
+
+@pytest.mark.parametrize(("endpoint", "name"), tuple(_bedrock_endpoint_names()))
+def test_a_codex_too_old_to_decode_a_bedrock_entry_gets_a_translated_one_instead(
+    endpoint: Literal["bedrock", "bedrock_mantle"], name: str
+) -> None:
+    built_from: Final = bundled_codex_models()[name].minimal_client_version
+    assert built_from is None or isinstance(built_from, str)
+    minimum: Final = max(
+        (version for version in (built_from, CODEX_BEDROCK_MODELS_MINIMAL_CLIENT_VERSION) if version is not None),
+        key=Version,
+    )
+    row: Final = _row("team-model", upstream_model=_BEDROCK_UPSTREAMS[endpoint][0].format(name=name))
+
+    older: Final = _only_entry(row, client_version=_just_below(minimum))
+    at_minimum: Final = _only_entry(row, client_version=minimum)
+
+    assert "model_messages" not in older and older["base_instructions"] == _PROMPT
+    assert at_minimum["model_messages"] == _bedrock_stock(endpoint, name)["model_messages"]
+
+
+@pytest.mark.parametrize("listed_as", ["{name}", "us.openai.{name}", "eu.openai.{name}"])
+def test_a_bedrock_model_listed_under_its_own_name_keeps_codexs_defaults_in_the_native_catalog(listed_as: str) -> None:
+    name: Final = next(
+        name for name in _bedrock_names("bedrock") if _bedrock_stock("bedrock", name).get("default_reasoning_level")
+    )
+    upstream_model: Final = f"bedrock/us.openai.{name}"
+    upstream: Final = json.dumps({"models": [{"slug": "upstream-model", "priority": 0}]}).encode()
+
+    merged: Final = merged_codex_catalog(
+        upstream,
+        (
+            _row(listed_as.format(name=name), upstream_model=upstream_model),
+            _row("team-model", upstream_model=upstream_model),
+        ),
+        instructions=_PROMPT,
+        client_version="0.166.0",
+    )
+
+    own_name, alias = json.loads(merged)["models"][1:]
+    assert own_name["default_reasoning_level"] == _bedrock_stock("bedrock", name)["default_reasoning_level"]
+    assert alias["default_reasoning_level"] is None
 
 
 def test_an_upstream_with_no_provider_prefix_matches_only_by_its_own_name() -> None:

tests/unit/proxy/proxy_server/test_routes_models.py

@@ -57,6 +57,7 @@ def patched_models(monkeypatch):
     router.get_deployment_by_model_group_name = MagicMock(return_value=deployment)
     router.get_routable_upstream_model = MagicMock(return_value="gpt-4")
     router.get_routable_upstream_provider = MagicMock(return_value=None)
+    router.get_routable_upstreams = MagicMock(return_value=())
     router.get_configured_display_name = MagicMock(return_value=None)
     router.get_configured_service_tiers = MagicMock(return_value=())
     router.get_configured_max_input_tokens = MagicMock(return_value=())

tests/unit/proxy/router_entry_point_stubs.py

@@ -103,6 +103,8 @@ def route_entry_points_to_stubs(router: _RouterDouble) -> _RouterDouble:
             return router.get_routable_upstream_provider(name, **_given(team_id=team_id))
         if kind == "search_tool_support":
             return router.get_search_tool_support(name, **_given(team_id=team_id))
+        if kind == "upstreams":
+            return router.get_routable_upstreams(name, **_given(team_id=team_id))
         if kind == "from_id":
             # The old call sites passed the provider positionally, exactly when the new ones pass it.
             if custom_llm_provider is _NOT_GIVEN:

tests/unit/test_router/core/test_gaps_l0001.py

@@ -331,6 +331,25 @@ def test_search_tool_support_comes_from_model_info_else_an_anthropic_request_sha
     assert registry.resolve("zzz", kind="search_tool_support") == ()  # pyright: ignore[reportCallIssue, reportArgumentType]
 
 
+def test_upstreams_name_every_routable_deployments_model_and_the_provider_it_declares(make_core: MakeCore) -> None:
+    core: Final = make_core(
+        model_list=[
+            deployment("g", "o1"),
+            deployment("g", "br", model="us.openai.gpt-6.1-sol", custom_llm_provider="bedrock"),
+            deployment("g", "mt", model="bedrock_mantle/openai.gpt-6.1-sol"),
+            deployment("g", "bare", model="gpt-6.1-sol"),
+        ]
+    )
+    registry: Final = core.registry()
+    assert registry.resolve("g", kind="upstreams") == (  # pyright: ignore[reportCallIssue, reportArgumentType]
+        ("openai/o1", "openai"),
+        ("us.openai.gpt-6.1-sol", "bedrock"),
+        ("bedrock_mantle/openai.gpt-6.1-sol", "bedrock_mantle"),
+        ("gpt-6.1-sol", None),
+    )
+    assert registry.resolve("zzz", kind="upstreams") == ()  # pyright: ignore[reportCallIssue, reportArgumentType]
+
+
 # --- per-request strategy overrides ---------------------------------------------------------------