Discover Codex models from latest stable release and account catalog

.env.example

@@ -25,6 +25,8 @@ CHATGPT_LOCAL_EXPOSE_REASONING_MODELS=false
 # Automatically discover account models; refresh stale data without restarting
 CHATGPT_LOCAL_MODEL_SYNC=true
 CHATGPT_LOCAL_MODEL_REFRESH_INTERVAL=3600
+# Optional: override the latest stable Codex release used for model discovery
+# CHATGPT_LOCAL_MODELS_CLIENT_VERSION=0.160.0
 
 # Enable default web search tool
 CHATGPT_LOCAL_ENABLE_WEB_SEARCH=false

DOCKER.md

@@ -34,6 +34,7 @@ Set options in `.env` or pass environment variables:
 - `CHATGPT_LOCAL_ENABLE_WEB_SEARCH`: `true|false` to enable default web search tool
 - `CHATGPT_LOCAL_MODEL_SYNC`: `true|false` to discover account models automatically (default `true`)
 - `CHATGPT_LOCAL_MODEL_REFRESH_INTERVAL`: model catalog refresh interval in seconds (default `3600`)
+- `CHATGPT_LOCAL_MODELS_CLIENT_VERSION`: optional stable Codex release version override; otherwise use the latest stable GitHub release, with a bundled fallback if GitHub is unavailable
 
 ## Logs
 Set `VERBOSE=true` to include extra logging for troubleshooting upstream or chat app requests. Please include and use these logs when submitting bug reports.

README.md

@@ -85,18 +85,13 @@ The server runs at `http://127.0.0.1:8000` by default. Use `http://127.0.0.1:800
 ## Supported Models
 
 ChatMock automatically discovers the models available to the signed-in ChatGPT
-account. The current catalog commonly includes:
-
-- `gpt-6-astra`
-- `gpt-6-sol`
-- `gpt-6-luna`
-- `gpt-5.6-sol`
-- `gpt-5.6-terra`
-- `gpt-5.6-luna`
-- `gpt-5.5`
-- `gpt-5.4`
-- `gpt-5.4-mini`
-- `gpt-5.3-codex-spark`
+account. It checks the latest stable Codex release on GitHub when refreshing the
+catalog and uses that version for the account's model request. An optional
+`CHATGPT_LOCAL_MODELS_CLIENT_VERSION` override accepts a stable release
+version only. If GitHub is unavailable, ChatMock uses the last successful version or
+its bundled verified fallback. The model list and capabilities come from the
+account catalog; `/v1/models` and `/api/tags` return 503 until one is available.
+The catalog refresh interval defaults to one hour.
 
 <br>
 

chatmock/cli.py

@@ -11,6 +11,7 @@
 from .app import create_app
 from .config import CLIENT_ID_DEFAULT
 from .limits import RateLimitWindow, compute_reset_at, load_rate_limit_snapshot
+from .model_registry import ALL_REASONING_EFFORTS
 from .oauth import OAuthHTTPServer, OAuthHandler, REQUIRED_PORT, URL_BASE, run_device_code_login
 from .utils import eprint, get_home_dir, load_chatgpt_tokens, parse_jwt_claims, read_auth_file
 
@@ -293,7 +294,7 @@ def main() -> None:
     )
     p_serve.add_argument(
         "--reasoning-effort",
-        choices=["none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra"],
+        choices=list(ALL_REASONING_EFFORTS),
         default=os.getenv("CHATGPT_LOCAL_REASONING_EFFORT", "medium").lower(),
         help="Reasoning effort level for Responses API (default: medium)",
     )

chatmock/fast_mode.py

@@ -5,24 +5,6 @@
 
 from .model_registry import model_supports_service_tier, normalize_model_name
 
-
-PRIORITY_SUPPORTED_MODELS = frozenset(
-    (
-        "gpt-5.4",
-        "gpt-5.6-sol",
-        "gpt-5.6-terra",
-        "gpt-5.6-luna",
-        "gpt-6-astra",
-        "gpt-6-sol",
-        "gpt-6-luna",
-        "gpt-5.2",
-        "gpt-5.1",
-        "gpt-5",
-        "gpt-5.1-codex",
-        "gpt-5-codex",
-    )
-)
-
 _TRUE_STRINGS = {"1", "true", "yes", "on"}
 _FALSE_STRINGS = {"0", "false", "no", "off"}
 
@@ -39,11 +21,8 @@ def parse_optional_bool(value: Any) -> bool | None:
     return None
 
 
-def supports_priority_service_tier(model: str | None) -> bool:
-    catalog_support = model_supports_service_tier(model, "priority")
-    if catalog_support is not None:
-        return catalog_support
-    return normalize_model_name(model) in PRIORITY_SUPPORTED_MODELS
+def supports_priority_service_tier(model: str | None) -> bool | None:
+    return model_supports_service_tier(model, "priority")
 
 
 @dataclass(frozen=True)
@@ -77,7 +56,7 @@ def resolve_service_tier(
         tier = "priority"
         used_server_default = True
 
-    if tier == "priority" and not supports_priority_service_tier(model):
+    if tier == "priority" and supports_priority_service_tier(model) is False:
         normalized = normalize_model_name(model)
         message = (
             f"Fast mode is not supported for model '{normalized}'. "

chatmock/model_catalog.py

@@ -3,6 +3,7 @@
 import datetime
 import json
 import os
+import re
 import threading
 import time
 from dataclasses import dataclass
@@ -24,9 +25,18 @@
 DEFAULT_REFRESH_INTERVAL_SECONDS = 60 * 60
 FAILED_REFRESH_RETRY_SECONDS = 60
 FETCH_TIMEOUT_SECONDS = 5
+RELEASE_FETCH_TIMEOUT_SECONDS = 2
 MODEL_CACHE_FILE = "chatmock_models_cache.json"
-# Bump only after verifying ChatMock against a newer Codex catalog contract.
 CODEX_MODELS_CLIENT_VERSION = "0.160.0"
+CODEX_RELEASE_URL = "https://api.github.com/repos/openai/codex/releases/latest"
+CLIENT_VERSION_ENV = "CHATGPT_LOCAL_MODELS_CLIENT_VERSION"
+CLIENT_VERSION_PATTERN = re.compile(r"^\d+\.\d+\.\d+$")
+STABLE_RELEASE_TAG_PATTERN = re.compile(r"^rust-v(\d+\.\d+\.\d+)$")
+
+
+def _version_parts(version: str) -> tuple[int, int, int]:
+    major, minor, patch = version.split("-", 1)[0].split(".")
+    return int(major), int(minor), int(patch)
 
 
 @dataclass(frozen=True)
@@ -37,6 +47,8 @@ class CatalogModel:
     priority: int
     visibility: str
     supported_in_api: bool
+    context_window: int | None = None
+    input_modalities: tuple[str, ...] = ()
 
 
 def _now_utc() -> datetime.datetime:
@@ -92,7 +104,15 @@ def _parse_models(value: Any) -> tuple[CatalogModel, ...]:
                 if isinstance(tier, str) and tier.strip():
                     service_tiers.add(tier.strip())
 
+        modalities: list[str] = []
+        raw_modalities = item.get("input_modalities")
+        if isinstance(raw_modalities, list):
+            for modality in raw_modalities:
+                if isinstance(modality, str) and modality.strip() and modality.strip() not in modalities:
+                    modalities.append(modality.strip())
+
         priority = item.get("priority")
+        context_window = item.get("context_window")
         models.append(
             CatalogModel(
                 slug=slug.strip(),
@@ -101,6 +121,10 @@ def _parse_models(value: Any) -> tuple[CatalogModel, ...]:
                 priority=priority if isinstance(priority, int) else 1_000_000,
                 visibility=item.get("visibility") if isinstance(item.get("visibility"), str) else "none",
                 supported_in_api=bool(item.get("supported_in_api", False)),
+                context_window=context_window
+                if isinstance(context_window, int) and not isinstance(context_window, bool) and context_window > 0
+                else None,
+                input_modalities=tuple(modalities),
             )
         )
     return tuple(models)
@@ -121,6 +145,11 @@ def __init__(
         self.refresh_interval_seconds = max(float(refresh_interval_seconds), 0.0)
         self.cache_path = Path(cache_path) if cache_path else Path(get_home_dir()) / MODEL_CACHE_FILE
         self._session = session or requests.Session()
+        override = (os.getenv(CLIENT_VERSION_ENV) or "").strip()
+        if override and not CLIENT_VERSION_PATTERN.fullmatch(override):
+            raise ValueError(f"{CLIENT_VERSION_ENV} must be a Codex version such as 0.160.0")
+        self._version_override = override or None
+        self._client_version = self._version_override or CODEX_MODELS_CLIENT_VERSION
         self._lock = threading.Lock()
         self._models: tuple[CatalogModel, ...] = ()
         self._raw_models: list[dict[str, Any]] = []
@@ -171,7 +200,7 @@ def refresh_if_due(self, *, wait_for_refresh: bool = False) -> None:
                 refresh_active = True
 
         if wait_for_refresh and refresh_active:
-            event.wait(FETCH_TIMEOUT_SECONDS + 1)
+            event.wait(RELEASE_FETCH_TIMEOUT_SECONDS + FETCH_TIMEOUT_SECONDS + 1)
 
     def _is_due_locked(self) -> bool:
         if not self._models or self._fetched_at is None:
@@ -201,12 +230,13 @@ def _fetch_and_apply(self) -> None:
         if not access_token or not account_id:
             return
 
-        response = self._request_models(access_token, account_id)
+        client_version = self._resolve_client_version()
+        response = self._request_models(access_token, account_id, client_version)
         if response.status_code == 401:
             access_token, account_id = get_effective_chatgpt_auth(force_refresh=True)
             if not access_token or not account_id:
                 return
-            response = self._request_models(access_token, account_id)
+            response = self._request_models(access_token, account_id, client_version)
         response.raise_for_status()
 
         payload = response.json()
@@ -223,12 +253,36 @@ def _fetch_and_apply(self) -> None:
             self._fetched_at = fetched_at
             self._etag = etag
             self._account_id = account_id
+            self._client_version = client_version
         self._persist_cache()
 
-    def _request_models(self, access_token: str, account_id: str) -> requests.Response:
+    def _resolve_client_version(self) -> str:
+        if self._version_override:
+            return self._version_override
+        try:
+            response = self._session.get(
+                CODEX_RELEASE_URL,
+                headers={"Accept": "application/vnd.github+json"},
+                timeout=RELEASE_FETCH_TIMEOUT_SECONDS,
+            )
+            response.raise_for_status()
+            release = response.json()
+            if not isinstance(release, dict) or release.get("draft") or release.get("prerelease"):
+                return self._client_version
+            tag = release.get("tag_name")
+            match = STABLE_RELEASE_TAG_PATTERN.fullmatch(tag) if isinstance(tag, str) else None
+            if match and _version_parts(match.group(1)) > _version_parts(self._client_version):
+                return match.group(1)
+        except (requests.RequestException, ValueError, TypeError):
+            pass
+        return self._client_version
+
+    def _request_models(
+        self, access_token: str, account_id: str, client_version: str
+    ) -> requests.Response:
         return self._session.get(
             f"{CHATGPT_CODEX_BASE_URL}/models",
-            params={"client_version": CODEX_MODELS_CLIENT_VERSION},
+            params={"client_version": client_version},
             headers={
                 "Authorization": f"Bearer {access_token}",
                 "Accept": "application/json",
@@ -248,8 +302,6 @@ def _load_cache(self) -> None:
             return
         if not isinstance(payload, dict):
             return
-        if payload.get("client_version") != CODEX_MODELS_CLIENT_VERSION:
-            return
 
         cached_account_id = payload.get("account_id")
         current_account_id = _account_id_from_auth_file()
@@ -260,14 +312,27 @@ def _load_cache(self) -> None:
         ):
             return
 
+        cached_version = payload.get("client_version")
+        if (
+            not self._version_override
+            and isinstance(cached_version, str)
+            and STABLE_RELEASE_TAG_PATTERN.fullmatch(f"rust-v{cached_version}")
+            and _version_parts(cached_version) > _version_parts(self._client_version)
+        ):
+            self._client_version = cached_version
+
         raw_models = payload.get("models")
         parsed_models = _parse_models(raw_models)
         if not parsed_models or not any(model.visibility == "list" for model in parsed_models):
             return
         with self._lock:
             self._models = parsed_models
             self._raw_models = [dict(item) for item in raw_models if isinstance(item, dict)]
-            self._fetched_at = _parse_timestamp(payload.get("fetched_at"))
+            self._fetched_at = (
+                _parse_timestamp(payload.get("fetched_at"))
+                if cached_version == self._client_version
+                else None
+            )
             self._etag = payload.get("etag") if isinstance(payload.get("etag"), str) else None
             self._account_id = cached_account_id
 
@@ -278,7 +343,7 @@ def _persist_cache(self) -> None:
                 if self._fetched_at
                 else None,
                 "etag": self._etag,
-                "client_version": CODEX_MODELS_CLIENT_VERSION,
+                "client_version": self._client_version,
                 "account_id": self._account_id,
                 "models": self._raw_models,
             }

chatmock/model_registry.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Iterable
-
 from .model_catalog import CatalogModel, current_model_catalog
 
 
+# Request default when no model is given and no account catalog is available.
+FALLBACK_DEFAULT_MODEL = "gpt-5.4"
 ALL_REASONING_EFFORTS = ("none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra")
 DEFAULT_REASONING_EFFORTS = frozenset(ALL_REASONING_EFFORTS)
 
@@ -83,27 +83,6 @@ class ModelSpec:
         allowed_efforts=frozenset(("low", "medium", "high", "xhigh", "max")),
         variant_efforts=("low", "medium", "high", "xhigh", "max"),
     ),
-    ModelSpec(
-        public_id="gpt-6-astra",
-        upstream_id="gpt-6-astra",
-        aliases=("gpt6-astra", "gpt-6-astra-latest"),
-        allowed_efforts=frozenset(("low", "medium", "high", "xhigh", "max", "ultra")),
-        variant_efforts=("low", "medium", "high", "xhigh", "max", "ultra"),
-    ),
-    ModelSpec(
-        public_id="gpt-6-sol",
-        upstream_id="gpt-6-sol",
-        aliases=("gpt6-sol", "gpt-6-sol-latest"),
-        allowed_efforts=frozenset(("low", "medium", "high", "xhigh", "max", "ultra")),
-        variant_efforts=("low", "medium", "high", "xhigh", "max", "ultra"),
-    ),
-    ModelSpec(
-        public_id="gpt-6-luna",
-        upstream_id="gpt-6-luna",
-        aliases=("gpt6-luna", "gpt-6-luna-latest"),
-        allowed_efforts=frozenset(("low", "medium", "high", "xhigh", "max")),
-        variant_efforts=("low", "medium", "high", "xhigh", "max"),
-    ),
     ModelSpec(
         public_id="gpt-5.3-codex",
         upstream_id="gpt-5.3-codex",
@@ -245,7 +224,9 @@ def normalize_model_name(model: str | None, debug_model: str | None = None) -> s
         return spec.upstream_id
     if isinstance(model, str) and model.strip():
         return model.strip()
-    return "gpt-5.4"
+    catalog = current_model_catalog()
+    visible = catalog.visible_models() if catalog is not None else ()
+    return visible[0].slug if visible else FALLBACK_DEFAULT_MODEL
 
 
 def allowed_efforts_for_model(model: str | None) -> frozenset[str]:
@@ -267,33 +248,27 @@ def extract_reasoning_from_model_name(model: str | None) -> dict[str, str] | Non
 
 def list_public_models(expose_reasoning_models: bool = False) -> list[str]:
     catalog = current_model_catalog()
-    if catalog is not None:
-        remote_models = catalog.visible_models(wait_for_refresh=True)
-        if remote_models:
-            model_ids: list[str] = []
-            for model in remote_models:
-                model_ids.append(model.slug)
-                if expose_reasoning_models:
-                    model_ids.extend(f"{model.slug}-{effort}" for effort in model.reasoning_efforts)
-            return model_ids
-
+    remote_models = catalog.visible_models(wait_for_refresh=True) if catalog is not None else ()
     model_ids: list[str] = []
-    for spec in _MODEL_SPECS:
-        model_ids.append(spec.public_id)
+    for model in remote_models:
+        model_ids.append(model.slug)
         if expose_reasoning_models:
-            model_ids.extend(f"{spec.public_id}-{effort}" for effort in spec.variant_efforts)
+            model_ids.extend(f"{model.slug}-{effort}" for effort in model.reasoning_efforts)
     return model_ids
 
 
-def iter_public_models() -> Iterable[ModelSpec]:
-    return _MODEL_SPECS
-
-
-def model_supports_service_tier(model: str | None, service_tier: str) -> bool | None:
+def catalog_model_for_name(model: str | None) -> CatalogModel | None:
     spec, _ = _resolve_remote_model(model)
     if spec is None:
         return None
     for remote_model in _remote_models():
         if remote_model.slug == spec.upstream_id:
-            return service_tier in remote_model.service_tiers
-    return False
+            return remote_model
+    return None
+
+
+def model_supports_service_tier(model: str | None, service_tier: str) -> bool | None:
+    remote_model = catalog_model_for_name(model)
+    if remote_model is None:
+        return None
+    return service_tier in remote_model.service_tiers