diff --git a/hermes_cli/web_routers/profiles.py b/hermes_cli/web_routers/profiles.py index 145806c6eb..b99bbcabe3 100644 --- a/hermes_cli/web_routers/profiles.py +++ b/hermes_cli/web_routers/profiles.py @@ -13,12 +13,17 @@ late-binding seam in :mod:`hermes_cli.web_deps` so tests that """ import asyncio # noqa: F401 — used by handlers +import copy +import functools +import inspect import json import logging import re import subprocess # noqa: F401 import sys # noqa: F401 +import threading import time # noqa: F401 +from collections import OrderedDict from pathlib import Path # noqa: F401 from typing import Any, Dict, List, Optional, Tuple # noqa: F401 @@ -79,6 +84,151 @@ _write_profile_mcp_servers = late("_write_profile_mcp_servers") _write_profile_model = late("_write_profile_model") +def _read_sidebar_cache_ttl() -> float: + """Return the bounded cache lifetime for the expensive sidebar scan.""" + raw = os.environ.get("HERMES_DASHBOARD_SIDEBAR_CACHE_TTL", "5") + try: + value = float(raw) + if not math.isfinite(value): + raise ValueError("non-finite TTL") + except (TypeError, ValueError): + _log.warning( + "invalid HERMES_DASHBOARD_SIDEBAR_CACHE_TTL=%r; using 5s", + raw, + ) + value = 5.0 + return min(max(value, 0.0), 30.0) + + +_SIDEBAR_CACHE_TTL_SECONDS = _read_sidebar_cache_ttl() +_SIDEBAR_CACHE_MAX_ENTRIES = 32 +_SIDEBAR_PROFILE_CACHE_MAX_ENTRIES = 256 +_SIDEBAR_PROFILE_CACHE = OrderedDict() +_SIDEBAR_PROFILE_CACHE_LOCK = threading.Lock() + + +def _stat_fingerprint(path: Path): + """Return identity + mutation metadata without opening the file.""" + try: + stat = path.stat() + except OSError: + return None + return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns) + + +def _sidebar_db_fingerprint(db_path: Path): + """Track SQLite content changes through the main DB and its WAL.""" + wal_path = Path(f"{db_path}-wal") + return (_stat_fingerprint(db_path), _stat_fingerprint(wal_path)) + + +def _sidebar_profile_cache_get(key): + with _SIDEBAR_PROFILE_CACHE_LOCK: + value = _SIDEBAR_PROFILE_CACHE.get(key) + if value is None: + return None + _SIDEBAR_PROFILE_CACHE.move_to_end(key) + return copy.deepcopy(value) + + +def _sidebar_profile_cache_put(key, value): + db_path, fingerprint = key[:2] + snapshot = copy.deepcopy(value) + with _SIDEBAR_PROFILE_CACHE_LOCK: + # A changed DB/WAL makes all older parameter variants for that profile + # obsolete. Remove them eagerly rather than waiting for LRU pressure. + stale = [ + existing + for existing in _SIDEBAR_PROFILE_CACHE + if existing[0] == db_path and existing[1] != fingerprint + ] + for existing in stale: + _SIDEBAR_PROFILE_CACHE.pop(existing, None) + _SIDEBAR_PROFILE_CACHE[key] = snapshot + _SIDEBAR_PROFILE_CACHE.move_to_end(key) + while len(_SIDEBAR_PROFILE_CACHE) > _SIDEBAR_PROFILE_CACHE_MAX_ENTRIES: + _SIDEBAR_PROFILE_CACHE.popitem(last=False) + + +def _sidebar_profile_cache_clear(): + with _SIDEBAR_PROFILE_CACHE_LOCK: + _SIDEBAR_PROFILE_CACHE.clear() + + +def _sidebar_singleflight_cache(func): + """Coalesce concurrent sidebar scans and briefly reuse their response. + + Every uncached refresh opens every profile database and runs up to three + session queries per profile. Desktop reconnect/focus/change bursts can + therefore overlap several identical scans in AnyIO worker threads, which + amplifies YAML/SQLite work and starves the uvicorn event loop for the GIL. + + The short TTL bounds UI staleness while the single-flight lock guarantees + only one expensive scan runs at a time. Cached values are copied on store + and hit so FastAPI serialization or a caller cannot mutate shared state. + """ + signature = inspect.signature(func) + cache = OrderedDict() + cache_lock = threading.Lock() + refresh_lock = threading.Lock() + miss = object() + + def _key(args, kwargs): + bound = signature.bind(*args, **kwargs) + bound.apply_defaults() + return tuple(bound.arguments.items()) + + def _lookup(key): + now = time.monotonic() + with cache_lock: + item = cache.get(key) + if item is None: + return miss + expires_at, value = item + if now >= expires_at: + cache.pop(key, None) + return miss + cache.move_to_end(key) + return copy.deepcopy(value) + + @functools.wraps(func) + def wrapped(*args, **kwargs): + ttl = _SIDEBAR_CACHE_TTL_SECONDS + if ttl <= 0: + return func(*args, **kwargs) + + key = _key(args, kwargs) + cached = _lookup(key) + if cached is not miss: + return cached + + # A plain Lock is intentional: FastAPI executes this sync handler in + # the AnyIO worker pool, so contenders sleep without holding the GIL. + with refresh_lock: + cached = _lookup(key) + if cached is not miss: + return cached + result = func(*args, **kwargs) + try: + snapshot = copy.deepcopy(result) + except Exception: + _log.exception("sidebar response could not be cached") + return result + with cache_lock: + cache[key] = (time.monotonic() + ttl, snapshot) + cache.move_to_end(key) + while len(cache) > _SIDEBAR_CACHE_MAX_ENTRIES: + cache.popitem(last=False) + return result + + def cache_clear(): + with cache_lock: + cache.clear() + + wrapped.cache_clear = cache_clear + return wrapped + + @sessions_router.get("/api/profiles/sessions") def get_profiles_sessions( # ``le=500`` caps the per-request page size (idea from #39200) — this @@ -123,8 +273,9 @@ def get_profiles_sessions( targets.append((name, home)) else: try: - infos = profiles_mod.list_profiles() - targets = [(info.name, info.path) for info in infos] + # This endpoint only needs name/path. Avoid list_profiles(), which + # parses config/meta and probes gateways/skills per profile. + targets = profiles_mod.profiles_to_serve(multiplex=True) except Exception: _log.exception("GET /api/profiles/sessions: list_profiles failed") targets = [] @@ -230,6 +381,7 @@ def get_profiles_sessions( @sessions_router.get("/api/profiles/sessions/sidebar") +@_sidebar_singleflight_cache def get_profiles_sessions_sidebar( recents_profile: str = "all", recents_limit: int = 20, @@ -263,8 +415,9 @@ def get_profiles_sessions_sidebar( from hermes_cli import profiles as profiles_mod try: - infos = profiles_mod.list_profiles() - targets: List[Tuple[str, Path]] = [(info.name, info.path) for info in infos] + # Session aggregation only needs name/path; the lightweight enumerator + # avoids YAML/meta/gateway/skill probes for all profiles per refresh. + targets: List[Tuple[str, Path]] = profiles_mod.profiles_to_serve(multiplex=True) except Exception: _log.exception("GET /api/profiles/sessions/sidebar: list_profiles failed") targets = [] @@ -323,38 +476,61 @@ def get_profiles_sessions_sidebar( db_path = Path(home) / "state.db" if not db_path.exists(): continue - try: - # Read-only with the stale-schema heal — same contract as the - # per-slice endpoint above (one-time writable reconcile when the - # store predates a schema addition, plain read-only otherwise). - db = _open_session_db_at_path(db_path, read_only=True) - except Exception as exc: - _warn_profile_read_error(name, exc) - errors.append({"profile": name, "error": str(exc)}) - continue - try: - profile_rows = _slice(db, exclude=recents_exclude_list, cap=recents_cap) - # A full window means more rows remain on disk. That is all the - # sidebar's "load more" needs, and unlike an exact COUNT(*) per - # profile per refresh it costs nothing beyond the rows already - # read. Discount pinned back-fills — they arrive past the LIMIT - # and would otherwise fake a full page on a short list. - unpinned_count = sum(1 for s in profile_rows if not s.get("pinned")) - recents_truncated[name] = unpinned_count >= recents_cap - recents_rows.extend(_tag(profile_rows, name)) - # Aggregated in SQL rather than over the window above: the window is - # a page, and a total that shrank when you scrolled would be worse - # than no total at all. - profile_totals[name] = db.usage_totals() - cron_rows.extend(_tag(_slice(db, source="cron", cap=cron_cap), name)) - messaging_rows.extend( - _tag(_slice(db, exclude=messaging_exclude_list, cap=messaging_cap), name) - ) - except Exception as exc: - _warn_profile_read_error(name, exc) - errors.append({"profile": name, "error": str(exc)}) - finally: - db.close() + fingerprint = _sidebar_db_fingerprint(db_path) + profile_cache_key = ( + str(db_path), + fingerprint, + recents_cap, + tuple(recents_exclude_list), + cron_cap, + messaging_cap, + tuple(messaging_exclude_list), + ) + slices = _sidebar_profile_cache_get(profile_cache_key) + if slices is None: + try: + # Read-only with the stale-schema heal — same contract as the + # per-slice endpoint above (one-time writable reconcile when the + # store predates a schema addition, plain read-only otherwise). + db = _open_session_db_at_path(db_path, read_only=True) + except Exception as exc: + _warn_profile_read_error(name, exc) + errors.append({"profile": name, "error": str(exc)}) + continue + try: + slices = { + "recents": _slice(db, exclude=recents_exclude_list, cap=recents_cap), + # Aggregated in SQL rather than over the recents window: the + # window is a page, and a total that shrank when you scrolled + # would be worse than no total at all. + "usage": db.usage_totals(), + "cron": _slice(db, source="cron", cap=cron_cap), + "messaging": _slice( + db, + exclude=messaging_exclude_list, + cap=messaging_cap, + ), + } + _sidebar_profile_cache_put(profile_cache_key, slices) + except Exception as exc: + _warn_profile_read_error(name, exc) + errors.append({"profile": name, "error": str(exc)}) + continue + finally: + db.close() + + profile_rows = slices["recents"] + # A full window means more rows remain on disk. That is all the + # sidebar's "load more" needs, and unlike an exact COUNT(*) per + # profile per refresh it costs nothing beyond the rows already + # read. Discount pinned back-fills — they arrive past the LIMIT + # and would otherwise fake a full page on a short list. + unpinned_count = sum(1 for s in profile_rows if not s.get("pinned")) + recents_truncated[name] = unpinned_count >= recents_cap + recents_rows.extend(_tag(profile_rows, name)) + profile_totals[name] = slices["usage"] + cron_rows.extend(_tag(slices["cron"], name)) + messaging_rows.extend(_tag(slices["messaging"], name)) def _window(rows: List[Dict[str, Any]], cap: int) -> List[Dict[str, Any]]: rows.sort(key=lambda s: s.get("last_active") or s.get("started_at") or 0, reverse=True) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 1ed365fe4f..3f527ed0f6 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -11997,10 +11997,23 @@ def _validate_dashboard_cron_context_from( def _cron_profile_dicts() -> List[Dict[str, Any]]: - """Return dashboard profile records, falling back to a directory scan.""" + """Return the minimal profile records needed by cron aggregation. + + The two callers only consume ``name``. ``list_profiles()`` also parses + config/distribution metadata, probes gateway processes, and counts skills + for every profile; polling cron jobs through that path creates avoidable + GIL pressure on large profile pools. + """ from hermes_cli import profiles as profiles_mod try: - return [_profile_to_dict(p) for p in profiles_mod.list_profiles()] + return [ + { + "name": name, + "path": str(home), + "is_default": name == "default", + } + for name, home in profiles_mod.profiles_to_serve(multiplex=True) + ] except Exception: _log.exception("Failed to list profiles for cron dashboard; falling back to directory scan") return _fallback_profile_dicts(profiles_mod) diff --git a/tests/hermes_cli/test_cron_profile_enumeration_lightweight.py b/tests/hermes_cli/test_cron_profile_enumeration_lightweight.py new file mode 100644 index 0000000000..b6de5a2382 --- /dev/null +++ b/tests/hermes_cli/test_cron_profile_enumeration_lightweight.py @@ -0,0 +1,37 @@ +"""Cron aggregation must not perform full profile metadata scans.""" + +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from hermes_cli import web_server + + +class CronProfileEnumerationTests(unittest.TestCase): + def test_uses_lightweight_name_path_enumerator(self): + with tempfile.TemporaryDirectory() as root: + homes = [ + ("default", Path(root)), + ("coder-01", Path(root) / "profiles" / "coder-01"), + ] + with ( + mock.patch( + "hermes_cli.profiles.profiles_to_serve", + return_value=homes, + ) as lightweight, + mock.patch( + "hermes_cli.profiles.list_profiles", + side_effect=AssertionError("full profile scan is forbidden"), + ), + ): + result = web_server._cron_profile_dicts() + + lightweight.assert_called_once_with(multiplex=True) + self.assertEqual([item["name"] for item in result], ["default", "coder-01"]) + self.assertTrue(result[0]["is_default"]) + self.assertFalse(result[1]["is_default"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/hermes_cli/test_profiles_sidebar_cache.py b/tests/hermes_cli/test_profiles_sidebar_cache.py new file mode 100644 index 0000000000..76d96b418f --- /dev/null +++ b/tests/hermes_cli/test_profiles_sidebar_cache.py @@ -0,0 +1,159 @@ +"""Regression tests for dashboard sidebar scan coalescing.""" + +import inspect +import tempfile +import threading +import time +import unittest +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from unittest import mock + +from hermes_cli.web_routers import profiles + + +class SidebarCacheTests(unittest.TestCase): + def setUp(self): + patcher = mock.patch.object(profiles, "_SIDEBAR_CACHE_TTL_SECONDS", 5.0) + patcher.start() + self.addCleanup(patcher.stop) + profiles._sidebar_profile_cache_clear() + self.addCleanup(profiles._sidebar_profile_cache_clear) + + def test_invalid_or_non_finite_ttl_falls_back_to_default(self): + for raw in ("invalid", "nan", "inf", "-inf"): + with self.subTest(raw=raw): + with mock.patch.dict(profiles.os.environ, {"HERMES_DASHBOARD_SIDEBAR_CACHE_TTL": raw}): + self.assertEqual(profiles._read_sidebar_cache_ttl(), 5.0) + + def test_profile_cache_uses_db_and_wal_fingerprint_and_defensive_copies(self): + with tempfile.TemporaryDirectory() as root: + db_path = Path(root) / "state.db" + wal_path = Path(f"{db_path}-wal") + db_path.write_bytes(b"db-v1") + wal_path.write_bytes(b"wal-v1") + first_fingerprint = profiles._sidebar_db_fingerprint(db_path) + first_key = (str(db_path), first_fingerprint, False, 0, (), 50, 100, ()) + payload = {"recents": None, "cron": [{"id": "one"}], "messaging": []} + + profiles._sidebar_profile_cache_put(first_key, payload) + cached = profiles._sidebar_profile_cache_get(first_key) + cached["cron"][0]["id"] = "mutated" + self.assertEqual( + profiles._sidebar_profile_cache_get(first_key)["cron"][0]["id"], + "one", + ) + + wal_path.write_bytes(b"wal-v2-is-different") + second_fingerprint = profiles._sidebar_db_fingerprint(db_path) + second_key = (str(db_path), second_fingerprint, False, 0, (), 50, 100, ()) + self.assertNotEqual(first_fingerprint, second_fingerprint) + self.assertIsNone(profiles._sidebar_profile_cache_get(second_key)) + + profiles._sidebar_profile_cache_put(second_key, payload) + self.assertIsNone(profiles._sidebar_profile_cache_get(first_key)) + + def test_profile_cache_is_lru_bounded(self): + with mock.patch.object(profiles, "_SIDEBAR_PROFILE_CACHE_MAX_ENTRIES", 2): + for index in range(3): + key = (f"/db/{index}", (index, None), False, 0, (), 50, 100, ()) + profiles._sidebar_profile_cache_put(key, {"index": index}) + self.assertEqual(len(profiles._SIDEBAR_PROFILE_CACHE), 2) + + def test_applies_defaults_and_returns_defensive_copies(self): + calls = 0 + + @profiles._sidebar_singleflight_cache + def scan(profile="all", limit=20): + nonlocal calls + calls += 1 + return {"profile": profile, "rows": [{"limit": limit}]} + + first = scan() + first["rows"][0]["limit"] = 999 + second = scan(profile="all", limit=20) + + self.assertEqual(calls, 1) + self.assertEqual(second, {"profile": "all", "rows": [{"limit": 20}]}) + + def test_coalesces_concurrent_identical_scans(self): + workers = 12 + entered = threading.Event() + release = threading.Event() + calls = 0 + calls_lock = threading.Lock() + + @profiles._sidebar_singleflight_cache + def scan(profile="all"): + nonlocal calls + with calls_lock: + calls += 1 + entered.set() + self.assertTrue(release.wait(timeout=2)) + return {"profile": profile, "rows": []} + + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = [pool.submit(scan, "default") for _ in range(workers)] + self.assertTrue(entered.wait(timeout=1)) + time.sleep(0.05) + release.set() + results = [future.result(timeout=2) for future in futures] + + self.assertEqual(calls, 1) + self.assertEqual(results, [{"profile": "default", "rows": []}] * workers) + + def test_expires(self): + clock = iter((100.0, 100.0, 100.0, 106.0, 106.0, 106.0)) + calls = 0 + + @profiles._sidebar_singleflight_cache + def scan(): + nonlocal calls + calls += 1 + return {"generation": calls} + + with mock.patch.object(profiles.time, "monotonic", side_effect=clock): + self.assertEqual(scan(), {"generation": 1}) + self.assertEqual(scan(), {"generation": 2}) + self.assertEqual(calls, 2) + + def test_does_not_cache_failures(self): + calls = 0 + + @profiles._sidebar_singleflight_cache + def scan(): + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("transient") + return {"ok": True} + + with self.assertRaisesRegex(RuntimeError, "transient"): + scan() + self.assertEqual(scan(), {"ok": True}) + self.assertEqual(scan(), {"ok": True}) + self.assertEqual(calls, 2) + + def test_can_be_disabled(self): + calls = 0 + + @profiles._sidebar_singleflight_cache + def scan(): + nonlocal calls + calls += 1 + return calls + + with mock.patch.object(profiles, "_SIDEBAR_CACHE_TTL_SECONDS", 0.0): + self.assertEqual((scan(), scan()), (1, 2)) + + def test_preserves_fastapi_signature(self): + def scan(profile: str = "all", limit: int = 20): + return profile, limit + + wrapped = profiles._sidebar_singleflight_cache(scan) + + self.assertEqual(inspect.signature(wrapped), inspect.signature(scan)) + + +if __name__ == "__main__": + unittest.main()