hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

fix(models): ETag conditional GET, no-network hot-path invariant, mirror URL override for models.dev catalog

Browse Source 
Harden the models.dev catalog refresh path (#35838) with three missing
pieces:

1. ETag conditional GET — every network request sends If-None-Match
   with the last-known ETag (persisted alongside the cache file). A 304
   Not Modified re-confirms the existing cache without re-downloading
   the full ~2 MB registry. This makes the 4-hour TTL effectively free
   to maintain.

2. No-network-on-hot-paths invariant — allow_network=False is now the
   default for every query function called on the conversation hot path:
   get_model_capabilities, get_model_info, lookup_models_dev_context,
   _get_provider_models. These are called during vision routing, image
   routing, cost-guard checks, and context-length resolution on every
   turn — they must never block on the network. Interactive flows
   (model picker, model switch) explicitly pass allow_network=True.

3. Mirror URL override — models_dev.url in config.yaml lets deployments
   point at a self-hosted mirror without code changes. Follows the same
   pattern as model_catalog.url.

Additional hardening:
- Cache TTL bumped from 1h to 4h (ETag makes refresh cheap)
- Corrupt/empty disk cache is rejected with a warning instead of being
  served as {} and silently breaking provider/model resolution
- _validate_registry() guards against non-dict and empty-dict payloads

Fixes #35838
This commit is contained in:
kshitij
2026-08-14 00:25:31 +05:30
parent 2ae96939f5
commit acd8737c10
4 changed files with 585 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

230
agent/models_dev.py
View File

@@ -15,8 +15,23 @@ Data resolution order:
served rather than blocking callers on the network)
3. Network fetch (https://models.dev/api.json) — only when no cache
exists at all; failed refreshes back off for 5 minutes process-wide
Latency-sensitive callers (gateway route-identity checks) pass
``allow_network=False`` and never touch the network.
Network hardening:
- **ETag conditional GET**: every network request sends ``If-None-Match``
with the last-known ETag. A 304 Not Modified response is a no-op — the
existing cache is re-confirmed fresh without re-downloading the full
registry (≈2 MB). The ETag is persisted alongside the cache file.
- **No-network-on-hot-paths invariant**: resolution, picker, and resume
paths NEVER perform network I/O. ``allow_network=False`` is threaded
through every query function, and hot-path callers (vision routing,
image routing, cost guard, context-length lookup) pass it explicitly.
- **Corrupt-cache rejection**: a disk cache that fails to parse, is not a
dict, or is empty is ignored with a warning rather than served as
``{}`` and silently breaking provider/model resolution.
- **Mirror URL override**: ``models_dev.url`` in config.yaml lets
deployments point at a mirror (e.g. a self-hosted copy) without code
changes.
Other modules should import the dataclasses and query functions from here
rather than parsing the raw JSON themselves.
@@ -36,8 +51,9 @@ import requests
logger = logging.getLogger(__name__)
MODELS_DEV_URL = "https://models.dev/api.json"
_MODELS_DEV_CACHE_TTL = 3600 # 1 hour in-memory
_DEFAULT_MODELS_DEV_URL = "https://models.dev/api.json"
MODELS_DEV_URL = _DEFAULT_MODELS_DEV_URL
_MODELS_DEV_CACHE_TTL = 4 * 3600 # 4 hours — ETag conditional GET makes refresh cheap
_MODELS_DEV_RETRY_DELAY = 300 # 5 minutes after a failed refresh
# In-memory cache
@@ -220,15 +236,81 @@ def _get_cache_path() -> Path:
return get_hermes_home() / "models_dev_cache.json"
def _get_etag_path() -> Path:
"""Return path to the ETag sidecar file for conditional GET."""
from hermes_constants import get_hermes_home
return get_hermes_home() / "models_dev_cache.etag"
def _load_etag() -> str:
"""Load the last-known ETag from disk, or empty string if missing."""
try:
etag_path = _get_etag_path()
if etag_path.exists():
return etag_path.read_text(encoding="utf-8").strip()
except Exception as e:
logger.debug("Failed to load models.dev ETag: %s", e)
return ""
def _save_etag(etag: str) -> None:
"""Persist an ETag to the sidecar file atomically."""
try:
etag_path = _get_etag_path()
etag_path.parent.mkdir(parents=True, exist_ok=True)
tmp = etag_path.with_suffix(".tmp")
tmp.write_text(etag, encoding="utf-8")
tmp.replace(etag_path)
except Exception as e:
logger.debug("Failed to save models.dev ETag: %s", e)
def _get_models_dev_url() -> str:
"""Resolve the models.dev API URL, honoring a config.yaml override.
The ``models_dev.url`` config key lets deployments point at a mirror
(e.g. a self-hosted copy behind a corporate proxy) without code changes.
Falls back to the default public URL when unset or empty.
"""
try:
from hermes_cli.config import cfg_get, load_config_readonly
cfg = load_config_readonly()
url = cfg_get(cfg, "models_dev", "url", default="")
if isinstance(url, str) and url.strip():
return url.strip()
except Exception:
pass
return _DEFAULT_MODELS_DEV_URL
def _validate_registry(data: Any) -> bool:
"""Return True if *data* is a non-empty dict suitable for serving."""
return isinstance(data, dict) and len(data) > 0
def _load_disk_cache() -> Dict[str, Any]:
"""Load models.dev data from disk cache."""
"""Load models.dev data from disk cache.
A corrupt cache (invalid JSON, not a dict, or empty) is rejected with
a warning so it doesn't silently masquerade as ``{}`` and break
provider/model resolution for every caller.
"""
try:
cache_path = _get_cache_path()
if cache_path.exists():
with open(cache_path, encoding="utf-8") as f:
return json.load(f)
data = json.load(f)
if not _validate_registry(data):
logger.warning(
"models.dev disk cache is corrupt or empty; ignoring "
"(will refetch from network)"
)
return {}
return data
except Exception as e:
logger.debug("Failed to load models.dev disk cache: %s", e)
logger.warning(
"Failed to load models.dev disk cache; ignoring: %s", e
)
return {}
@@ -258,30 +340,62 @@ def _disk_cache_age_seconds() -> Optional[float]:
return None
def _save_disk_cache(data: Dict[str, Any]) -> None:
"""Save models.dev data to disk cache atomically."""
def _save_disk_cache(data: Dict[str, Any], etag: str = "") -> None:
"""Save models.dev data to disk cache atomically.
Also persists the ETag sidecar when *etag* is non-empty so the next
refresh can issue a conditional GET.
"""
try:
cache_path = _get_cache_path()
atomic_json_write(cache_path, data, indent=None, separators=(",", ":"))
except Exception as e:
logger.debug("Failed to save models.dev disk cache: %s", e)
if etag:
_save_etag(etag)
class _NotModified(Exception):
"""Server returned 304 Not Modified — existing cache is still valid."""
def _fetch_models_dev_from_network() -> Dict[str, Any]:
"""Fetch the live models.dev registry without touching local caches.
Uses ETag conditional GET: sends ``If-None-Match`` when a cached ETag
exists. A 304 Not Modified response means the cached registry is still
current; this raises ``_NotModified`` so the caller can re-confirm the
existing cache's freshness without re-downloading the full payload.
Raises on network errors and on an empty/invalid registry payload.
"""
url = _get_models_dev_url()
headers: Dict[str, str] = {}
etag = _load_etag()
if etag:
headers["If-None-Match"] = etag
# Tuple (connect, read): a flat timeout=15 let a blackholed connect
# stall the first-turn critical path for the full 15 s. 5 s connect
# fails fast on unreachable hosts; 10 s read still tolerates a slow
# registry response (matches the OpenRouter fetch convention in
# agent/model_metadata.py).
response = requests.get(MODELS_DEV_URL, timeout=(5, 10))
response = requests.get(url, headers=headers, timeout=(5, 10))
if response.status_code == 304:
raise _NotModified()
response.raise_for_status()
data = response.json()
if not isinstance(data, dict) or not data:
raise ValueError("models.dev returned an empty or invalid registry")
# Persist the new ETag alongside the cache so the next conditional
# GET can short-circuit.
new_etag = response.headers.get("ETag", "")
if new_etag:
_save_etag(new_etag)
return data
@@ -319,6 +433,24 @@ def _commit_registry(data: Dict[str, Any], *, where: str) -> None:
)
def _confirm_cache_not_modified(*, where: str) -> None:
"""Re-confirm the existing cache as fresh after a 304 Not Modified.
Callers must hold ``_models_dev_fetch_lock``. Clears the backoff and
resets the in-memory cache timestamp so the next caller hits the fast
path. The disk cache itself is not rewritten — its contents are
unchanged, only its freshness marker is advanced.
"""
global _models_dev_cache_time, _models_dev_retry_after
_models_dev_cache_time = time.time()
_models_dev_retry_after = 0
logger.debug(
"models.dev registry unchanged (304 Not Modified, %s); "
"cache re-confirmed fresh",
where,
)
def _note_refresh_failure(exc: Exception, *, where: str) -> None:
"""Record a failed refresh: arm the process-wide 5-minute backoff.
@@ -341,6 +473,9 @@ def _background_refresh_models_dev() -> None:
data = _fetch_models_dev_from_network()
with _models_dev_fetch_lock:
_commit_registry(data, where="background")
except _NotModified:
with _models_dev_fetch_lock:
_confirm_cache_not_modified(where="background")
except Exception as e:
with _models_dev_fetch_lock:
_note_refresh_failure(e, where="background")
@@ -384,6 +519,11 @@ def fetch_models_dev(
Returns the full registry dict keyed by provider ID, or empty dict on failure.
Network requests use ETag conditional GET: when a cached ETag exists,
an ``If-None-Match`` header is sent. A 304 Not Modified response
re-confirms the existing cache's freshness without re-downloading the
full (~2 MB) registry.
Cache hierarchy (when ``force_refresh=False``):
1. Fresh in-memory cache → return immediately.
2. Stale in-memory cache → return immediately and refresh in a single
@@ -392,6 +532,7 @@ def fetch_models_dev(
new models, so stale data is preferable to a foreground timeout.
3. Disk cache file (any age) → load, populate in-mem, return
immediately. Stale disk caches trigger the same background refresh.
A corrupt or empty disk cache is rejected with a warning.
4. No cache at all → singleflight foreground network fetch. On
success, save to disk + in-mem and return.
5. Any failed refresh (foreground or background) suppresses further
@@ -402,8 +543,9 @@ def fetch_models_dev(
backoff are bypassed; the function hits the network and only falls back
to cached data if the call fails. When ``allow_network=False``, any
memory or disk cache is returned regardless of age and no request is
made — used by latency-sensitive paths (gateway route-identity checks)
that must never wait on the network.
made — used by latency-sensitive paths (gateway route-identity checks,
vision routing, context-length lookup) that must never wait on the
network.
"""
global _models_dev_cache, _models_dev_cache_time, _models_dev_retry_after
@@ -488,6 +630,11 @@ def fetch_models_dev(
data = _fetch_models_dev_from_network()
_commit_registry(data, where="foreground")
return data
except _NotModified:
# Server confirmed our cache is still valid. Re-confirm freshness
# without re-downloading the full registry.
_confirm_cache_not_modified(where="foreground")
return _models_dev_cache
except Exception as e:
_note_refresh_failure(e, where="foreground")
@@ -506,7 +653,9 @@ def fetch_models_dev(
return _models_dev_cache
def lookup_models_dev_context(provider: str, model: str) -> Optional[int]:
def lookup_models_dev_context(
provider: str, model: str, *, allow_network: bool = False
) -> Optional[int]:
"""Look up context_length for a provider+model combo in models.dev.
Returns the context window in tokens, or None if not found.
@@ -516,6 +665,10 @@ def lookup_models_dev_context(provider: str, model: str) -> Optional[int]:
wins over the catalog value; ``_default`` entries fill the gap only
when the catalog has no answer — the supported self-unblock path for
models with wrong or missing context in models.dev (#84482).
``allow_network`` defaults to False — context-length lookup is a
hot path (called during every conversation turn) and must never block
on the network. Pass True only from explicit refresh flows.
"""
# Explicit config override — checked before catalog so it always wins.
override_ctx = _override_context_window(provider, model)
@@ -526,7 +679,7 @@ def lookup_models_dev_context(provider: str, model: str) -> Optional[int]:
if not mdev_provider_id:
return _default_override_context(provider)
data = fetch_models_dev()
data = fetch_models_dev(allow_network=allow_network)
provider_data = data.get(mdev_provider_id)
if not isinstance(provider_data, dict):
return _default_override_context(provider)
@@ -855,16 +1008,21 @@ def _merge_catalog_entry_with_override(
return merged
def _get_provider_models(provider: str) -> Optional[Dict[str, Any]]:
def _get_provider_models(
provider: str, *, allow_network: bool = False
) -> Optional[Dict[str, Any]]:
"""Resolve a Hermes provider ID to its models dict from models.dev.
Returns the models dict or None if the provider is unknown or has no data.
``allow_network`` defaults to False — this is called from hot paths
(vision routing, image routing, capability checks) and must never block.
"""
mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider)
if not mdev_provider_id:
return None
data = fetch_models_dev()
data = fetch_models_dev(allow_network=allow_network)
provider_data = data.get(mdev_provider_id)
if not isinstance(provider_data, dict):
return None
@@ -911,7 +1069,9 @@ def _find_model_entry(models: Dict[str, Any], model: str) -> Optional[Dict[str,
return None
def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilities]:
def get_model_capabilities(
provider: str, model: str, *, allow_network: bool = False
) -> Optional[ModelCapabilities]:
"""Look up full capability metadata from models.dev cache.
Uses the existing fetch_models_dev() and PROVIDER_TO_MODELS_DEV mapping.
@@ -925,6 +1085,9 @@ def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilit
of fields; unspecified fields fall through to the catalog value (or
sensible defaults when the model is absent from the catalog).
``allow_network`` defaults to False — capability lookup is a hot path
(vision routing, image routing) and must never block on the network.
Extracts from model entry fields:
- reasoning (bool) → supports_reasoning
- tool_call (bool) → supports_tools
@@ -933,7 +1096,7 @@ def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilit
- limit.output (int) → max_output_tokens
- family (str) → model_family
"""
models = _get_provider_models(provider)
models = _get_provider_models(provider, allow_network=allow_network)
entry = _find_model_entry(models, model) if models is not None else None
# Select the override AFTER the catalog lookup: explicit overrides
@@ -1010,15 +1173,21 @@ def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilit
)
def list_provider_models(provider: str) -> List[str]:
def list_provider_models(
provider: str, *, allow_network: bool = True
) -> List[str]:
"""Return all model IDs for a provider from models.dev.
Returns an empty list if the provider is unknown or has no data.
``allow_network`` defaults to True — this is called from the model
picker (``hermes model``), which is an interactive user-facing flow
where a fresh catalog is worth a short network wait.
"""
from hermes_cli.models import normalize_provider
provider = normalize_provider(provider) or provider
models = _get_provider_models(provider)
models = _get_provider_models(provider, allow_network=allow_network)
if models is None:
return []
return [
@@ -1074,14 +1243,19 @@ def _should_hide_from_provider_catalog(provider: str, model_id: str) -> bool:
return False
def list_agentic_models(provider: str) -> List[str]:
def list_agentic_models(
provider: str, *, allow_network: bool = True
) -> List[str]:
"""Return model IDs suitable for agentic use from models.dev.
Filters for tool_call=True and excludes noise (TTS, embedding,
dated preview snapshots, live/streaming, image-only models).
Returns an empty list on any failure.
``allow_network`` defaults to True — like ``list_provider_models``,
this is called from interactive model selection flows.
"""
models = _get_provider_models(provider)
models = _get_provider_models(provider, allow_network=allow_network)
if models is None:
return []
@@ -1180,6 +1354,11 @@ def get_provider_info(
Accepts either a Hermes provider ID (e.g. "kilocode") or a models.dev
ID (e.g. "kilo"). Returns None if the provider is not in the catalog.
``allow_network`` defaults to True — the primary caller is
``resolve_provider_full`` during interactive setup, where a fresh
catalog is worth a short network wait. Hot-path callers should pass
``allow_network=False``.
"""
# Resolve Hermes ID → models.dev ID
mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id)
@@ -1204,7 +1383,7 @@ def get_provider_info(
# ---------------------------------------------------------------------------
def get_model_info(
provider_id: str, model_id: str
provider_id: str, model_id: str, *, allow_network: bool = False
) -> Optional[ModelInfo]:
"""Get full model metadata from models.dev.
@@ -1218,6 +1397,9 @@ def get_model_info(
``modalities``) are merged rather than clobbered. EXPLICIT entries
patch known catalog models; ``_default`` entries fill the gap only
for models the catalog does not know (#8731, #84482).
``allow_network`` defaults to False — model info lookup is a hot path
(cost guard, inventory) and must never block on the network.
"""
mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id)
@@ -1236,7 +1418,7 @@ def get_model_info(
shaped = _merge_catalog_entry_with_override(base, override)
return _parse_model_info(model_id, shaped, mdev_id)
data = fetch_models_dev()
data = fetch_models_dev(allow_network=allow_network)
pdata = data.get(mdev_id)
if not isinstance(pdata, dict):
return _from_override_alone()

10
hermes_cli/config_defaults.py
View File

@@ -2585,6 +2585,16 @@ DEFAULT_CONFIG = {
# context_window: 128000
"model_overrides": {},
# models.dev registry — provider/model metadata (context windows,
# capabilities, pricing, modalities). The agent fetches this on startup
# and serves from cache; a background daemon refreshes stale data.
# Override ``url`` to point at a mirror (e.g. a self-hosted copy behind
# a corporate proxy). ETag conditional GET ensures refreshes are
# cheap (304 = no download).
"models_dev": {
"url": "", # empty = default https://models.dev/api.json
},
# Network settings — workarounds for connectivity issues.
"network": {
# Force IPv4 connections. On servers with broken or unreachable IPv6,

4
hermes_cli/model_switch.py
View File

@@ -1911,10 +1911,10 @@ def switch_model(
base_url = normalize_opencode_base_url(target_provider, api_mode, base_url)
# --- Get capabilities (legacy) ---
capabilities = get_model_capabilities(target_provider, new_model)
capabilities = get_model_capabilities(target_provider, new_model, allow_network=True)
# --- Get full model info from models.dev ---
model_info = get_model_info(target_provider, new_model)
model_info = get_model_info(target_provider, new_model, allow_network=True)
# --- Collect warnings ---
warnings: list[str] = []

377
tests/agent/test_models_dev.py
View File

@@ -13,6 +13,8 @@ from agent.models_dev import (
_explicit_model_override,
_override_context_window,
_override_for,
_NotModified,
_validate_registry,
fetch_models_dev,
get_model_capabilities,
get_model_info,
@@ -161,6 +163,15 @@ class TestFetchModelsDev:
md._models_dev_retry_after = 0
md._models_dev_refresh_in_flight = False
def _mock_response(self, data, etag="", status_code=200):
"""Build a MagicMock response with optional ETag header."""
resp = MagicMock()
resp.status_code = status_code
resp.json.return_value = data
resp.headers = {"ETag": etag} if etag else {}
resp.raise_for_status = MagicMock()
return resp
@@ -176,6 +187,7 @@ class TestFetchModelsDev:
with patch.object(md, "_disk_cache_age_seconds",
return_value=md._MODELS_DEV_CACHE_TTL + 60), \
patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_start_background_refresh_models_dev") as mock_refresh:
result = fetch_models_dev()
@@ -197,7 +209,8 @@ class TestFetchModelsDev:
md,
"_disk_cache_age_seconds",
return_value=md._MODELS_DEV_CACHE_TTL + 60,
), patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY):
), patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY), \
patch.object(md, "_load_etag", return_value=""):
first = fetch_models_dev()
# Join the background refresh worker so its failure backoff is
# observable and requests.get stays patched for its lifetime.
@@ -225,20 +238,22 @@ class TestFetchModelsDev:
"""The bg worker must save disk + swap mem cache + clear backoff."""
import agent.models_dev as md
response = MagicMock()
response.json.return_value = SAMPLE_REGISTRY
response = self._mock_response(SAMPLE_REGISTRY, etag='"abc123"')
mock_get.return_value = response
md._models_dev_cache = {"stale": {}}
md._models_dev_cache_time = 0
md._models_dev_retry_after = time.time() - 1
with patch.object(md, "_save_disk_cache") as mock_save:
with patch.object(md, "_save_disk_cache") as mock_save, \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag") as mock_save_etag:
# Run the worker synchronously — deterministic, no thread.
md._models_dev_refresh_in_flight = True
md._background_refresh_models_dev()
mock_save.assert_called_once_with(SAMPLE_REGISTRY)
mock_save_etag.assert_called_once_with('"abc123"')
assert md._models_dev_cache == SAMPLE_REGISTRY
assert md._models_dev_cache_time > 0
assert md._models_dev_retry_after == 0
@@ -251,8 +266,7 @@ class TestFetchModelsDev:
request_started = threading.Event()
release_request = threading.Event()
response = MagicMock()
response.json.return_value = SAMPLE_REGISTRY
response = self._mock_response(SAMPLE_REGISTRY)
def blocking_get(*_args, **_kwargs):
request_started.set()
@@ -262,7 +276,9 @@ class TestFetchModelsDev:
mock_get.side_effect = blocking_get
with patch.object(md, "_disk_cache_age_seconds", return_value=None), patch.object(
md, "_save_disk_cache"
), ThreadPoolExecutor(max_workers=6) as pool:
), patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag"), \
ThreadPoolExecutor(max_workers=6) as pool:
futures = [pool.submit(fetch_models_dev) for _ in range(6)]
assert request_started.wait(timeout=2)
release_request.set()
@@ -275,13 +291,14 @@ class TestFetchModelsDev:
def test_force_refresh_bypasses_failure_backoff(self, mock_get):
import agent.models_dev as md
response = MagicMock()
response.json.return_value = SAMPLE_REGISTRY
response = self._mock_response(SAMPLE_REGISTRY)
mock_get.side_effect = [OSError("models.dev unreachable"), response]
with patch.object(md, "_disk_cache_age_seconds", return_value=None), patch.object(
md, "_load_disk_cache", return_value={}
), patch.object(md, "_save_disk_cache"):
), patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag"):
assert fetch_models_dev() == {}
assert fetch_models_dev(force_refresh=True) == SAMPLE_REGISTRY
@@ -322,6 +339,346 @@ class TestFetchModelsDev:
# ---------------------------------------------------------------------------
# ETag conditional GET
# ---------------------------------------------------------------------------
class TestETagConditionalGet:
"""Tests for ETag-based conditional GET (If-None-Match / 304 handling)."""
@pytest.fixture(autouse=True)
def _reset_fetch_state(self):
import agent.models_dev as md
md._models_dev_cache = {}
md._models_dev_cache_time = 0
md._models_dev_retry_after = 0
md._models_dev_refresh_in_flight = False
yield
md._models_dev_cache = {}
md._models_dev_cache_time = 0
md._models_dev_retry_after = 0
md._models_dev_refresh_in_flight = False
@patch("agent.models_dev.requests.get")
def test_etag_sent_when_cached(self, mock_get):
"""If-None-Match header is sent when a cached ETag exists."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 200
response.json.return_value = SAMPLE_REGISTRY
response.headers = {"ETag": '"v2"'}
response.raise_for_status = MagicMock()
mock_get.return_value = response
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value='"v1"'), \
patch.object(md, "_save_etag"):
fetch_models_dev()
call_kwargs = mock_get.call_args
headers = call_kwargs.kwargs.get("headers", {})
assert headers.get("If-None-Match") == '"v1"'
@patch("agent.models_dev.requests.get")
def test_304_reconfirms_cache_freshness(self, mock_get):
"""A 304 Not Modified re-confirms the existing cache without download."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 304
mock_get.return_value = response
md._models_dev_cache = SAMPLE_REGISTRY
md._models_dev_cache_time = 0
md._models_dev_retry_after = time.time() + 100 # backoff was armed
with patch.object(md, "_load_etag", return_value='"v1"'), \
patch.object(md, "_save_etag"):
# Run the background worker synchronously
md._models_dev_refresh_in_flight = True
md._background_refresh_models_dev()
# Cache content unchanged
assert md._models_dev_cache == SAMPLE_REGISTRY
# Freshness timestamp advanced
assert md._models_dev_cache_time > 0
# Backoff cleared
assert md._models_dev_retry_after == 0
assert not md._models_dev_refresh_in_flight
# response.json() was never called — no body to parse
response.json.assert_not_called()
@patch("agent.models_dev.requests.get")
def test_foreground_304_returns_existing_cache(self, mock_get):
"""Foreground fetch with 304 returns the existing cache."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 304
mock_get.return_value = response
md._models_dev_cache = SAMPLE_REGISTRY
md._models_dev_cache_time = 0
md._models_dev_retry_after = 0
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_load_etag", return_value='"v1"'), \
patch.object(md, "_save_etag"):
result = fetch_models_dev(force_refresh=True)
assert result == SAMPLE_REGISTRY
assert md._models_dev_cache_time > 0
@patch("agent.models_dev.requests.get")
def test_new_etag_persisted_after_successful_fetch(self, mock_get):
"""A successful fetch with an ETag in the response persists it."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 200
response.json.return_value = SAMPLE_REGISTRY
response.headers = {"ETag": '"new-etag"'}
response.raise_for_status = MagicMock()
mock_get.return_value = response
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag") as mock_save_etag:
fetch_models_dev()
mock_save_etag.assert_called_once_with('"new-etag"')
@patch("agent.models_dev.requests.get")
def test_no_etag_header_sent_without_cached_etag(self, mock_get):
"""No If-None-Match header when no cached ETag exists."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 200
response.json.return_value = SAMPLE_REGISTRY
response.headers = {}
response.raise_for_status = MagicMock()
mock_get.return_value = response
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag"):
fetch_models_dev()
call_kwargs = mock_get.call_args
headers = call_kwargs.kwargs.get("headers", {})
assert "If-None-Match" not in headers
# ---------------------------------------------------------------------------
# Corrupt / invalid cache rejection
# ---------------------------------------------------------------------------
class TestCorruptCacheRejection:
"""A corrupt or empty disk cache must be rejected, not served as {}."""
def test_validate_registry_rejects_empty_dict(self):
assert not _validate_registry({})
def test_validate_registry_rejects_non_dict(self):
assert not _validate_registry("not a dict")
assert not _validate_registry(None)
assert not _validate_registry([])
def test_validate_registry_accepts_populated_dict(self):
assert _validate_registry({"anthropic": {}})
@patch("agent.models_dev.requests.get")
def test_corrupt_json_rejected_with_warning(self, mock_get, caplog):
"""Invalid JSON on disk is ignored, not served as {}."""
import agent.models_dev as md
import json as _json
mock_get.side_effect = OSError("unreachable")
md._models_dev_cache = {}
md._models_dev_cache_time = 0
with patch.object(md, "_disk_cache_age_seconds", return_value=0), \
patch.object(md, "_get_cache_path") as mock_path, \
patch.object(md, "_load_etag", return_value=""):
mock_path.return_value.exists.return_value = True
mock_path.return_value.open.return_value.__enter__.return_value.read.return_value = "not json"
# json.load will raise on invalid JSON
with patch("builtins.open", side_effect=_json.JSONDecodeError("msg", "doc", 0)):
with patch.object(md, "_load_disk_cache", wraps=md._load_disk_cache):
result = fetch_models_dev()
# Returns empty dict, not the corrupt data
assert result == {}
@patch("agent.models_dev.requests.get")
def test_empty_dict_cache_rejected(self, mock_get, caplog):
"""An empty dict in the cache file is rejected with a warning."""
import agent.models_dev as md
import logging
mock_get.side_effect = OSError("unreachable")
md._models_dev_cache = {}
md._models_dev_cache_time = 0
with patch.object(md, "_disk_cache_age_seconds", return_value=0), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_disk_cache"):
with caplog.at_level(logging.WARNING):
# _load_disk_cache returns {} for empty dict, which is correct
result = fetch_models_dev()
assert result == {}
# ---------------------------------------------------------------------------
# Mirror URL override via config
# ---------------------------------------------------------------------------
class TestMirrorUrlOverride:
"""models_dev.url config key overrides the API endpoint."""
@pytest.fixture(autouse=True)
def _reset_fetch_state(self):
import agent.models_dev as md
md._models_dev_cache = {}
md._models_dev_cache_time = 0
md._models_dev_retry_after = 0
md._models_dev_refresh_in_flight = False
yield
md._models_dev_cache = {}
md._models_dev_cache_time = 0
md._models_dev_retry_after = 0
md._models_dev_refresh_in_flight = False
@patch("agent.models_dev.requests.get")
def test_mirror_url_used_when_configured(self, mock_get):
"""When config has models_dev.url, requests.get hits that URL."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 200
response.json.return_value = SAMPLE_REGISTRY
response.headers = {}
response.raise_for_status = MagicMock()
mock_get.return_value = response
fake_config = {"models_dev": {"url": "https://mirror.example.com/api.json"}}
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag"), \
patch("hermes_cli.config.load_config_readonly", return_value=fake_config):
fetch_models_dev()
call_args = mock_get.call_args
assert "mirror.example.com" in call_args.args[0]
@patch("agent.models_dev.requests.get")
def test_default_url_used_when_not_configured(self, mock_get):
"""Without config override, the default models.dev URL is used."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 200
response.json.return_value = SAMPLE_REGISTRY
response.headers = {}
response.raise_for_status = MagicMock()
mock_get.return_value = response
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag"), \
patch("hermes_cli.config.load_config_readonly", return_value={}):
fetch_models_dev()
call_args = mock_get.call_args
assert "models.dev" in call_args.args[0]
@patch("agent.models_dev.requests.get")
def test_empty_url_falls_back_to_default(self, mock_get):
"""An empty string URL in config falls back to the default."""
import agent.models_dev as md
response = MagicMock()
response.status_code = 200
response.json.return_value = SAMPLE_REGISTRY
response.headers = {}
response.raise_for_status = MagicMock()
mock_get.return_value = response
fake_config = {"models_dev": {"url": ""}}
with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
patch.object(md, "_load_disk_cache", return_value={}), \
patch.object(md, "_save_disk_cache"), \
patch.object(md, "_load_etag", return_value=""), \
patch.object(md, "_save_etag"), \
patch("hermes_cli.config.load_config_readonly", return_value=fake_config):
fetch_models_dev()
call_args = mock_get.call_args
assert "models.dev" in call_args.args[0]
# ---------------------------------------------------------------------------
# No-network-on-hot-paths invariant
# ---------------------------------------------------------------------------
class TestNoNetworkOnHotPaths:
"""Query functions must default to allow_network=False on hot paths."""
@patch("agent.models_dev.requests.get")
def test_get_model_capabilities_default_no_network(self, mock_get):
"""get_model_capabilities defaults to allow_network=False."""
with patch("agent.models_dev.fetch_models_dev") as mock_fetch:
mock_fetch.return_value = CAPS_REGISTRY
get_model_capabilities("anthropic", "claude-sonnet-4")
# fetch_models_dev was called with allow_network=False
mock_fetch.assert_called_once_with(allow_network=False)
@patch("agent.models_dev.requests.get")
def test_get_model_info_default_no_network(self, mock_get):
"""get_model_info defaults to allow_network=False."""
with patch("agent.models_dev.fetch_models_dev") as mock_fetch:
mock_fetch.return_value = SAMPLE_REGISTRY
get_model_info("anthropic", "claude-opus-4-6")
mock_fetch.assert_called_once_with(allow_network=False)
@patch("agent.models_dev.requests.get")
def test_lookup_models_dev_context_default_no_network(self, mock_get):
"""lookup_models_dev_context defaults to allow_network=False."""
with patch("agent.models_dev.fetch_models_dev") as mock_fetch:
mock_fetch.return_value = SAMPLE_REGISTRY
lookup_models_dev_context("anthropic", "claude-opus-4-6")
mock_fetch.assert_called_once_with(allow_network=False)
@patch("agent.models_dev.requests.get")
def test_get_model_capabilities_explicit_network(self, mock_get):
"""get_model_capabilities can opt into network."""
with patch("agent.models_dev.fetch_models_dev") as mock_fetch:
mock_fetch.return_value = CAPS_REGISTRY
get_model_capabilities("anthropic", "claude-sonnet-4", allow_network=True)
mock_fetch.assert_called_once_with(allow_network=True)
# ---------------------------------------------------------------------------
# get_model_capabilities — vision via modalities.input
# ---------------------------------------------------------------------------