fix: speed up session index metadata refresh
This commit is contained in:
parent
95cee09c31
commit
017179b189
2 changed files with 262 additions and 18 deletions
135
api/models.py
135
api/models.py
|
|
@ -99,6 +99,39 @@ def _cleanup_stale_tmp_files() -> None:
|
|||
pass # SESSION_DIR may not exist yet; that's fine
|
||||
|
||||
|
||||
_PERSISTED_SESSION_IDS_CACHE: tuple[Path | None, int | None, frozenset[str]] = (None, None, frozenset())
|
||||
|
||||
|
||||
def _persisted_session_ids_snapshot() -> frozenset[str]:
|
||||
"""Return persisted session ids, caching the directory snapshot by mtime.
|
||||
|
||||
`/api/sessions` and incremental index writes may run every few seconds. A
|
||||
full `SESSION_DIR.glob('*.json')` on a large session directory is expensive,
|
||||
and doing that scan while request threads contend on LOCK makes the sidebar
|
||||
look like it was designed by a committee of glaciers. Cache the listing until
|
||||
the directory mtime changes, and let callers take the snapshot before
|
||||
entering critical sections.
|
||||
"""
|
||||
global _PERSISTED_SESSION_IDS_CACHE
|
||||
try:
|
||||
dir_mtime_ns = SESSION_DIR.stat().st_mtime_ns
|
||||
except Exception:
|
||||
dir_mtime_ns = None
|
||||
cached_dir, cached_mtime_ns, cached_ids = _PERSISTED_SESSION_IDS_CACHE
|
||||
if cached_dir == SESSION_DIR and cached_mtime_ns == dir_mtime_ns:
|
||||
return cached_ids
|
||||
try:
|
||||
ids = frozenset(
|
||||
p.stem
|
||||
for p in SESSION_DIR.glob('*.json')
|
||||
if not p.name.startswith('_')
|
||||
)
|
||||
except Exception:
|
||||
ids = frozenset()
|
||||
_PERSISTED_SESSION_IDS_CACHE = (SESSION_DIR, dir_mtime_ns, ids)
|
||||
return ids
|
||||
|
||||
|
||||
def _rebuild_session_index_background() -> None:
|
||||
try:
|
||||
_write_session_index(updates=None)
|
||||
|
|
@ -210,18 +243,13 @@ def _write_session_index(updates=None):
|
|||
# This avoids loading every session file on every single save().
|
||||
_fallback = False
|
||||
try:
|
||||
# Avoid N filesystem exists() checks under LOCK by collecting
|
||||
# on-disk IDs once before entering the critical section.
|
||||
on_disk_ids = _persisted_session_ids_snapshot()
|
||||
with LOCK:
|
||||
existing = json.loads(SESSION_INDEX_FILE.read_text(encoding='utf-8'))
|
||||
in_memory_ids = set(SESSIONS.keys())
|
||||
|
||||
# Avoid N filesystem exists() checks under LOCK by collecting
|
||||
# on-disk IDs once.
|
||||
on_disk_ids = {
|
||||
p.stem
|
||||
for p in SESSION_DIR.glob('*.json')
|
||||
if not p.name.startswith('_')
|
||||
}
|
||||
|
||||
existing = [
|
||||
e for e in existing
|
||||
if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids)
|
||||
|
|
@ -786,8 +814,10 @@ class Session:
|
|||
parsed['messages'] = []
|
||||
parsed['tool_calls'] = []
|
||||
session = cls(**parsed)
|
||||
index_message_count = _lookup_index_message_count(sid)
|
||||
sidecar_message_count = _parse_nonnegative_int(parsed.get('message_count'))
|
||||
index_message_count = None
|
||||
if sidecar_message_count is None:
|
||||
index_message_count = _lookup_index_message_count(sid)
|
||||
# The sidebar index is a cache and can lag behind external sidecar
|
||||
# appends/backfills. Prefer the largest known count so metadata-only
|
||||
# active-session polls do not miss real remote updates.
|
||||
|
|
@ -2530,6 +2560,77 @@ def _strip_sidebar_internal_flags(sessions: list[dict]) -> None:
|
|||
session.pop('_show_pre_compression_snapshot', None)
|
||||
|
||||
|
||||
def _row_may_need_sidecar_metadata_refresh(session: dict) -> bool:
|
||||
"""Return True when a row needs canonical sidecar runtime/snapshot metadata.
|
||||
|
||||
Compression lineage fields are enriched from state.db in one batched query
|
||||
later in all_sessions(). Loading hundreds of lineage sidecars on every
|
||||
/api/sessions poll turns the sidebar into molasses, so keep this refresh
|
||||
limited to the few rows whose transient runtime or snapshot state is not
|
||||
cheaply available from state.db.
|
||||
"""
|
||||
is_runtime_row = bool(
|
||||
session.get('active_stream_id')
|
||||
or session.get('has_pending_user_message')
|
||||
or session.get('pending_user_message')
|
||||
)
|
||||
snapshot_missing_sidebar_metadata = bool(
|
||||
session.get('pre_compression_snapshot')
|
||||
and (
|
||||
session.get('message_count') is None
|
||||
or session.get('last_message_at') is None
|
||||
)
|
||||
)
|
||||
return is_runtime_row or snapshot_missing_sidebar_metadata
|
||||
|
||||
|
||||
def _refresh_index_rows_from_sidecar_metadata(sessions: list[dict]) -> list[dict]:
|
||||
"""Overlay fuller sidecar metadata onto stale sidebar index rows.
|
||||
|
||||
``_index.json`` is a cache and can lag behind the canonical session sidecar
|
||||
during compression/continuation writes. Keep this read-only and limited to
|
||||
lineage/runtime-shaped rows so ordinary sidebar refreshes do not scan every
|
||||
historical transcript.
|
||||
"""
|
||||
out: list[dict] = []
|
||||
for session in sessions:
|
||||
if not _row_may_need_sidecar_metadata_refresh(session):
|
||||
out.append(session)
|
||||
continue
|
||||
sid = session.get('session_id')
|
||||
if not sid:
|
||||
out.append(session)
|
||||
continue
|
||||
sidecar = Session.load_metadata_only(sid)
|
||||
if not sidecar:
|
||||
out.append(session)
|
||||
continue
|
||||
compact = sidecar.compact(include_runtime=True)
|
||||
refreshed = dict(session)
|
||||
for key in (
|
||||
'message_count', 'updated_at', 'last_message_at', 'title', 'workspace',
|
||||
'model', 'model_provider', 'created_at', 'pinned', 'archived', 'project_id',
|
||||
'profile', 'pre_compression_snapshot', 'parent_session_id', 'source_tag',
|
||||
'raw_source', 'session_source', 'source_label', 'active_stream_id',
|
||||
'has_pending_user_message', 'pending_user_message', 'pending_started_at',
|
||||
):
|
||||
value = compact.get(key)
|
||||
if value is not None:
|
||||
refreshed[key] = value
|
||||
try:
|
||||
refreshed['message_count'] = max(
|
||||
int(session.get('message_count') or 0),
|
||||
int(compact.get('message_count') or 0),
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
if _session_sort_timestamp(compact) > _session_sort_timestamp(session):
|
||||
refreshed['updated_at'] = compact.get('updated_at', refreshed.get('updated_at'))
|
||||
refreshed['last_message_at'] = compact.get('last_message_at', refreshed.get('last_message_at'))
|
||||
out.append(refreshed)
|
||||
return out
|
||||
|
||||
|
||||
def _active_state_db_path() -> Path:
|
||||
"""Return state.db for the active Hermes profile, degrading to HERMES_HOME."""
|
||||
try:
|
||||
|
|
@ -2607,14 +2708,7 @@ def all_sessions(diag=None):
|
|||
_diag_stage(diag, "all_sessions.prune_index")
|
||||
with LOCK:
|
||||
in_memory_ids = set(SESSIONS.keys())
|
||||
try:
|
||||
persisted_ids = {
|
||||
p.stem
|
||||
for p in SESSION_DIR.glob('*.json')
|
||||
if not p.name.startswith('_')
|
||||
}
|
||||
except Exception:
|
||||
persisted_ids = None
|
||||
persisted_ids = _persisted_session_ids_snapshot()
|
||||
index = [
|
||||
s for s in index
|
||||
if (
|
||||
|
|
@ -2658,6 +2752,13 @@ def all_sessions(diag=None):
|
|||
include_runtime=True,
|
||||
active_stream_ids=active_stream_ids,
|
||||
)
|
||||
_diag_stage(diag, "all_sessions.refresh_sidecar_metadata")
|
||||
refreshed_index_rows = _refresh_index_rows_from_sidecar_metadata(list(index_map.values()))
|
||||
index_map = {
|
||||
row['session_id']: row
|
||||
for row in refreshed_index_rows
|
||||
if row.get('session_id')
|
||||
}
|
||||
_diag_stage(diag, "all_sessions.sort_filter")
|
||||
result = sorted(index_map.values(), key=lambda s: (s.get('pinned', False), _session_sort_timestamp(s)), reverse=True)
|
||||
# Hide empty Untitled sessions from the UI entirely — they are ephemeral
|
||||
|
|
|
|||
|
|
@ -37,12 +37,16 @@ def _isolate_session_dir(tmp_path, monkeypatch):
|
|||
# Also patch the module-level references that Session uses
|
||||
monkeypatch.setattr(models.Session, "__module__", models.__name__)
|
||||
|
||||
# Clear the in-memory SESSIONS cache to avoid bleed
|
||||
# Clear the in-memory SESSIONS and persisted-id caches to avoid bleed.
|
||||
models.SESSIONS.clear()
|
||||
if hasattr(models, "_PERSISTED_SESSION_IDS_CACHE"):
|
||||
models._PERSISTED_SESSION_IDS_CACHE = (None, None, frozenset())
|
||||
|
||||
yield session_dir, index_file
|
||||
|
||||
models.SESSIONS.clear()
|
||||
if hasattr(models, "_PERSISTED_SESSION_IDS_CACHE"):
|
||||
models._PERSISTED_SESSION_IDS_CACHE = (None, None, frozenset())
|
||||
|
||||
|
||||
def _make_session(session_id, title="Untitled", updated_at=None):
|
||||
|
|
@ -515,6 +519,145 @@ def test_newer_continuation_beats_older_fuller_snapshot(monkeypatch):
|
|||
assert rows[0]["message_count"] == 2
|
||||
|
||||
|
||||
def test_all_sessions_uses_sidecar_metadata_for_runtime_rows_when_index_message_count_is_stale(monkeypatch):
|
||||
"""Runtime-shaped rows may refresh from fuller sidecar metadata."""
|
||||
session = Session(
|
||||
session_id="stale_index_sid",
|
||||
title="Reachy Mini Integration",
|
||||
messages=[
|
||||
{"role": "user", "content": "one", "timestamp": 100.0},
|
||||
{"role": "assistant", "content": "two", "timestamp": 101.0},
|
||||
{"role": "user", "content": "three", "timestamp": 102.0},
|
||||
],
|
||||
updated_at=102.0,
|
||||
)
|
||||
session.save(touch_updated_at=False)
|
||||
_write_index_file(
|
||||
models.SESSION_INDEX_FILE,
|
||||
[
|
||||
{
|
||||
"session_id": "stale_index_sid",
|
||||
"title": "Reachy Mini Integration",
|
||||
"message_count": 1,
|
||||
"created_at": 100.0,
|
||||
"updated_at": 100.0,
|
||||
"last_message_at": 100.0,
|
||||
"pinned": False,
|
||||
"archived": False,
|
||||
"active_stream_id": "stream-stale-index",
|
||||
}
|
||||
],
|
||||
)
|
||||
monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
|
||||
|
||||
rows = models.all_sessions()
|
||||
|
||||
assert rows[0]["session_id"] == "stale_index_sid"
|
||||
assert rows[0]["message_count"] == 3
|
||||
assert rows[0]["last_message_at"] == 102.0
|
||||
|
||||
|
||||
def test_all_sessions_sidecar_refresh_stays_metadata_only(monkeypatch):
|
||||
"""Refreshing runtime sidebar rows must not hydrate large sidecar messages."""
|
||||
session = Session(
|
||||
session_id="metadata_refresh_sid",
|
||||
title="Metadata Refresh",
|
||||
messages=[
|
||||
{"role": "user", "content": "one", "timestamp": 100.0},
|
||||
{"role": "assistant", "content": "x" * 200_000, "timestamp": 101.0},
|
||||
{"role": "user", "content": "three", "timestamp": 102.0},
|
||||
],
|
||||
parent_session_id="parent_sid",
|
||||
updated_at=102.0,
|
||||
)
|
||||
session.save(touch_updated_at=False)
|
||||
_write_index_file(
|
||||
models.SESSION_INDEX_FILE,
|
||||
[
|
||||
{
|
||||
"session_id": "metadata_refresh_sid",
|
||||
"title": "Metadata Refresh",
|
||||
"message_count": 1,
|
||||
"created_at": 100.0,
|
||||
"updated_at": 100.0,
|
||||
"last_message_at": 100.0,
|
||||
"pinned": False,
|
||||
"archived": False,
|
||||
"active_stream_id": "stream-metadata-refresh",
|
||||
}
|
||||
],
|
||||
)
|
||||
monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
|
||||
|
||||
with patch.object(Session, "load", side_effect=AssertionError("full sidecar load should not run")):
|
||||
rows = models.all_sessions()
|
||||
|
||||
assert rows[0]["session_id"] == "metadata_refresh_sid"
|
||||
assert rows[0]["message_count"] == 3
|
||||
assert rows[0]["last_message_at"] == 102.0
|
||||
|
||||
|
||||
def test_all_sessions_does_not_refresh_lineage_rows_from_sidecars(monkeypatch):
|
||||
"""Lineage rows are enriched from state.db; do not read every sidecar per poll."""
|
||||
_write_index_file(
|
||||
models.SESSION_INDEX_FILE,
|
||||
[
|
||||
{
|
||||
"session_id": "lineage_sid",
|
||||
"title": "Lineage Row",
|
||||
"message_count": 7,
|
||||
"created_at": 100.0,
|
||||
"updated_at": 101.0,
|
||||
"last_message_at": 101.0,
|
||||
"pinned": False,
|
||||
"archived": False,
|
||||
"parent_session_id": "parent_sid",
|
||||
"_lineage_root_id": "root_sid",
|
||||
"_compression_segment_count": 2,
|
||||
}
|
||||
],
|
||||
)
|
||||
(models.SESSION_DIR / "lineage_sid.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"session_id": "lineage_sid",
|
||||
"title": "Lineage Row",
|
||||
"messages": [{"role": "user", "content": "sidecar"}],
|
||||
"message_count": 99,
|
||||
"updated_at": 200.0,
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
|
||||
|
||||
with patch.object(Session, "load_metadata_only", side_effect=AssertionError("lineage rows must not refresh sidecars")):
|
||||
rows = models.all_sessions()
|
||||
|
||||
assert rows[0]["session_id"] == "lineage_sid"
|
||||
assert rows[0]["message_count"] == 7
|
||||
|
||||
|
||||
def test_load_metadata_only_skips_index_read_when_sidecar_has_message_count(monkeypatch):
|
||||
"""Modern sidecars already carry message_count; avoid an _index.json read per row."""
|
||||
session = Session(
|
||||
session_id="metadata_count_sid",
|
||||
title="Metadata Count",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
)
|
||||
session.save(touch_updated_at=False)
|
||||
|
||||
def _fail_index_lookup(_sid):
|
||||
raise AssertionError("message_count sidecar should not need index lookup")
|
||||
|
||||
monkeypatch.setattr(models, "_lookup_index_message_count", _fail_index_lookup)
|
||||
|
||||
meta = Session.load_metadata_only("metadata_count_sid")
|
||||
|
||||
assert meta is not None
|
||||
assert meta.compact()["message_count"] == 1
|
||||
|
||||
|
||||
def test_session_save_does_not_persist_metadata_message_count_hint():
|
||||
s = Session(
|
||||
session_id="sess_private_hint",
|
||||
|
|
|
|||
Loading…
Reference in a new issue