From 5fa01f2fe0d9d2ab2dc67b6a2b48ba1becfc714f Mon Sep 17 00:00:00 2001 From: Robin Fernandes Date: Sun, 20 Sep 2026 09:21:52 +1000 Subject: [PATCH] fix(models): reduce repeated Nous recommendation traffic --- hermes_cli/models.py | 48 ++++--- .../test_nous_recommendation_disk_cache.py | 123 ++++++++++++++++++ 2 files changed, 155 insertions(+), 16 deletions(-) create mode 100644 tests/hermes_cli/test_nous_recommendation_disk_cache.py diff --git a/hermes_cli/models.py b/hermes_cli/models.py index 3ea93debd2..320ceb33dd 100644 --- a/hermes_cli/models.py +++ b/hermes_cli/models.py @@ -10,6 +10,7 @@ from __future__ import annotations import contextvars import copy +import gzip import json import logging import os @@ -85,7 +86,10 @@ def _get_json( call time so monkeypatching ``_urlopen_model_catalog_request`` still applies). Raises on failure.""" req = urllib.request.Request(url, headers=headers or {}) with (opener or _urlopen_model_catalog_request)(req, timeout=timeout, **open_kwargs) as resp: - return json.loads(resp.read().decode()) + body = resp.read() + if req.get_header("Accept-encoding") == "gzip" and resp.headers.get("Content-Encoding", "").lower() == "gzip": + body = gzip.decompress(body) + return json.loads(body.decode()) def _read_json_cache(path: Path, *, errors=Exception) -> Optional[dict]: @@ -322,8 +326,8 @@ def check_nous_free_tier(*, force_fresh: bool = False, cached_only: bool = False NOUS_RECOMMENDED_MODELS_PATH = "/api/nous/recommended-models" _NOUS_RECOMMENDED_CACHE_TTL: int = 600 # seconds (10 minutes) -# (result_dict, timestamp) keyed by portal_base_url so staging vs prod don't collide. -_nous_recommended_cache: dict[str, tuple[dict[str, Any], float]] = {} +# (result_dict, monotonic timestamp), scoped to the profile and portal. +_nous_recommended_cache: dict[tuple[str, str], tuple[dict[str, Any], float]] = {} def _nous_recommended_disk_path() -> "Path": @@ -331,13 +335,18 @@ def _nous_recommended_disk_path() -> "Path": return get_hermes_home() / "cache" / "nous_recommended_cache.json" -def _read_nous_recommended_disk(base: str) -> dict[str, Any] | None: - """Last-known-good payload for ``base`` from the per-base disk map - ``{"": {"data": {...}, "ts": }}`` (staging and prod don't collide), or None.""" - blob = _read_json_cache(_nous_recommended_disk_path(), errors=(OSError, json.JSONDecodeError)) +def _read_nous_recommended_disk(base: str) -> tuple[dict[str, Any], float] | None: + """Return the last good payload and its age for the freshness check.""" + blob = _read_json_cache(_nous_recommended_disk_path(), errors=(OSError, json.JSONDecodeError, UnicodeDecodeError)) entry = (blob or {}).get(base) data = entry.get("data") if isinstance(entry, dict) else None - return data if isinstance(data, dict) and data else None + if not isinstance(data, dict) or not data: + return None + try: + age = time.time() - float(entry.get("ts", 0)) + except (TypeError, ValueError, OverflowError): + age = float("inf") + return data, age def _write_nous_recommended_disk(base: str, data: dict[str, Any]) -> None: @@ -347,7 +356,7 @@ def _write_nous_recommended_disk(base: str, data: dict[str, Any]) -> None: return path = _nous_recommended_disk_path() try: - blob = _read_json_cache(path, errors=(OSError, json.JSONDecodeError)) or {} + blob = _read_json_cache(path, errors=(OSError, json.JSONDecodeError, UnicodeDecodeError)) or {} blob[base] = {"data": data, "ts": time.time()} _write_json_cache(path, blob, indent=2) except OSError as exc: @@ -359,18 +368,25 @@ def fetch_nous_recommended_models( ) -> dict[str, Any]: """Fetch the Portal's public ``/api/nous/recommended-models`` payload (no auth). - Cached per portal URL for ``_NOUS_RECOMMENDED_CACHE_TTL`` seconds in process (``force_refresh`` - bypasses); a successful fetch is also persisted as last-known-good on disk, which serves a live - failure so a transient Portal hiccup doesn't drop the recommendations. + Reuse successful results for ``_NOUS_RECOMMENDED_CACHE_TTL`` seconds, including across + process restarts. ``force_refresh`` bypasses both caches. Stale disk data remains a fallback + on live failure; reading it never renews its freshness. """ base = (portal_base_url or "https://portal.nousresearch.com").rstrip("/") now = time.monotonic() - cached = _nous_recommended_cache.get(base) + cache_key = (_pricing_profile_key(), base) + cached = _nous_recommended_cache.get(cache_key) if not force_refresh and cached is not None and now - cached[1] < _NOUS_RECOMMENDED_CACHE_TTL: return cached[0] + disk = _read_nous_recommended_disk(base) + if not force_refresh and disk is not None and 0 <= disk[1] < _NOUS_RECOMMENDED_CACHE_TTL: + data, age = disk + _nous_recommended_cache[cache_key] = (data, now - age) + return data try: data = _get_json( - f"{base}{NOUS_RECOMMENDED_MODELS_PATH}", timeout=timeout, headers={"Accept": "application/json"} + f"{base}{NOUS_RECOMMENDED_MODELS_PATH}", timeout=timeout, + headers={"Accept": "application/json", "Accept-Encoding": "gzip"} ) if not isinstance(data, dict): data = {} @@ -379,8 +395,8 @@ def fetch_nous_recommended_models( if data: _write_nous_recommended_disk(base, data) else: - data = _read_nous_recommended_disk(base) or data - _nous_recommended_cache[base] = (data, now) + data = disk[0] if disk is not None else data + _nous_recommended_cache[cache_key] = (data, now) return data diff --git a/tests/hermes_cli/test_nous_recommendation_disk_cache.py b/tests/hermes_cli/test_nous_recommendation_disk_cache.py new file mode 100644 index 0000000000..fafc98349f --- /dev/null +++ b/tests/hermes_cli/test_nous_recommendation_disk_cache.py @@ -0,0 +1,123 @@ +"""Recommendation requests reuse disk entries and negotiate gzip across real processes.""" +import gzip +import json +import os +from pathlib import Path +import subprocess +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + + +@pytest.fixture +def portal(): + requests = [] + state = {"version": 1, "status": 200, "compress": True} + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + requests.append(self.headers.get("Accept-Encoding")) + body = json.dumps({"paidRecommendedModels": [{"modelName": f"model-{state['version']}"}]}).encode() + compressed = state["compress"] and requests[-1] == "gzip" + if compressed: + body = gzip.compress(body) + self.send_response(state["status"]) + self.send_header("Content-Type", "application/json") + if compressed: + self.send_header("Content-Encoding", "gzip") + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", requests, state + finally: + server.shutdown() + server.server_close() + thread.join() + + +def fetch_in_new_process(home, base, *, force=False): + result = subprocess.run( + [sys.executable, "-c", "from hermes_cli.models import fetch_nous_recommended_models; " + "import json, sys; print(json.dumps(fetch_nous_recommended_models(sys.argv[1], force_refresh=sys.argv[2] == 'True')))", + base, str(force)], + cwd=Path(__file__).resolve().parents[2], + env={**os.environ, "HERMES_HOME": str(home)}, + capture_output=True, text=True, check=True, timeout=30, + ) + return json.loads(result.stdout) + + +def test_process_restarts_reuse_fresh_disk_but_force_and_expiry_fetch(tmp_path, portal): + base, requests, state = portal + home = tmp_path / "client" + first = fetch_in_new_process(home, base) + assert fetch_in_new_process(home, base) == first + assert requests == ["gzip"] + + # A server may ignore Accept-Encoding; the refresh must also accept plain JSON. + state.update(version=2, compress=False) + second = fetch_in_new_process(home, base, force=True) + assert second and second != first + assert len(requests) == 2 + + path = home / "cache" / "nous_recommended_cache.json" + disk = json.loads(path.read_text(encoding="utf-8")) + disk[base]["ts"] = time.time() - 3600 + path.write_text(json.dumps(disk), encoding="utf-8") + state["version"] = 3 + assert fetch_in_new_process(home, base) != second + assert len(requests) == 3 + + # A failed refresh retains the last good payload without making it fresh on disk. + disk = json.loads(path.read_text(encoding="utf-8")) + disk[base]["ts"] = time.time() - 3600 + path.write_text(json.dumps(disk), encoding="utf-8") + state["status"] = 503 + assert fetch_in_new_process(home, base) == disk[base]["data"] + assert json.loads(path.read_text(encoding="utf-8"))[base]["ts"] == disk[base]["ts"] + + # Bad local cache bytes/metadata must never stop a healthy API refresh. + state["status"] = 200 + path.write_bytes(b"\xff") + assert fetch_in_new_process(home, base) == disk[base]["data"] + for timestamp in [None, True, "invalid", float("nan"), time.time() + 3600, 10 ** 400]: + disk[base]["ts"] = timestamp + path.write_text(json.dumps(disk), encoding="utf-8") + before = len(requests) + assert fetch_in_new_process(home, base) == disk[base]["data"] + assert len(requests) == before + 1 + + +def test_disk_cache_is_scoped_to_home_and_portal(tmp_path, portal): + base, requests, state = portal + a, b = tmp_path / "a", tmp_path / "b" + first = fetch_in_new_process(a, base) + state["version"] = 2 + assert fetch_in_new_process(b, base) != first + assert fetch_in_new_process(a, base) == first + assert len(requests) == 2 + assert fetch_in_new_process(a, base + "/staging") != first + assert len(requests) == 3 + + # The same process can switch profiles under the multiplex gateway too. + from hermes_cli.models import fetch_nous_recommended_models, _nous_recommended_cache + from hermes_constants import set_hermes_home_override, reset_hermes_home_override + _nous_recommended_cache.clear() + for home, expected in [(a, first), (b, {"paidRecommendedModels": [{"modelName": "model-2"}]}), (a, first)]: + token = set_hermes_home_override(home) + try: + assert fetch_nous_recommended_models(base) == expected + finally: + reset_hermes_home_override(token) + assert len(requests) == 3 + _nous_recommended_cache.clear()