#!/usr/bin/env python3 """Models.dev catalog refresh / snapshot automation for CodeWhale (#4117). Fetches the public Models.dev combined catalog, validates offline bundled seed shape, and supports OpenRouter public-listing inspection. The automation is intentionally validate/dry-run only: it never accepts, prints, or persists API keys / auth headers, and it does not write fetched JSON to disk. Usage examples: # Dry-run: fetch + validate, print counts (no write) scripts/catalog_models_dev.py refresh # Validate the in-repo offline seed still parses as Models.dev-shaped JSON scripts/catalog_models_dev.py snapshot --check \\ crates/config/assets/models_dev.bundled.json # OpenRouter public /models listing (no key), dry-run only scripts/catalog_models_dev.py refresh --provider openrouter \\ --sort newest --limit 100 Environment: CODEWHALE_MODELS_DEV_URL Override Models.dev catalog URL CODEWHALE_MODELS_DEV_PATH Read catalog JSON from a local file instead of network """ from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.request from pathlib import Path from typing import Any DEFAULT_MODELS_DEV_URL = "https://models.dev/catalog.json" DEFAULT_OPENROUTER_MODELS_URL = "https://openrouter.ai/api/v1/models" USER_AGENT = "CodeWhale-catalog-automation/0.9.0 (+https://github.com/Hmbown/CodeWhale)" FETCH_TIMEOUT_SECS = 60 def die(msg: str, code: int = 1) -> None: print(f"error: {msg}", file=sys.stderr) raise SystemExit(code) def load_json_bytes(raw: bytes, source: str) -> Any: try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: die(f"{source}: not utf-8 ({exc})") try: return json.loads(text) except json.JSONDecodeError as exc: die(f"{source}: invalid JSON ({exc})") def fetch_url(url: str) -> bytes: req = urllib.request.Request( url, headers={ "User-Agent": USER_AGENT, "Accept": "application/json", # Explicitly no Authorization header — public endpoints only. }, method="GET", ) try: with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT_SECS) as resp: # Refuse to follow into non-JSON surprise payloads larger than 64 MiB. data = resp.read(64 * 1024 * 1024 + 1) if len(data) > 64 * 1024 * 1024: die(f"{url}: response exceeds 64 MiB safety cap") ctype = resp.headers.get("Content-Type", "") if "json" not in ctype.lower() and not data.lstrip().startswith((b"{", b"[")): die(f"{url}: unexpected Content-Type {ctype!r}") return data except urllib.error.HTTPError as exc: die(f"{url}: HTTP {exc.code} {exc.reason}") except urllib.error.URLError as exc: die(f"{url}: {exc.reason}") def load_models_dev_catalog() -> tuple[dict[str, Any], str, bool]: """Return (document, source_label, is_local_file). Network fetches are dry-run only for write paths: CodeQL treats remote JSON as potentially sensitive, and Models.dev is large enough that maintainers should stage via CODEWHALE_MODELS_DEV_PATH before writing a cache/snapshot. """ path_override = os.environ.get("CODEWHALE_MODELS_DEV_PATH", "").strip() if path_override: p = Path(path_override) if not p.is_file(): die(f"CODEWHALE_MODELS_DEV_PATH not a file: {p}") raw = p.read_bytes() data = load_json_bytes(raw, str(p)) return ensure_models_dev_shape(data, str(p)), f"file:{p}", True url = os.environ.get("CODEWHALE_MODELS_DEV_URL", DEFAULT_MODELS_DEV_URL).strip() if not url: url = DEFAULT_MODELS_DEV_URL raw = fetch_url(url) data = load_json_bytes(raw, url) return ensure_models_dev_shape(data, url), f"url:{url}", False def ensure_models_dev_shape(data: Any, source: str) -> dict[str, Any]: if not isinstance(data, dict): die(f"{source}: expected object root") # Allow optional _meta (CodeWhale offline seed) and require models+providers # when present so we never write a partial secret leak document. models = data.get("models") providers = data.get("providers") if models is None and providers is None: die(f"{source}: missing both 'models' and 'providers'") if models is not None and not isinstance(models, dict): die(f"{source}: 'models' must be an object") if providers is not None and not isinstance(providers, dict): die(f"{source}: 'providers' must be an object") # Rebuild a public document from allowlisted top-level keys only so we never # persist credential-shaped fields even if a future Models.dev field adds them. return public_models_dev_document(data) def is_credential_key(key: str) -> bool: banned_exact = { "api_key", "apikey", "authorization", "token", "access_token", "refresh_token", "secret", "password", "client_secret", } lowered = key.lower() return lowered in banned_exact or lowered.endswith("_api_key") or lowered.endswith("_secret") def scrub_secrets(node: Any) -> Any: """Drop keys that look like credentials; never persist auth material.""" if isinstance(node, dict): out: dict[str, Any] = {} for key, value in node.items(): if not isinstance(key, str) or is_credential_key(key): continue out[key] = scrub_secrets(value) return out if isinstance(node, list): return [scrub_secrets(item) for item in node] if isinstance(node, (str, int, float, bool)) or node is None: return node # Drop non-JSON-scalar oddities rather than serializing them. return None def public_models_dev_document(data: dict[str, Any]) -> dict[str, Any]: """Construct a write-safe Models.dev-shaped document (public metadata only).""" out: dict[str, Any] = {} if isinstance(data.get("_meta"), dict): out["_meta"] = scrub_secrets(data["_meta"]) if isinstance(data.get("models"), dict): out["models"] = scrub_secrets(data["models"]) if isinstance(data.get("providers"), dict): out["providers"] = scrub_secrets(data["providers"]) return out def public_source_label(source: str) -> str: """Log a catalog origin without query/fragment (tokens live there).""" if source.startswith("url:"): url = source[4:] for sep in ("?", "#"): url = url.split(sep, 1)[0] return f"url:{url}" return source def public_limit_value(value: Any) -> str: """Format a catalog limit for logs. Never print credential-shaped strings. Remote catalog JSON is tainted for clear-text-logging rules. Only numeric limits are meaningful here; anything else (including token-shaped strings) is replaced with a constant so the raw value cannot reach stdout. """ if isinstance(value, bool): return "redacted" if value is None: return "null" if isinstance(value, int): return str(value) if isinstance(value, float): return format(value, ".6g") return "redacted" def catalog_stats(data: dict[str, Any]) -> str: models = data.get("models") or {} providers = data.get("providers") or {} offerings = 0 if isinstance(providers, dict): for prov in providers.values(): if isinstance(prov, dict): models_map = prov.get("models") or {} if isinstance(models_map, dict): offerings += len(models_map) return ( f"providers={len(providers) if isinstance(providers, dict) else 0} " f"canonical_models={len(models) if isinstance(models, dict) else 0} " f"provider_offerings={offerings}" ) def cmd_refresh(args: argparse.Namespace) -> None: if args.provider and args.provider.lower() == "openrouter": refresh_openrouter(args) return if args.provider: die( f"unsupported --provider {args.provider!r} " "(supported: openrouter, or omit for Models.dev)" ) data, source, _is_local = load_models_dev_catalog() print(f"loaded Models.dev catalog from {source}") print(catalog_stats(data)) if args.write_cache or args.write: die( "disk writes are intentionally unsupported (secret-free by design); " "use `snapshot --check PATH` to validate a local Models.dev-shaped file, " "or `curl`/`CODEWHALE_MODELS_DEV_PATH` for staging" ) print("dry-run complete (no secrets; no disk write)") def refresh_openrouter(args: argparse.Namespace) -> None: url = DEFAULT_OPENROUTER_MODELS_URL raw = fetch_url(url) data = load_json_bytes(raw, url) if not isinstance(data, dict) or "data" not in data: die(f"{url}: expected {{ data: [...] }} envelope") rows = data["data"] if not isinstance(rows, list): die(f"{url}: data is not a list") # Optional sort / limit for local inspection — never secrets. if args.sort == "newest": def created_key(row: Any) -> float: if not isinstance(row, dict): return 0.0 created = row.get("created") try: return float(created) except (TypeError, ValueError): return 0.0 rows = sorted(rows, key=created_key, reverse=True) if args.limit is not None and args.limit > 0: rows = rows[: args.limit] # Project only public catalog fields — never the raw response object — # so credential-shaped keys cannot reach disk even if OpenRouter adds them. public_rows: list[dict[str, Any]] = [] allowed = { "id", "name", "created", "description", "context_length", "architecture", "pricing", "top_provider", "per_request_limits", "supported_parameters", } for row in rows: if not isinstance(row, dict): continue projected: dict[str, Any] = {} for key in allowed: if key in row and not is_credential_key(key): projected[key] = scrub_secrets(row[key]) if projected.get("id"): public_rows.append(projected) payload = { "_meta": { "source": "openrouter.ai/api/v1/models", "note": "Public model listing for cache dogfood; not the Models.dev SoT.", "count": len(public_rows), "sort": args.sort, "limit": args.limit, }, "data": public_rows, } print(f"loaded OpenRouter models: {len(public_rows)} rows (sort={args.sort}, limit={args.limit})") if args.write_cache: # OpenRouter listing is always network-sourced; avoid disk write of remote JSON. die( "OpenRouter refresh is dry-run only (no disk write). " "Use Models.dev with CODEWHALE_MODELS_DEV_PATH for offline snapshots." ) else: print("dry-run complete (OpenRouter writes disabled; use Models.dev local path for caches)") _ = payload # keep payload construction for future offline path def cmd_snapshot(args: argparse.Namespace) -> None: target = Path(args.path) if args.check: if not target.is_file(): die(f"--check: missing {target}") raw = target.read_bytes() data = load_json_bytes(raw, str(target)) ensure_models_dev_shape(data, str(target)) print(f"ok: {target} is Models.dev-shaped ({catalog_stats(data)})") return data, source, _is_local = load_models_dev_catalog() print(f"loaded Models.dev catalog from {source}") print(catalog_stats(data)) if args.write or args.force_full: die( "disk writes are intentionally unsupported for this automation; " "validate with --check, or stage a file outside this tool" ) print("dry-run complete (use --check PATH to validate an existing snapshot)") def _limit_fields(entry: Any) -> dict[str, Any]: if not isinstance(entry, dict): return {} limit = entry.get("limit") return limit if isinstance(limit, dict) else {} def _collect_limit_drift( seed_value: Any, upstream_value: Any, path: str, drift: list[str], ) -> None: seed_limit = _limit_fields(seed_value) upstream_limit = _limit_fields(upstream_value) for field in ("context", "output"): bundled = seed_limit.get(field) upstream = upstream_limit.get(field) if bundled != upstream: drift.append( f"{path}: limit.{field} " f"bundled={public_limit_value(bundled)} " f"upstream={public_limit_value(upstream)}" ) def cmd_drift(args: argparse.Namespace) -> None: """Diff the bundled seed against upstream for limit.output / limit.context. Dry-run only: no API key, no disk write. Network access (unless CODEWHALE_MODELS_DEV_PATH points at a local file) reads the public Models.dev catalog. Wiring this as a CI gate is a maintainer decision and is deliberately not activated here. """ seed_path = Path(args.seed) if not seed_path.is_file(): die(f"drift: missing bundled seed {seed_path}") seed = load_json_bytes(seed_path.read_bytes(), str(seed_path)) ensure_models_dev_shape(seed, str(seed_path)) upstream, source, _is_local = load_models_dev_catalog() drift: list[str] = [] missing_upstream: list[str] = [] seed_models = seed.get("models") or {} upstream_models = upstream.get("models") or {} if isinstance(seed_models, dict) and isinstance(upstream_models, dict): for model_id in sorted(seed_models): if model_id not in upstream_models: missing_upstream.append(f"models.{model_id}") continue _collect_limit_drift( seed_models[model_id], upstream_models[model_id], f"models.{model_id}", drift, ) seed_providers = seed.get("providers") or {} upstream_providers = upstream.get("providers") or {} if isinstance(seed_providers, dict) and isinstance(upstream_providers, dict): for provider_id in sorted(seed_providers): seed_provider = seed_providers[provider_id] seed_offerings = ( seed_provider.get("models") if isinstance(seed_provider, dict) else {} ) if not isinstance(seed_offerings, dict): continue upstream_provider = upstream_providers.get(provider_id) upstream_offerings = ( upstream_provider.get("models") if isinstance(upstream_provider, dict) else {} ) if not isinstance(upstream_offerings, dict): upstream_offerings = {} for model_id in sorted(seed_offerings): path = f"providers.{provider_id}.models.{model_id}" if model_id not in upstream_offerings: missing_upstream.append(path) continue _collect_limit_drift( seed_offerings[model_id], upstream_offerings[model_id], path, drift, ) print(f"bundled seed: {seed_path}") print(f"upstream: {public_source_label(source)}") if missing_upstream: print("removed upstream (bundled id no longer present):") for path in missing_upstream: print(f" - {path}") if drift: print(f"limit drift detected ({len(drift)}):") for path in drift: print(f" - {path}") # Non-zero so a future CI gate can fail on drift; activating that gate # is a maintainer decision. die( "bundled seed has drifted from upstream for limit.output / limit.context", code=2, ) if missing_upstream: print("note: removed-upstream ids are reported above (non-fatal).") print("no limit.output / limit.context drift") def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Secret-free Models.dev / OpenRouter catalog automation (#4117)" ) sub = p.add_subparsers(dest="cmd", required=True) refresh = sub.add_parser("refresh", help="Fetch live catalog / provider models") refresh.add_argument( "--provider", default=None, help="Optional provider id (currently: openrouter). Omit for Models.dev.", ) refresh.add_argument( "--sort", default="newest", choices=["newest", "none"], help="OpenRouter sort order (default: newest)", ) refresh.add_argument( "--limit", type=int, default=100, help="OpenRouter row cap (default: 100; 0 = no cap)", ) refresh.add_argument( "--write-cache", metavar="PATH", help="Deprecated/unsupported: validate-only automation never writes fetched JSON", ) refresh.add_argument( "--write", metavar="PATH", help="Deprecated/unsupported alias of --write-cache", ) refresh.set_defaults(func=cmd_refresh) snapshot = sub.add_parser( "snapshot", help="Validate or write a Models.dev-shaped snapshot document", ) snapshot.add_argument( "path", nargs="?", default="crates/config/assets/models_dev.bundled.json", help="Snapshot path (default: offline seed asset)", ) snapshot.add_argument( "--check", action="store_true", help="Validate existing file only (no network)", ) snapshot.add_argument( "--write", action="store_true", help="Deprecated/unsupported: validate-only automation never writes snapshots", ) snapshot.add_argument( "--force-full", action="store_true", help="Deprecated/unsupported with --write; retained for clear failure messages", ) snapshot.set_defaults(func=cmd_snapshot) drift = sub.add_parser( "drift", help="Diff bundled seed against upstream for limit.output / limit.context", ) drift.add_argument( "--seed", default="crates/config/assets/models_dev.bundled.json", help="Bundled seed path (default: crates/config/assets/models_dev.bundled.json)", ) drift.set_defaults(func=cmd_drift) return p def main(argv: list[str] | None = None) -> None: parser = build_parser() args = parser.parse_args(argv) args.func(args) if __name__ == "__main__": main()