1  """Remote model catalog fetcher.
  2  
  3  The Hermes docs site hosts a JSON manifest of curated models for providers
  4  we want to update without shipping a release (currently OpenRouter and
  5  Nous Portal). This module fetches, validates, and caches that manifest,
  6  falling back to the in-repo hardcoded lists when the network is unavailable.
  7  
  8  Pipeline
  9  --------
 10  1. ``get_catalog()`` — returns a parsed manifest dict.
 11     - Checks in-process cache (invalidated by TTL).
 12     - Reads disk cache at ``~/.hermes/cache/model_catalog.json``.
 13     - Fetches the master URL if disk cache is stale or missing.
 14     - On any fetch failure, keeps using the stale cache (or empty dict).
 15  
 16  2. ``get_curated_openrouter_models()`` / ``get_curated_nous_models()`` —
 17     thin accessors returning the shapes existing callers expect. Each
 18     falls back to the in-repo hardcoded list on any lookup failure.
 19  
 20  Schema (version 1)
 21  ------------------
 22  ::
 23  
 24      {
 25        "version": 1,
 26        "updated_at": "2026-04-25T22:00:00Z",
 27        "metadata": {...},                # free-form
 28        "providers": {
 29          "openrouter": {
 30            "metadata": {...},            # free-form
 31            "models": [
 32              {"id": "vendor/model", "description": "recommended",
 33               "metadata": {...}}          # free-form, model-level
 34            ]
 35          },
 36          "nous": {...}
 37        }
 38      }
 39  
 40  Unknown fields are ignored — extra metadata can be added at either level
 41  without bumping ``version``. ``version`` bumps are reserved for
 42  breaking changes (renaming ``providers``, changing ``models`` shape).
 43  """
 44  
 45  from __future__ import annotations
 46  
 47  import json
 48  import logging
 49  import time
 50  import urllib.error
 51  import urllib.request
 52  from pathlib import Path
 53  from typing import Any
 54  
 55  from hermes_cli import __version__ as _HERMES_VERSION
 56  from utils import atomic_replace
 57  
 58  logger = logging.getLogger(__name__)
 59  
 60  # ---------------------------------------------------------------------------
 61  # Constants
 62  # ---------------------------------------------------------------------------
 63  
 64  DEFAULT_CATALOG_URL = (
 65      "https://hermes-agent.nousresearch.com/docs/api/model-catalog.json"
 66  )
 67  DEFAULT_TTL_HOURS = 24
 68  DEFAULT_FETCH_TIMEOUT = 8.0
 69  SUPPORTED_SCHEMA_VERSION = 1
 70  
 71  _HERMES_USER_AGENT = f"hermes-cli/{_HERMES_VERSION}"
 72  
 73  # In-process cache to avoid repeated disk + parse work across multiple
 74  # calls within the same session. Invalidated by TTL against the disk file's
 75  # mtime, so calling code never has to think about this.
 76  _catalog_cache: dict[str, Any] | None = None
 77  _catalog_cache_source_mtime: float = 0.0
 78  
 79  
 80  # ---------------------------------------------------------------------------
 81  # Config
 82  # ---------------------------------------------------------------------------
 83  
 84  
 85  def _load_catalog_config() -> dict[str, Any]:
 86      """Load the ``model_catalog`` config block with defaults filled in."""
 87      try:
 88          from hermes_cli.config import load_config
 89          cfg = load_config() or {}
 90      except Exception:
 91          cfg = {}
 92  
 93      raw = cfg.get("model_catalog")
 94      if not isinstance(raw, dict):
 95          raw = {}
 96  
 97      return {
 98          "enabled": bool(raw.get("enabled", True)),
 99          "url": str(raw.get("url") or DEFAULT_CATALOG_URL),
100          "ttl_hours": float(raw.get("ttl_hours") or DEFAULT_TTL_HOURS),
101          "providers": raw.get("providers") if isinstance(raw.get("providers"), dict) else {},
102      }
103  
104  
105  def _cache_path() -> Path:
106      """Return the disk cache path. Import lazily so tests can monkeypatch home."""
107      from hermes_constants import get_hermes_home
108      return get_hermes_home() / "cache" / "model_catalog.json"
109  
110  
111  # ---------------------------------------------------------------------------
112  # Fetch + validate + cache
113  # ---------------------------------------------------------------------------
114  
115  
116  def _fetch_manifest(url: str, timeout: float) -> dict[str, Any] | None:
117      """HTTP GET the manifest URL and return a parsed dict, or None on failure."""
118      try:
119          req = urllib.request.Request(
120              url,
121              headers={
122                  "Accept": "application/json",
123                  "User-Agent": _HERMES_USER_AGENT,
124              },
125          )
126          with urllib.request.urlopen(req, timeout=timeout) as resp:
127              data = json.loads(resp.read().decode())
128      except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc:
129          logger.info("model catalog fetch failed (%s): %s", url, exc)
130          return None
131      except Exception as exc:  # pragma: no cover — defensive
132          logger.info("model catalog fetch errored (%s): %s", url, exc)
133          return None
134  
135      if not _validate_manifest(data):
136          logger.info("model catalog at %s failed schema validation", url)
137          return None
138  
139      return data
140  
141  
142  def _validate_manifest(data: Any) -> bool:
143      """Return True when ``data`` matches the minimum manifest shape."""
144      if not isinstance(data, dict):
145          return False
146      version = data.get("version")
147      if not isinstance(version, int) or version > SUPPORTED_SCHEMA_VERSION:
148          # Future schema version we don't understand — refuse rather than
149          # guess. Older schemas (version < 1) aren't supported either.
150          return False
151      providers = data.get("providers")
152      if not isinstance(providers, dict):
153          return False
154      for pname, pblock in providers.items():
155          if not isinstance(pname, str) or not isinstance(pblock, dict):
156              return False
157          models = pblock.get("models")
158          if not isinstance(models, list):
159              return False
160          for m in models:
161              if not isinstance(m, dict):
162                  return False
163              if not isinstance(m.get("id"), str) or not m["id"].strip():
164                  return False
165      return True
166  
167  
168  def _read_disk_cache() -> tuple[dict[str, Any] | None, float]:
169      """Return ``(data_or_none, mtime)``. mtime is 0 if file is missing."""
170      path = _cache_path()
171      try:
172          mtime = path.stat().st_mtime
173      except (OSError, FileNotFoundError):
174          return (None, 0.0)
175      try:
176          with open(path) as fh:
177              data = json.load(fh)
178      except (OSError, json.JSONDecodeError):
179          return (None, 0.0)
180      if not _validate_manifest(data):
181          return (None, 0.0)
182      return (data, mtime)
183  
184  
185  def _write_disk_cache(data: dict[str, Any]) -> None:
186      path = _cache_path()
187      try:
188          path.parent.mkdir(parents=True, exist_ok=True)
189          tmp = path.with_suffix(path.suffix + ".tmp")
190          with open(tmp, "w") as fh:
191              json.dump(data, fh, indent=2)
192              fh.write("\n")
193          atomic_replace(tmp, path)
194      except OSError as exc:
195          logger.info("model catalog cache write failed: %s", exc)
196  
197  
198  # ---------------------------------------------------------------------------
199  # Public API
200  # ---------------------------------------------------------------------------
201  
202  
203  def get_catalog(*, force_refresh: bool = False) -> dict[str, Any]:
204      """Return the parsed model catalog manifest, or an empty dict on failure.
205  
206      Callers should treat a missing provider/model as "use the in-repo fallback"
207      — never raise from this function so the CLI keeps working offline.
208      """
209      global _catalog_cache, _catalog_cache_source_mtime
210  
211      cfg = _load_catalog_config()
212      if not cfg["enabled"]:
213          return {}
214  
215      ttl_seconds = max(0.0, cfg["ttl_hours"] * 3600.0)
216  
217      disk_data, disk_mtime = _read_disk_cache()
218      now = time.time()
219      disk_fresh = disk_data is not None and (now - disk_mtime) < ttl_seconds
220  
221      # In-process cache hit: disk hasn't changed since we loaded it and still fresh.
222      if (
223          not force_refresh
224          and _catalog_cache is not None
225          and disk_data is not None
226          and disk_mtime == _catalog_cache_source_mtime
227          and disk_fresh
228      ):
229          return _catalog_cache
230  
231      # Disk is fresh enough — use it without a network hit.
232      if not force_refresh and disk_fresh and disk_data is not None:
233          _catalog_cache = disk_data
234          _catalog_cache_source_mtime = disk_mtime
235          return disk_data
236  
237      # Need to (re)fetch. If it fails, fall back to any stale disk copy.
238      fetched = _fetch_manifest(cfg["url"], DEFAULT_FETCH_TIMEOUT)
239      if fetched is not None:
240          _write_disk_cache(fetched)
241          new_disk_data, new_mtime = _read_disk_cache()
242          if new_disk_data is not None:
243              _catalog_cache = new_disk_data
244              _catalog_cache_source_mtime = new_mtime
245              return new_disk_data
246          _catalog_cache = fetched
247          _catalog_cache_source_mtime = now
248          return fetched
249  
250      if disk_data is not None:
251          _catalog_cache = disk_data
252          _catalog_cache_source_mtime = disk_mtime
253          return disk_data
254  
255      return {}
256  
257  
258  def _fetch_provider_override(provider: str) -> dict[str, Any] | None:
259      """If ``model_catalog.providers.<name>.url`` is set, fetch that instead."""
260      cfg = _load_catalog_config()
261      if not cfg["enabled"]:
262          return None
263      provider_cfg = cfg["providers"].get(provider)
264      if not isinstance(provider_cfg, dict):
265          return None
266      override_url = provider_cfg.get("url")
267      if not isinstance(override_url, str) or not override_url.strip():
268          return None
269      # Override fetches skip the disk cache because they're usually
270      # third-party self-hosted. Re-request on every call but with a short
271      # timeout so they don't block the picker.
272      return _fetch_manifest(override_url.strip(), DEFAULT_FETCH_TIMEOUT)
273  
274  
275  def _get_provider_block(provider: str) -> dict[str, Any] | None:
276      """Return the provider's manifest block, respecting per-provider overrides."""
277      override = _fetch_provider_override(provider)
278      if override is not None:
279          block = override.get("providers", {}).get(provider)
280          if isinstance(block, dict):
281              return block
282  
283      catalog = get_catalog()
284      if not catalog:
285          return None
286      block = catalog.get("providers", {}).get(provider)
287      return block if isinstance(block, dict) else None
288  
289  
290  def get_curated_openrouter_models() -> list[tuple[str, str]] | None:
291      """Return OpenRouter's curated ``[(id, description), ...]`` from the manifest.
292  
293      Returns ``None`` when the manifest is unavailable, so callers can fall
294      back to their hardcoded list.
295      """
296      block = _get_provider_block("openrouter")
297      if not block:
298          return None
299      out: list[tuple[str, str]] = []
300      for m in block.get("models", []):
301          mid = str(m.get("id") or "").strip()
302          if not mid:
303              continue
304          desc = str(m.get("description") or "")
305          out.append((mid, desc))
306      return out or None
307  
308  
309  def get_curated_nous_models() -> list[str] | None:
310      """Return Nous Portal's curated list of model ids from the manifest.
311  
312      Returns ``None`` when the manifest is unavailable.
313      """
314      block = _get_provider_block("nous")
315      if not block:
316          return None
317      out: list[str] = []
318      for m in block.get("models", []):
319          mid = str(m.get("id") or "").strip()
320          if mid:
321              out.append(mid)
322      return out or None
323  
324  
325  def reset_cache() -> None:
326      """Clear the in-process cache. Used by tests and ``hermes model --refresh``."""
327      global _catalog_cache, _catalog_cache_source_mtime
328      _catalog_cache = None
329      _catalog_cache_source_mtime = 0.0