codex-mmo/libexec/mmo_catalog_data.py

#!/usr/bin/env python3
"""Provider/model catalog loading, overlays, and semantic validation."""

from __future__ import annotations

import math
import re
from collections.abc import Mapping
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

from mmo_util import (
    config_root,
    deep_merge,
    install_root,
    read_toml,
    valid_absolute_uri,
    valid_http_header_value,
    validate_id,
)
from mmo_version import MMO_SCHEMA_VERSION

ALLOWED_DRIVERS = {
    "switchyard",
    "codex_builtin",
    "codex_custom",
    "codex_oss",
    "catalog_only",
}
ALLOWED_WIRE_PROTOCOLS = {
    "openai_chat",
    "openai_responses",
    "anthropic_messages",
    "codex_builtin",
    "codex_oss",
    "catalog_only",
}
ALLOWED_BILLING_MODES = {"api", "subscription", "chatgpt_subscription", "local", "catalog_only"}
ALLOWED_MODEL_KINDS = {
    "chat",
    "vision_chat",
    "ocr",
    "image_generation",
    "video_generation",
    "audio_transcription",
    "agent_service",
}
ALLOWED_MODALITIES = {"text", "image", "audio", "video", "file"}
ALLOWED_REASONING = {"none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra"}


def user_catalog_root() -> Path:
    return config_root() / "catalog.d"


ROUTE_FIELDS = {
    "driver",
    "name",
    "api_operator",
    "access_product",
    "wire_protocol",
    "billing_mode",
    "base_url",
    "provider_id",
    "auth",
    "auth_link_mode",
    "wire_api",
    "credential_envs",
    "http_headers",
    "env_http_headers",
    "extra_headers",
    "request_max_retries",
    "stream_max_retries",
    "stream_idle_timeout_ms",
    "max_retries",
    "transport_modalities",
    "transport_output_modalities",
    "tool_calling",
    "parallel_tool_calls",
    "preserves_tool_media",
    "tool_result_modalities",
    "supports_documents",
    "resource_group",
    "inventory",
    "openrouter_policy",
}
ROUTE_COMMON_FIELDS = {
    "driver",
    "name",
    "api_operator",
    "access_product",
    "wire_protocol",
    "billing_mode",
    "transport_modalities",
    "transport_output_modalities",
    "tool_calling",
    "parallel_tool_calls",
    "preserves_tool_media",
    "tool_result_modalities",
    "supports_documents",
    "resource_group",
    "inventory",
    "openrouter_policy",
}
ROUTE_DRIVER_FIELDS = {
    "switchyard": {
        "base_url",
        "credential_envs",
        "extra_headers",
        "max_retries",
    },
    "codex_custom": {
        "base_url",
        "wire_api",
        "credential_envs",
        "http_headers",
        "env_http_headers",
        "request_max_retries",
        "stream_max_retries",
        "stream_idle_timeout_ms",
    },
    "codex_builtin": {"provider_id", "auth", "auth_link_mode"},
    "codex_oss": {"provider_id"},
    # Catalog-only providers retain endpoint/auth facts for inventory and
    # operator diagnostics, but none of these fields is emitted for execution.
    "catalog_only": {
        "base_url",
        "wire_api",
        "credential_envs",
        "auth",
    },
}
MODEL_FIELDS = {
    "maker",
    "route",
    "upstream_id",
    "display_name",
    "description",
    "kind",
    "agent_compatible",
    "context_window",
    "max_output_tokens",
    "reasoning_levels",
    "default_reasoning",
    "modalities",
    "output_modalities",
    "supports_tool_images",
    "supports_documents",
    "tool_calling",
    "supports_custom_tools",
    "parallel_tool_calls",
    "supports_reasoning_summaries",
    "structured_output",
    "availability",
    "capability_confidence",
    "source",
    "availability_source",
    "capability_source",
    "pricing_source",
    "inventory",
    "resource_group",
    "input_cost_per_million",
    "cached_input_cost_per_million",
    "cache_write_input_cost_per_million",
    "output_cost_per_million",
    "unit_cost_usd",
    "extra_body",
    "route_policy",
}
OPENROUTER_POLICY_FIELDS = {
    "only",
    "order",
    "allow_fallbacks",
    "require_parameters",
    "data_collection",
    "zdr",
    "quantizations",
    "sort",
    "max_price",
}
RESOURCE_FIELDS = {"description", "lock_key", "max_active"}


def _reject_unknown_fields(value: Mapping[str, Any], allowed: set[str], label: str) -> None:
    unknown = sorted(set(value) - allowed)
    if unknown:
        raise ValueError(f"{label} has unknown fields: {', '.join(unknown)}")


def _schema_version(value: Any, expected: int, label: str) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value != expected:
        raise ValueError(f"unsupported {label} schema_version")
    return value


def _nonempty_string(value: Any, label: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{label} must be a non-empty string")
    return value


def _optional_nonempty_string(value: Any, label: str) -> str | None:
    if value is None:
        return None
    return _nonempty_string(value, label)


def _enum_string(value: Any, allowed: set[str], label: str) -> str:
    if not isinstance(value, str) or value not in allowed:
        raise ValueError(f"{label} must be one of {sorted(allowed)}")
    return value


def _load_catalog_fragment(path: Path) -> dict[str, Any]:
    data = read_toml(path)
    _reject_unknown_fields(
        data, {"schema_version", "routes", "models", "resources"}, f"catalog {path}"
    )
    try:
        _schema_version(data.get("schema_version"), MMO_SCHEMA_VERSION, "catalog")
    except ValueError as exc:
        raise ValueError(f"unsupported catalog schema_version in {path}") from exc
    result: dict[str, Any] = {}
    for section in ("routes", "models", "resources"):
        value = data.get(section, {})
        if not isinstance(value, Mapping):
            raise ValueError(f"catalog {section!r} must be a table in {path}")
        if not all(isinstance(item, Mapping) for item in value.values()):
            raise ValueError(f"catalog {section!r} entries must be tables in {path}")
        result[section] = value
    return result


def load_global_catalog() -> dict[str, Any]:
    """Load the bundled catalog plus deterministic user overlays.

    User fragments are merged lexicographically so an operator can override or
    extend a provider/model without editing installed files. Profile-local
    fragments are applied separately by :func:`load_catalog`.
    """

    catalog = _load_catalog_fragment(install_root() / "config" / "catalog.toml")
    root = user_catalog_root()
    if root.is_dir():
        for path in sorted(root.glob("*.toml")):
            catalog = deep_merge(catalog, _load_catalog_fragment(path))
    return catalog


def validate_catalog_data(catalog: Mapping[str, Any], *, label: str = "catalog") -> dict[str, Any]:
    """Validate one fully merged catalog and all cross-section references."""

    if not isinstance(catalog, Mapping):
        raise ValueError(f"{label} must be a table")
    _reject_unknown_fields(catalog, {"schema_version", "routes", "models", "resources"}, label)
    if "schema_version" in catalog:
        _schema_version(catalog["schema_version"], MMO_SCHEMA_VERSION, label)
    sections: dict[str, Mapping[str, Any]] = {}
    for section in ("routes", "models", "resources"):
        value = catalog.get(section, {})
        if not isinstance(value, Mapping):
            raise ValueError(f"{label}.{section} must be a table")
        if not all(isinstance(item, Mapping) for item in value.values()):
            raise ValueError(f"{label}.{section} entries must be tables")
        sections[section] = value
    routes = {key: _validate_route(key, value) for key, value in sections["routes"].items()}
    resources = _validate_resources(sections["resources"])
    models = _validate_models(sections["models"], routes)
    for key, route in routes.items():
        resource = route.get("resource_group")
        if resource is not None and resource not in resources:
            raise ValueError(f"route {key}: unknown resource group {resource!r}")
    for key, model in models.items():
        resource = model.get("resource_group")
        if resource is not None and resource not in resources:
            raise ValueError(f"model {key}: unknown resource group {resource!r}")
    return {"routes": routes, "models": models, "resources": resources}


def validated_global_catalog() -> dict[str, Any]:
    """Return the merged global catalog after full semantic validation."""

    return validate_catalog_data(load_global_catalog(), label="global catalog")


def load_catalog(profile_dir: Path, profile_data: Mapping[str, Any]) -> dict[str, Any]:
    catalog = load_global_catalog()
    profile_catalog = profile_data.get("catalog", "catalog.toml")
    if not isinstance(profile_catalog, str) or not profile_catalog:
        raise ValueError("profile.catalog must be a non-empty relative path string")
    if Path(profile_catalog).is_absolute():
        raise ValueError("profile.catalog must be a non-empty relative path string")
    path = (profile_dir / profile_catalog).resolve()
    if profile_dir not in path.parents:
        raise ValueError("profile catalog path escapes the profile pack")
    if path.is_file():
        catalog = deep_merge(catalog, _load_catalog_fragment(path))
    elif "catalog" in profile_data:
        raise ValueError(f"explicit profile catalog file is missing: {profile_catalog}")
    return catalog


def _string_list(value: Any, label: str, *, allow_empty: bool = True) -> list[str]:
    if value is None and allow_empty:
        return []
    if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
        raise ValueError(f"{label} must be an array of strings")
    if any(not item.strip() for item in value):
        raise ValueError(f"{label} cannot contain empty strings")
    if not allow_empty and not value:
        raise ValueError(f"{label} cannot be empty")
    if len(value) != len(set(value)):
        raise ValueError(f"{label} cannot contain duplicates")
    return list(value)


def _positive_int(value: Any, label: str, minimum: int = 1, maximum: int = 1_000_000) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or not minimum <= value <= maximum:
        raise ValueError(f"{label} must be an integer between {minimum} and {maximum}")
    return value


_HEADER_NAME = re.compile(r"[!#$%&'*+.^_`|~0-9A-Za-z-]+")
_ENVIRONMENT_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")


def _validate_http_base_url(value: Any, label: str) -> str:
    """Validate an HTTP API root before downstream endpoint concatenation."""

    if not isinstance(value, str) or not valid_absolute_uri(value):
        raise ValueError(f"{label}: valid HTTP(S) base_url is required")
    try:
        parsed = urlsplit(value)
        # Accessing ``port`` performs the range and syntax checks that
        # ``urlsplit`` intentionally defers.
        _parsed_port = parsed.port
    except ValueError as exc:
        raise ValueError(f"{label}: valid HTTP(S) base_url is required") from exc
    if (
        parsed.scheme.lower() not in {"http", "https"}
        or parsed.hostname is None
        or parsed.username is not None
        or parsed.password is not None
        # Codex 0.149 trims slashes before appending the endpoint path, while
        # Switchyard 0.2.0 also normalizes known endpoint suffixes.  Neither
        # treats a query or fragment embedded in the base as a URI join input.
        or "?" in value
        or "#" in value
    ):
        raise ValueError(f"{label}: valid HTTP(S) base_url is required")
    return value


def _header_map(value: Any, label: str, *, environment_values: bool = False) -> dict[str, str]:
    if not isinstance(value, Mapping):
        raise ValueError(f"{label} must be a string-to-string table")
    result: dict[str, str] = {}
    seen: set[str] = set()
    for raw_name, raw_value in value.items():
        if not isinstance(raw_name, str) or not _HEADER_NAME.fullmatch(raw_name):
            raise ValueError(f"{label}: invalid HTTP header name {raw_name!r}")
        normalized_name = raw_name.lower()
        if normalized_name in seen:
            raise ValueError(f"{label}: duplicate case-insensitive HTTP header name {raw_name!r}")
        seen.add(normalized_name)
        if not isinstance(raw_value, str):
            raise ValueError(f"{label}.{raw_name} must be a string")
        if environment_values:
            if not _ENVIRONMENT_NAME.fullmatch(raw_value):
                raise ValueError(
                    f"{label}.{raw_name}: invalid environment variable name {raw_value!r}"
                )
        elif not valid_http_header_value(raw_value):
            raise ValueError(f"{label}.{raw_name} contains a prohibited control character")
        result[raw_name] = raw_value
    return result


def _json_compatible_value(value: Any, label: str) -> Any:
    """Validate a value destined for Switchyard's serde_json::Value map."""

    if isinstance(value, (str, bool)):
        return value
    if isinstance(value, int) and not isinstance(value, bool):
        if -(2**63) <= value <= 2**63 - 1:
            return value
        raise ValueError(f"{label} integer is outside the TOML/JSON target range")
    if isinstance(value, float):
        if math.isfinite(value):
            return value
        raise ValueError(f"{label} must not contain a non-finite number")
    if isinstance(value, list):
        return [
            _json_compatible_value(item, f"{label}[{index}]") for index, item in enumerate(value)
        ]
    if isinstance(value, Mapping):
        result: dict[str, Any] = {}
        for key, item in value.items():
            if not isinstance(key, str):
                raise ValueError(f"{label} object keys must be strings")
            result[key] = _json_compatible_value(item, f"{label}.{key}")
        return result
    raise ValueError(f"{label} must contain only JSON-compatible TOML values")


def _credential_envs(route: Mapping[str, Any], label: str) -> list[str]:
    """Validate a route's ordered credential environment names."""

    raw = route.get("credential_envs", [])
    values = _string_list(raw, f"{label}.credential_envs")
    for value in values:
        if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", value):
            raise ValueError(f"{label}: invalid credential environment variable {value!r}")
    return list(dict.fromkeys(values))


def _validate_openrouter_policy(value: Any, label: str) -> dict[str, Any] | None:
    if value is None:
        return None
    if not isinstance(value, Mapping):
        raise ValueError(f"{label} must be a table")
    _reject_unknown_fields(value, OPENROUTER_POLICY_FIELDS, label)
    result = dict(value)
    for field in ("only", "order", "quantizations"):
        if field in result:
            result[field] = _string_list(result[field], f"{label}.{field}", allow_empty=False)
    for field in ("allow_fallbacks", "require_parameters", "zdr"):
        if field in result and not isinstance(result[field], bool):
            raise ValueError(f"{label}.{field} must be boolean")
    if "data_collection" in result:
        result["data_collection"] = _enum_string(
            result["data_collection"], {"allow", "deny"}, f"{label}.data_collection"
        )
    if "sort" in result:
        result["sort"] = _enum_string(
            result["sort"], {"price", "throughput", "latency"}, f"{label}.sort"
        )
    if "max_price" in result:
        price = result["max_price"]
        if not isinstance(price, Mapping):
            raise ValueError(f"{label}.max_price must be a table")
        unknown = sorted(set(price) - {"prompt", "completion", "request", "image"})
        if unknown:
            raise ValueError(f"{label}.max_price has unknown fields: {', '.join(unknown)}")
        normalized: dict[str, float] = {}
        for field, raw in price.items():
            if not isinstance(raw, (int, float)) or isinstance(raw, bool) or raw < 0:
                raise ValueError(f"{label}.max_price.{field} must be non-negative and finite")
            try:
                numeric = float(raw)
            except OverflowError:
                raise ValueError(
                    f"{label}.max_price.{field} must be non-negative and finite"
                ) from None
            if not math.isfinite(numeric):
                raise ValueError(f"{label}.max_price.{field} must be non-negative and finite")
            normalized[str(field)] = numeric
        result["max_price"] = normalized
    return result


def _validate_route(key: str, route: Mapping[str, Any]) -> dict[str, Any]:
    validate_id(key, "route id")
    _reject_unknown_fields(route, ROUTE_FIELDS, f"route {key}")
    driver = _enum_string(route.get("driver"), ALLOWED_DRIVERS, f"route {key}.driver")
    unsupported = sorted(set(route) - ROUTE_COMMON_FIELDS - ROUTE_DRIVER_FIELDS[driver])
    if unsupported:
        raise ValueError(f"route {key}: fields unsupported by driver {driver!r}: {unsupported}")
    result = dict(route)
    result.setdefault("name", key)
    result["name"] = _nonempty_string(result["name"], f"route {key}.name")
    for field in ("api_operator", "access_product"):
        result[field] = _nonempty_string(result.get(field), f"route {key}.{field}")
    result["wire_protocol"] = _enum_string(
        result.get("wire_protocol"), ALLOWED_WIRE_PROTOCOLS, f"route {key}.wire_protocol"
    )
    result["billing_mode"] = _enum_string(
        result.get("billing_mode"), ALLOWED_BILLING_MODES, f"route {key}.billing_mode"
    )
    expected_protocols = {
        "switchyard": {"openai_chat", "openai_responses", "anthropic_messages"},
        "codex_custom": {"openai_responses"},
        "codex_builtin": {"codex_builtin"},
        "codex_oss": {"codex_oss"},
        "catalog_only": {"catalog_only", "openai_chat", "openai_responses", "anthropic_messages"},
    }[driver]
    if result["wire_protocol"] not in expected_protocols:
        raise ValueError(f"route {key}.wire_protocol is incompatible with driver {driver!r}")
    result["openrouter_policy"] = _validate_openrouter_policy(
        result.get("openrouter_policy"), f"route {key}.openrouter_policy"
    )
    if result["openrouter_policy"] is not None and result["api_operator"] != "openrouter":
        raise ValueError(f"route {key}: openrouter_policy requires api_operator='openrouter'")
    result.setdefault("transport_modalities", ["text"])
    result["transport_modalities"] = _string_list(
        result["transport_modalities"], f"route {key}.transport_modalities", allow_empty=False
    )
    unknown_modalities = sorted(set(result["transport_modalities"]) - ALLOWED_MODALITIES)
    if unknown_modalities:
        raise ValueError(f"route {key}: invalid transport modalities {unknown_modalities}")
    result.setdefault(
        "transport_output_modalities",
        ["text", "image", "video", "audio", "file"] if driver == "catalog_only" else ["text"],
    )
    result["transport_output_modalities"] = _string_list(
        result["transport_output_modalities"],
        f"route {key}.transport_output_modalities",
        allow_empty=False,
    )
    unknown_output_modalities = sorted(
        set(result["transport_output_modalities"]) - ALLOWED_MODALITIES
    )
    if unknown_output_modalities:
        raise ValueError(f"route {key}: invalid output modalities {unknown_output_modalities}")
    result.setdefault("tool_calling", driver != "catalog_only")
    result.setdefault("parallel_tool_calls", driver != "catalog_only")
    result.setdefault(
        "preserves_tool_media",
        driver in {"codex_builtin", "codex_custom"},
    )
    result.setdefault(
        "tool_result_modalities",
        list(result["transport_modalities"]) if result["preserves_tool_media"] else ["text"],
    )
    result["tool_result_modalities"] = _string_list(
        result["tool_result_modalities"],
        f"route {key}.tool_result_modalities",
        allow_empty=False,
    )
    unknown_tool_modalities = sorted(set(result["tool_result_modalities"]) - ALLOWED_MODALITIES)
    if unknown_tool_modalities:
        raise ValueError(f"route {key}: invalid tool-result modalities {unknown_tool_modalities}")
    result.setdefault("supports_documents", "file" in result["transport_modalities"])
    for boolean_field in (
        "tool_calling",
        "parallel_tool_calls",
        "preserves_tool_media",
        "supports_documents",
    ):
        if not isinstance(result[boolean_field], bool):
            raise ValueError(f"route {key}.{boolean_field} must be boolean")
    if result["parallel_tool_calls"] and not result["tool_calling"]:
        raise ValueError(f"route {key}: parallel_tool_calls requires tool_calling")
    if not result["preserves_tool_media"] and set(result["tool_result_modalities"]) - {"text"}:
        raise ValueError(
            f"route {key}: non-text tool_result_modalities require preserves_tool_media=true"
        )
    result.setdefault("resource_group", None)
    result.setdefault("inventory", None)
    for field in ("resource_group", "inventory"):
        result[field] = _optional_nonempty_string(result[field], f"route {key}.{field}")
    credentials = _credential_envs(result, f"route {key}")
    result["credential_envs"] = credentials
    if driver == "switchyard":
        if "extra_headers" in result:
            result["extra_headers"] = _header_map(
                result["extra_headers"], f"route {key}.extra_headers"
            )
        result["base_url"] = _validate_http_base_url(result.get("base_url"), f"route {key}")
        result.setdefault("max_retries", 1)
        result["max_retries"] = _positive_int(
            result["max_retries"], f"route {key}.max_retries", 0, 10
        )
    elif driver == "codex_custom":
        if "http_headers" in result:
            result["http_headers"] = _header_map(
                result["http_headers"], f"route {key}.http_headers"
            )
        if "env_http_headers" in result:
            result["env_http_headers"] = _header_map(
                result["env_http_headers"],
                f"route {key}.env_http_headers",
                environment_values=True,
            )
        static_names = {name.lower() for name in result.get("http_headers", {})}
        environment_names = {name.lower() for name in result.get("env_http_headers", {})}
        overlap = sorted(static_names & environment_names)
        if overlap:
            raise ValueError(
                f"route {key} defines headers in both http_headers and env_http_headers: "
                + ", ".join(overlap)
            )
        result["base_url"] = _validate_http_base_url(result.get("base_url"), f"route {key}")
        result.setdefault("wire_api", "responses")
        if not isinstance(result["wire_api"], str) or result["wire_api"] != "responses":
            raise ValueError(f"route {key}: Codex custom routes require wire_api='responses'")
        for field, default, minimum, maximum in (
            ("request_max_retries", 1, 0, 2**63 - 1),
            ("stream_max_retries", 1, 0, 2**63 - 1),
            ("stream_idle_timeout_ms", 600_000, 1_000, 2**63 - 1),
        ):
            result.setdefault(field, default)
            result[field] = _positive_int(result[field], f"route {key}.{field}", minimum, maximum)
    elif driver == "codex_builtin":
        provider_id = result.get("provider_id")
        provider_id = _nonempty_string(provider_id, f"route {key}.provider_id")
        if provider_id not in {"openai", "amazon-bedrock"}:
            raise ValueError(
                f"route {key}: Codex built-in provider_id must be openai or amazon-bedrock"
            )
        result["provider_id"] = provider_id
        result.setdefault("auth", "chatgpt" if provider_id == "openai" else "builtin")
        expected_auth = "chatgpt" if provider_id == "openai" else "builtin"
        if result.get("auth") != expected_auth:
            raise ValueError(
                f"route {key}: auth must be {expected_auth!r} for built-in {provider_id!r}"
            )
        if result.get("auth_link_mode") is not None and not isinstance(
            result.get("auth_link_mode"), str
        ):
            raise ValueError(f"route {key}: invalid auth_link_mode")
        if result.get("auth_link_mode") not in {None, "shared", "copy", "none"}:
            raise ValueError(f"route {key}: invalid auth_link_mode")
    elif driver == "codex_oss":
        provider_id = result.get("provider_id")
        if not isinstance(provider_id, str) or provider_id not in {"ollama", "lmstudio"}:
            raise ValueError(f"route {key}: codex_oss provider_id must be ollama or lmstudio")
    elif driver == "catalog_only":
        if "base_url" in result:
            result["base_url"] = _validate_http_base_url(result["base_url"], f"route {key}")
        for field in ("wire_api", "auth"):
            if field in result:
                result[field] = _nonempty_string(result[field], f"route {key}.{field}")
    header_fields = {"http_headers", "env_http_headers", "extra_headers"}
    allowed_header_fields = {
        "switchyard": {"extra_headers"},
        "codex_custom": {"http_headers", "env_http_headers"},
    }.get(driver, set())
    unsupported_headers = sorted((header_fields & set(result)) - allowed_header_fields)
    if unsupported_headers:
        detail = f"header fields are unsupported by driver {driver}"
        raise ValueError(f"route {key}: {detail}: {unsupported_headers}")
    return result


def validate_model_entry(
    key: str, model: Mapping[str, Any], routes: Mapping[str, Any]
) -> dict[str, Any]:
    validate_id(key, "model id")
    _reject_unknown_fields(model, MODEL_FIELDS, f"model {key}")
    route_key = model.get("route")
    if not isinstance(route_key, str) or route_key not in routes:
        raise ValueError(f"model {key}: unknown route {route_key!r}")
    if not key.startswith(f"{route_key}__"):
        raise ValueError(
            f"model {key}: catalog key must start with exact route namespace "
            f"{route_key!r} followed by '__'"
        )
    if not isinstance(model.get("upstream_id"), str) or not model["upstream_id"].strip():
        raise ValueError(f"model {key}: upstream_id is required")
    if not isinstance(model.get("maker"), str) or not model["maker"].strip():
        raise ValueError(f"model {key}: maker is required")
    result = dict(model)
    result.setdefault("display_name", key)
    result.setdefault("description", "")
    result.setdefault("kind", "chat")
    result["display_name"] = _nonempty_string(result["display_name"], f"model {key}.display_name")
    if not isinstance(result["description"], str):
        raise ValueError(f"model {key}.description must be a string")
    result["kind"] = _enum_string(result["kind"], ALLOWED_MODEL_KINDS, f"model {key}.kind")
    result.setdefault(
        "agent_compatible",
        result["kind"] in {"chat", "vision_chat"} and routes[route_key]["driver"] != "catalog_only",
    )
    if not isinstance(result["agent_compatible"], bool):
        raise ValueError(f"model {key}.agent_compatible must be boolean")
    result.setdefault("context_window", 131072 if result["agent_compatible"] else 0)
    result["context_window"] = _positive_int(
        result["context_window"],
        f"model {key}.context_window",
        1024 if result["agent_compatible"] else 0,
        20_000_000,
    )
    result.setdefault("reasoning_levels", ["none", "low", "medium", "high"])
    result["reasoning_levels"] = _string_list(
        result["reasoning_levels"], f"model {key}.reasoning_levels", allow_empty=False
    )
    unknown_reasoning = sorted(set(result["reasoning_levels"]) - ALLOWED_REASONING)
    if unknown_reasoning:
        raise ValueError(f"model {key}: invalid reasoning levels {unknown_reasoning}")
    result.setdefault("default_reasoning", result["reasoning_levels"][-1])
    if (
        not isinstance(result["default_reasoning"], str)
        or result["default_reasoning"] not in result["reasoning_levels"]
    ):
        raise ValueError(f"model {key}: default_reasoning is not supported")
    route = routes[route_key]
    result["route_policy"] = _validate_openrouter_policy(
        result.get("route_policy"), f"model {key}.route_policy"
    )
    if result["route_policy"] is not None and route["api_operator"] != "openrouter":
        raise ValueError(f"model {key}: route_policy requires an OpenRouter route")
    if "extra_body" in result:
        if route["driver"] != "switchyard":
            raise ValueError(f"model {key}: extra_body is supported only by Switchyard targets")
        if not isinstance(result["extra_body"], Mapping):
            raise ValueError(f"model {key}.extra_body must be a table")
        result["extra_body"] = _json_compatible_value(
            result["extra_body"], f"model {key}.extra_body"
        )
    result.setdefault("modalities", route.get("transport_modalities", ["text"]))
    result["modalities"] = _string_list(
        result["modalities"], f"model {key}.modalities", allow_empty=False
    )
    unknown_modalities = sorted(set(result["modalities"]) - ALLOWED_MODALITIES)
    if unknown_modalities:
        raise ValueError(f"model {key}: invalid modalities {unknown_modalities}")
    missing_input_transport = sorted(set(result["modalities"]) - set(route["transport_modalities"]))
    if missing_input_transport:
        raise ValueError(
            f"model {key}: route transport cannot carry input modalities {missing_input_transport}"
        )
    result.setdefault("output_modalities", ["text"])
    result["output_modalities"] = _string_list(
        result["output_modalities"], f"model {key}.output_modalities", allow_empty=False
    )
    unknown_output_modalities = sorted(set(result["output_modalities"]) - ALLOWED_MODALITIES)
    if unknown_output_modalities:
        raise ValueError(f"model {key}: invalid output modalities {unknown_output_modalities}")
    missing_output_transport = sorted(
        set(result["output_modalities"]) - set(route["transport_output_modalities"])
    )
    if missing_output_transport:
        raise ValueError(
            f"model {key}: route transport cannot carry output modalities "
            f"{missing_output_transport}"
        )
    result.setdefault(
        "supports_tool_images",
        "image" in result["modalities"]
        and route.get("preserves_tool_media", False)
        and "image" in route.get("tool_result_modalities", []),
    )
    result.setdefault(
        "supports_documents",
        "file" in result["modalities"] and route.get("supports_documents", False),
    )
    for boolean_field in ("supports_tool_images", "supports_documents"):
        if not isinstance(result[boolean_field], bool):
            raise ValueError(f"model {key}.{boolean_field} must be boolean")
    if result["supports_tool_images"] and "image" not in result["modalities"]:
        raise ValueError(f"model {key}: supports_tool_images requires image input")
    if result["supports_documents"] and "file" not in result["modalities"]:
        raise ValueError(f"model {key}: supports_documents requires file input")
    result.setdefault("tool_calling", route.get("tool_calling", True))
    # OpenAI Responses distinguishes schema-defined function tools from
    # free-form custom tools.  Some compatible endpoints implement the former
    # but reject the latter.  Keep that narrower capability independent so a
    # model can still use shell and MCP function tools without receiving
    # Codex's free-form apply_patch tool.
    result.setdefault("supports_custom_tools", result["tool_calling"])
    result.setdefault("parallel_tool_calls", route.get("parallel_tool_calls", True))
    result.setdefault("supports_reasoning_summaries", False)
    result.setdefault("structured_output", False)
    for boolean_field in (
        "tool_calling",
        "supports_custom_tools",
        "parallel_tool_calls",
        "supports_reasoning_summaries",
        "structured_output",
    ):
        if not isinstance(result[boolean_field], bool):
            raise ValueError(f"model {key}.{boolean_field} must be boolean")
    if result["tool_calling"] and not route["tool_calling"]:
        raise ValueError(f"model {key}: tool_calling exceeds route transport capability")
    if result["supports_custom_tools"] and not result["tool_calling"]:
        raise ValueError(f"model {key}: supports_custom_tools requires tool_calling")
    if result["parallel_tool_calls"] and not result["tool_calling"]:
        raise ValueError(f"model {key}: parallel_tool_calls requires tool_calling")
    if result["parallel_tool_calls"] and not route["parallel_tool_calls"]:
        raise ValueError(f"model {key}: parallel_tool_calls exceeds route transport capability")
    result.setdefault("max_output_tokens", None)
    if result["max_output_tokens"] is not None:
        result["max_output_tokens"] = _positive_int(
            result["max_output_tokens"], f"model {key}.max_output_tokens", 1, 20_000_000
        )
    result.setdefault("availability", "current")
    result.setdefault("capability_confidence", "documented")
    result.setdefault("source", "bundled-catalog")
    for field in ("availability", "capability_confidence", "source"):
        result[field] = _nonempty_string(result[field], f"model {key}.{field}")
    for field in (
        "availability_source",
        "capability_source",
        "pricing_source",
        "inventory",
        "resource_group",
    ):
        if field in result:
            result[field] = _optional_nonempty_string(result[field], f"model {key}.{field}")
    result.setdefault("resource_group", route.get("resource_group"))
    for field in (
        "input_cost_per_million",
        "cached_input_cost_per_million",
        "cache_write_input_cost_per_million",
        "output_cost_per_million",
        "unit_cost_usd",
    ):
        if field not in result:
            continue
        value = result[field]
        if (
            not isinstance(value, (int, float))
            or isinstance(value, bool)
            or (isinstance(value, float) and not math.isfinite(value))
            or value < 0
        ):
            raise ValueError(f"model {key}.{field} must be a non-negative finite number")
    return result


def _validate_models(
    values: Mapping[str, Any], routes: Mapping[str, Any]
) -> dict[str, dict[str, Any]]:
    """Validate model rows and reject duplicate route/upstream identities."""

    result: dict[str, dict[str, Any]] = {}
    identities: dict[tuple[str, str], str] = {}
    for key, value in values.items():
        model = validate_model_entry(key, value, routes)
        identity = (str(model["route"]), str(model["upstream_id"]))
        previous = identities.get(identity)
        if previous is not None:
            raise ValueError(
                f"models {previous!r} and {key!r} duplicate route/upstream binding {identity!r}"
            )
        identities[identity] = key
        result[key] = model
    return result


def _validate_resource(key: str, resource: Mapping[str, Any]) -> dict[str, Any]:
    validate_id(key, "resource group id")
    _reject_unknown_fields(resource, RESOURCE_FIELDS, f"resource {key}")
    result = dict(resource)
    result.setdefault("description", "")
    if not isinstance(result["description"], str):
        raise ValueError(f"resource {key}.description must be a string")
    result.setdefault("lock_key", key)
    if not isinstance(result["lock_key"], str) or not result["lock_key"]:
        raise ValueError(f"resource {key}: lock_key is required")
    result.setdefault("max_active", 4)
    result["max_active"] = _positive_int(
        result["max_active"], f"resource {key}.max_active", 1, 1024
    )
    return result


def _validate_resources(values: Mapping[str, Any]) -> dict[str, dict[str, Any]]:
    result = {key: _validate_resource(key, value) for key, value in values.items()}
    capacities: dict[str, tuple[str, int]] = {}
    for key, resource in result.items():
        lock_key = str(resource["lock_key"])
        maximum = int(resource["max_active"])
        previous = capacities.get(lock_key)
        if previous is not None and previous[1] != maximum:
            raise ValueError(
                f"resources {previous[0]!r} and {key!r} share lock_key {lock_key!r} "
                f"with conflicting max_active values {previous[1]} and {maximum}"
            )
        capacities[lock_key] = (key, maximum)
    return result