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

perf(dashboard): bound multi-profile sidebar polling

Browse Source
This commit is contained in:
Lucas Oliveira
2026-08-03 10:28:38 -03:00
committed by Teknium
parent 868c400e54
commit f0cfe5a56f
4 changed files with 423 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_routers/profiles.py
+212 -36
View File
@@ -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)
hermes_cli/web_server.py
+15 -2
View File
@@ -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)
tests/hermes_cli/test_cron_profile_enumeration_lightweight.py
+37
View File
@@ -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()
tests/hermes_cli/test_profiles_sidebar_cache.py
+159
View File
@@ -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()