"""Codex model discovery from API, local cache, and config.""" from __future__ import annotations import json import logging import os from pathlib import Path from typing import List, Optional from hermes_cli.route_identity import normalize_route_base_url logger = logging.getLogger(__name__) # Curated offline fallback (first-run, transient API failure). Only slugs the ChatGPT Codex # OAuth backend actually accepts: the public API's "-pro" variants and the retired # gpt-5.3-codex / gpt-5.2-codex / gpt-5.1-codex-max / gpt-5.1-codex-mini return HTTP 400 there # ("not supported when using Codex with a ChatGPT account"), so listing them leaked dead picker # choices (#52492). If OpenAI re-enables any, live discovery (_fetch_models_from_api) picks them # up automatically. DEFAULT_CODEX_MODELS: List[str] = [ "gpt-6-sol", "gpt-6-luna", "gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna", "gpt-5.5", "gpt-5.4-mini", "gpt-5.4", # Research preview exposed ONLY via the Codex OAuth backend for ChatGPT Pro subscribers — # not in the public API, so it stays out of the "openai" catalog in hermes_cli/models.py. # The backend reports ``supported_in_api: false`` for it; that flag describes API # availability, not Codex availability, so fetch/cache paths must not filter on it. "gpt-5.3-codex-spark"] # gpt-5.3-codex-spark is in research preview and is exposed *only* via the Codex CLI / OAuth backend # (chatgpt.com/backend-api/codex/models) for ChatGPT Pro subscribers. It is NOT available in the public # OpenAI API, so it intentionally stays out of the "openai" provider catalog in hermes_cli/models.py — only # the openai-codex (OAuth) provider surfaces it. The Codex backend reports ``supported_in_api: false`` for # this slug; that flag describes API availability, not Codex backend availability, so the fetch/cache code # paths below intentionally do not filter on it. PR #12994 removed this entry on the assumption it was # unsupported — that was wrong; restored here. Keep it in the curated fallback so Pro users still see Spark # in `/model` when live discovery is unavailable (offline first run, transient API failure). _FORWARD_COMPAT_TEMPLATE_MODELS: List[tuple[str, tuple[str, ...]]] = [ ("gpt-6-sol", ("gpt-5.6-sol", "gpt-5.5")), ("gpt-6-luna", ("gpt-5.6-luna", "gpt-5.5")), ("gpt-5.6-sol", ("gpt-5.5", "gpt-5.4")), ("gpt-5.6-terra", ("gpt-5.5", "gpt-5.4")), ("gpt-5.6-luna", ("gpt-5.5", "gpt-5.4")), ("gpt-5.5", ("gpt-5.4", "gpt-5.4-mini")), # Spark surfaces whenever a compatible template is present; the backend (not Hermes) # gates real availability by ChatGPT Pro entitlement. ("gpt-5.3-codex-spark", ("gpt-5.4", "gpt-5.5"))] def _dedupe(model_ids) -> List[str]: """Order-preserving dedupe.""" return list(dict.fromkeys(model_ids)) def _add_forward_compat_models(model_ids: List[str]) -> List[str]: """Surface newer Codex slugs missing from live discovery when an older compatible template is present (Clawdbot-style synthetic forward-compat catalog).""" ordered = _dedupe(model_ids) seen = set(ordered) for synthetic_model, template_models in _FORWARD_COMPAT_TEMPLATE_MODELS: if synthetic_model not in seen and any(template in seen for template in template_models): ordered.append(synthetic_model) seen.add(synthetic_model) return ordered def _add_context_variants(model_ids: List[str]) -> List[str]: """Insert ``-900k`` large-context picker variants after eligible base slugs. Base slugs keep the cheaper advertised 272K limit; the variant opts into the large window. The suffix is Hermes-side only — stripped before the id hits the wire (agent/transports/codex.py, agent/auxiliary_client.py). """ from agent.model_metadata import CODEX_CONTEXT_VARIANT_SUFFIX, has_codex_context_variant out: List[str] = [] present = set(model_ids) for model_id in model_ids: out.append(model_id) variant = model_id + CODEX_CONTEXT_VARIANT_SUFFIX if variant in present or variant in out: continue if has_codex_context_variant(model_id): out.append(variant) return out def _finalize_codex_models(model_ids: List[str]) -> List[str]: """Forward-compat synthesis + large-context variant synthesis.""" return _add_context_variants(_add_forward_compat_models(model_ids)) def _drop_undiscovered_astra(model_ids: List[str]) -> List[str]: """Astra is account-gated: only the live account-scoped catalog may advertise it. A stale ``models_cache.json`` or a ``config.toml`` default is a compatibility hint, not entitlement.""" from agent.reasoning_effort import is_astra_model return [model for model in model_ids if not is_astra_model(model)] def codex_catalog_credential_identity() -> str: """Identity of the credential live discovery would use right now, for the catalog cache key. Access/refresh tokens rotate in place while the account-scoped catalog stays authoritative for the same ChatGPT principal and route, so the key includes the principal and resolved base URL, not the rotating token. An expired token is its own state (per route): ``_codex_catalog`` serves the static fallback for it, and that fallback must not outlive the refresh under the healthy principal's key. Opaque non-JWT tokens fall back to the token itself (the caller hashes every part before anything is persisted). """ from hermes_cli.auth import _codex_access_token_is_expiring, resolve_codex_runtime_credentials try: creds = resolve_codex_runtime_credentials(read_only=True) token = str(creds.get("api_key") or "") route = normalize_route_base_url(str(creds.get("base_url") or "").strip()) except Exception: # AuthError (no/exhausted creds) or the pytest seat belt: no live catalog either way token = "" if not token: return "missing" if _codex_access_token_is_expiring(token, 0): return "expired\n" + route from agent.credential_pool import _codex_principal_identity principal = _codex_principal_identity(token) return ("/".join(principal) if principal else token) + "\n" + route def _ranked_slugs(entries: object) -> List[str]: """Visible slugs from a Codex catalog ``models`` list, sorted by (priority, slug), deduped. Does not filter on ``supported_in_api``: that flag describes the public OpenAI API, while the OAuth-backed Codex backend still accepts slugs marked false there (gpt-5.3-codex-spark). """ sortable = [] for item in entries: if not isinstance(item, dict): continue slug = item.get("slug") if not isinstance(slug, str) or not slug.strip(): continue visibility = item.get("visibility") if isinstance(visibility, str) and visibility.strip().lower() in {"hide", "hidden"}: continue priority = item.get("priority") rank = int(priority) if isinstance(priority, (int, float)) else 10_000 sortable.append((rank, slug.strip())) sortable.sort() return _dedupe(slug for _, slug in sortable) def _fetch_models_from_api(access_token: str, base_url: Optional[str] = None) -> List[str]: """Fetch available models from the Codex API. Returns visible models sorted by priority. ``base_url`` is the host the credential is routed to (resolved together with it); the catalog is fetched there, never from a host the credential does not belong to (#121486). """ try: from agent.model_metadata import _codex_catalog_probe_allowed from hermes_cli.auth_codex import _codex_base_url catalog_base = (base_url or "").strip().rstrip("/") or _codex_base_url() if not _codex_catalog_probe_allowed(access_token, catalog_base): return [] import httpx # The per-account catalog needs ChatGPT-Account-ID (else ``{"models":[]}`` with HTTP 200 # masquerades as "no models") and, for residency-enforced workspaces, the residency header. from agent.codex_headers import codex_account_headers headers = {"Authorization": f"Bearer {access_token}", **codex_account_headers(access_token)} from agent.model_metadata import fetch_codex_catalog_entries entries, _status = fetch_codex_catalog_entries( lambda url: httpx.get(url, headers=headers, timeout=10), base_url=catalog_base) except Exception as exc: logger.debug("Failed to fetch Codex models from API: %s", exc) return [] return _finalize_codex_models(_ranked_slugs(entries)) def _read_default_model(codex_home: Path) -> Optional[str]: config_path = codex_home / "config.toml" if not config_path.exists(): return None try: import tomllib import tomllib payload = tomllib.loads(config_path.read_text(encoding="utf-8-sig")) except Exception: return None model = payload.get("model") if isinstance(payload, dict) else None return model.strip() if isinstance(model, str) and model.strip() else None def _read_cache_models(codex_home: Path) -> List[str]: cache_path = codex_home / "models_cache.json" if not cache_path.exists(): return [] try: raw = json.loads(cache_path.read_text(encoding="utf-8-sig")) except Exception: return [] entries = raw.get("models") if isinstance(raw, dict) else None return _ranked_slugs(entries if isinstance(entries, list) else []) def get_codex_model_ids(access_token: Optional[str] = None, base_url: Optional[str] = None) -> List[str]: """Available Codex model IDs: live API (if token) > config.toml default > local cache > defaults. Pass the ``base_url`` resolved together with ``access_token`` (runtime/pool route) so live discovery asks the credential's own host. Without a live answer the result is ``CuratedFallbackModels``: a cache placeholder that never replaces a verified catalog, since account-gated rows such as Astra are absent from the offline hints.""" codex_home = Path(os.getenv("CODEX_HOME", "").strip() or str(Path.home() / ".codex")).expanduser() if access_token: api_models = _fetch_models_from_api(access_token, base_url=base_url) if api_models: return _finalize_codex_models(api_models) # Late: models_catalog_static builds its codex table from this module at import time. from hermes_cli.models_catalog_static import CuratedFallbackModels default_model = _read_default_model(codex_home) return CuratedFallbackModels(_finalize_codex_models(_drop_undiscovered_astra(_dedupe([ *([default_model] if default_model else []), *_read_cache_models(codex_home), *DEFAULT_CODEX_MODELS]))))