From 017179b1896d421bd543cd3d46152974ae06b143 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Fri, 29 May 2026 23:34:07 +0200 Subject: [PATCH] fix: speed up session index metadata refresh --- api/models.py | 135 ++++++++++++++++++++++++++++----- tests/test_session_index.py | 145 +++++++++++++++++++++++++++++++++++- 2 files changed, 262 insertions(+), 18 deletions(-) diff --git a/api/models.py b/api/models.py index 0be70f1f..569f1dfc 100644 --- a/api/models.py +++ b/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 diff --git a/tests/test_session_index.py b/tests/test_session_index.py index bb71ed3f..f4889e75 100644 --- a/tests/test_session_index.py +++ b/tests/test_session_index.py @@ -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",