David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #2149 into stage-351

Browse source 
perf(sessions): cache CLI session scans (starship-s)

Conflict resolution on api/routes.py:
  (1) Master grew a new helper '_messages_include_tool_metadata()' that
      pr-2149 doesn't have. Kept it (unrelated function — detects whether
      returned messages contain tool metadata, used elsewhere).
  (2) pr-2149 renames the CLI-metadata gate from '_needs_cli_session_metadata'
      to '_session_requires_cli_metadata_lookup' AND broadens it to cover
      legacy-imported sidecars with 'read_only=False' but persisted 'is_cli_session'
      or session_source markers. The new gate is strictly more inclusive than
      the master version — covers (a) is_cli_session, (b) read_only=True,
      (c) session_source in {messaging, external_agent}, AND (d) source_tag,
      raw_source, source, source_label, platform markers. All sessions that
      previously took the slow path still do, plus a few more legacy shapes
      that needed CLI metadata for correct display.
  (3) Removed the obsolete '_needs_cli_session_metadata()' definition from
      master (only consumer migrated to the new name).

29/29 tests pass across test_session_cli_scan_fast_path (new), claude_code
session import, session_index, and session_lineage_full_transcript.
This commit is contained in:
Hermes Agent 2026-05-13 23:54:15 +00:00
commit 2accf6335c
5 changed files with 471 additions and 117 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

298
api/models.py
View file

@ -1,5 +1,6 @@
"""Hermes Web UI -- Session model and in-memory session store."""
import collections
import copy
import datetime
import hashlib
import json
@ -22,6 +23,9 @@ from api.agent_sessions import read_importable_agent_session_rows, read_session_
logger = logging.getLogger(__name__)
CLI_VISIBLE_SESSION_LIMIT = 20
_CLI_SESSIONS_CACHE_TTL_SECONDS = 5.0
_CLI_SESSIONS_CACHE_LOCK = threading.Lock()
_CLI_SESSIONS_CACHE = {}
# ---------------------------------------------------------------------------
# Stale temp-file cleanup
@ -383,6 +387,7 @@ class Session:
self.raw_source = kwargs.get('raw_source')
self.session_source = kwargs.get('session_source')
self.source_label = kwargs.get('source_label')
self.read_only = bool(kwargs.get('read_only', False))
self.enabled_toolsets = enabled_toolsets # List[str] or None — per-session toolset override
self.composer_draft = composer_draft if isinstance(composer_draft, dict) else {}
self._metadata_message_count = None
@ -426,7 +431,7 @@ class Session:
'gateway_routing', 'gateway_routing_history', 'llm_title_generated',
'parent_session_id',
'worktree_path', 'worktree_branch', 'worktree_repo_root', 'worktree_created_at',
'is_cli_session', 'source_tag', 'raw_source', 'session_source', 'source_label',
'is_cli_session', 'source_tag', 'raw_source', 'session_source', 'source_label', 'read_only',
'enabled_toolsets', 'composer_draft',
]
meta = {k: getattr(self, k, None) for k in METADATA_FIELDS}
@ -610,6 +615,7 @@ class Session:
'raw_source': self.raw_source,
'session_source': self.session_source,
'source_label': self.source_label,
'read_only': self.read_only,
'enabled_toolsets': self.enabled_toolsets,
'composer_draft': self.composer_draft if isinstance(self.composer_draft, dict) else {},
'is_streaming': _is_streaming_session(
@ -1016,9 +1022,11 @@ def all_sessions(diag=None):
_diag_stage(diag, "all_sessions.read_index")
index = json.loads(SESSION_INDEX_FILE.read_text(encoding='utf-8'))
_diag_stage(diag, "all_sessions.prune_index")
with LOCK:
in_memory_ids = set(SESSIONS.keys())
index = [
s for s in index
if _index_entry_exists(s.get('session_id'))
if _index_entry_exists(s.get('session_id'), in_memory_ids=in_memory_ids)
]
backfilled = []
for i, s in enumerate(index):
@ -1536,20 +1544,51 @@ def get_claude_code_session_messages(sid, projects_dir: Path | str | None = None
return []
def get_cli_sessions() -> list:
"""Read CLI sessions from the agent's SQLite store and return them as
dicts in a format the WebUI sidebar can render alongside local sessions.
def clear_cli_sessions_cache() -> None:
with _CLI_SESSIONS_CACHE_LOCK:
_CLI_SESSIONS_CACHE.clear()
Returns empty list if the SQLite DB is missing or any error occurs -- the
bridge is purely additive and never crashes the WebUI.
"""
import os
cli_sessions = []
def _copy_cli_sessions(sessions: list) -> list:
return copy.deepcopy(sessions)
def _cli_sessions_cache_ttl_seconds() -> float:
try:
cli_sessions.extend(get_claude_code_sessions())
except Exception:
logger.debug("Claude Code session scan failed", exc_info=True)
return max(0.0, float(_CLI_SESSIONS_CACHE_TTL_SECONDS))
except (TypeError, ValueError):
return 5.0
def _path_cache_key(path) -> str | None:
if path is None:
return None
try:
return str(Path(path).expanduser().resolve(strict=False))
except Exception:
return str(path)
def _path_stat_cache_key(path):
if path is None:
return None
try:
st = Path(path).stat()
return (st.st_mtime_ns, st.st_size)
except OSError:
return None
def _sqlite_file_stat_cache_key(db_path: Path):
"""Return a cheap invalidation key for a SQLite DB and WAL sidecars."""
return (
_path_stat_cache_key(db_path),
_path_stat_cache_key(Path(f"{db_path}-wal")),
_path_stat_cache_key(Path(f"{db_path}-shm")),
)
def _resolve_cli_sessions_context():
# Use the active WebUI profile's HERMES_HOME to find state.db.
# The active profile is determined by what the user has selected in the UI
# (stored in the server's runtime config). This means:
@ -1564,17 +1603,35 @@ def get_cli_sessions() -> list:
except Exception:
hermes_home = Path(os.getenv('HERMES_HOME', str(HOME / '.hermes'))).expanduser().resolve()
db_path = hermes_home / 'state.db'
if not db_path.exists():
return cli_sessions
# Try to resolve the active CLI profile so imported sessions integrate
# with the WebUI profile filter (available since Sprint 22).
try:
from api.profiles import get_active_profile_name
_cli_profile = get_active_profile_name()
except ImportError:
_cli_profile = None # older agent -- fall back to no profile
cli_profile = get_active_profile_name()
except Exception:
cli_profile = None
db_path = hermes_home / 'state.db'
projects_dir = _default_claude_code_projects_dir()
cache_key = (
str(hermes_home),
str(cli_profile or ''),
str(db_path),
_sqlite_file_stat_cache_key(db_path),
_path_cache_key(projects_dir),
_path_stat_cache_key(projects_dir),
_path_stat_cache_key(SESSION_INDEX_FILE),
)
return hermes_home, db_path, cli_profile, cache_key
def _load_cli_sessions_uncached(hermes_home: Path, db_path: Path, _cli_profile) -> list:
cli_sessions = []
try:
cli_sessions.extend(get_claude_code_sessions())
except Exception:
logger.debug("Claude Code session scan failed", exc_info=True)
if not db_path.exists():
return cli_sessions
# Memoize the cron project ID for this scan so we don't pay a lock-acquire +
# disk-read of projects.json per cron session in the loop below.
@ -1585,96 +1642,129 @@ def get_cli_sessions() -> list:
_cron_pid_cache[0] = ensure_cron_project()
return _cron_pid_cache[0]
try:
for row in read_importable_agent_session_rows(
db_path,
limit=CLI_VISIBLE_SESSION_LIMIT,
log=logger,
exclude_sources=None,
):
sid = row['id']
raw_ts = row['last_activity'] or row['started_at']
# Prefer the CLI session's own profile from the DB; fall back to
# the active CLI profile so sidebar filtering works either way.
profile = _cli_profile # CLI DB has no profile column; use active profile
for row in read_importable_agent_session_rows(
db_path,
limit=CLI_VISIBLE_SESSION_LIMIT,
log=logger,
exclude_sources=None,
):
sid = row['id']
raw_ts = row['last_activity'] or row['started_at']
# Prefer the CLI session's own profile from the DB; fall back to
# the active CLI profile so sidebar filtering works either way.
profile = _cli_profile # CLI DB has no profile column; use active profile
_source = row['source'] or 'cli'
_title = row['title']
if not _title and _source == 'cron' and sid.startswith('cron_'):
# Extract job_id from session ID (cron_{job_id}_{timestamp})
# and look up the human-friendly job name from jobs.json
parts = sid.split('_')
if len(parts) >= 3:
_job_id = parts[1]
try:
_jobs_path = hermes_home / 'cron' / 'jobs.json'
if _jobs_path.exists():
import json as _json
_jobs_data = _json.loads(_jobs_path.read_text())
for _j in _jobs_data.get('jobs', []):
if _j.get('id') == _job_id:
_title = _j.get('name') or _title
break
except Exception:
pass # degrade gracefully
# If a WebUI JSON file exists for this session (e.g. previously
# imported or renamed in the sidebar), prefer its title over the
# state.db title. This fixes rename-not-persisting for CLI sessions
# after compression chain extension (#1486).
_source = row['source'] or 'cli'
_title = row['title']
if not _title and _source == 'cron' and sid.startswith('cron_'):
# Extract job_id from session ID (cron_{job_id}_{timestamp})
# and look up the human-friendly job name from jobs.json
parts = sid.split('_')
if len(parts) >= 3:
_job_id = parts[1]
try:
_jobs_path = hermes_home / 'cron' / 'jobs.json'
if _jobs_path.exists():
import json as _json
_jobs_data = _json.loads(_jobs_path.read_text())
for _j in _jobs_data.get('jobs', []):
if _j.get('id') == _job_id:
_title = _j.get('name') or _title
break
except Exception:
pass # degrade gracefully
# If a WebUI JSON file exists for this session (e.g. previously
# imported or renamed in the sidebar), prefer its title over the
# state.db title. This fixes rename-not-persisting for CLI sessions
# after compression chain extension (#1486).
try:
_webui_meta = Session.load_metadata_only(sid)
if _webui_meta and getattr(_webui_meta, 'title', None):
_title = _webui_meta.title
except Exception:
pass
_display_title = _title or f'{_source.title()} Session'
cli_sessions.append({
'session_id': sid,
'title': _display_title,
'workspace': str(get_last_workspace()),
'model': row['model'] or None,
'message_count': row['message_count'] or row['actual_message_count'] or 0,
'created_at': row['started_at'],
'updated_at': raw_ts,
'pinned': False,
'archived': False,
'project_id': _cron_pid() if is_cron_session(sid, _source) else None,
'profile': profile,
'source_tag': _source,
'raw_source': row.get('raw_source'),
'user_id': row.get('user_id'),
'chat_id': row.get('chat_id') or row.get('origin_chat_id'),
'chat_type': row.get('chat_type'),
'thread_id': row.get('thread_id'),
'session_key': row.get('session_key'),
'platform': row.get('platform'),
'session_source': row.get('session_source'),
'source_label': row.get('source_label'),
'parent_session_id': row.get('parent_session_id'),
'parent_title': row.get('parent_title'),
'parent_source': row.get('parent_source'),
'relationship_type': row.get('relationship_type'),
'_parent_lineage_root_id': row.get('_parent_lineage_root_id'),
'end_reason': row.get('end_reason'),
'actual_message_count': row.get('actual_message_count'),
'user_message_count': row.get('actual_user_message_count'),
'_lineage_root_id': row.get('_lineage_root_id'),
'_lineage_tip_id': row.get('_lineage_tip_id'),
'_compression_segment_count': row.get('_compression_segment_count'),
'is_cli_session': True,
})
return cli_sessions
def get_cli_sessions() -> list:
"""Read CLI sessions from the agent's SQLite store and return them as
dicts in a format the WebUI sidebar can render alongside local sessions.
Returns empty list if the SQLite DB is missing or any error occurs -- the
bridge is purely additive and never crashes the WebUI.
"""
hermes_home, db_path, cli_profile, cache_key = _resolve_cli_sessions_context()
ttl = _cli_sessions_cache_ttl_seconds()
now = time.monotonic()
if ttl > 0:
with _CLI_SESSIONS_CACHE_LOCK:
cached = _CLI_SESSIONS_CACHE.get(cache_key)
if cached:
expires_at, cached_sessions = cached
if expires_at > now:
return _copy_cli_sessions(cached_sessions)
_CLI_SESSIONS_CACHE.pop(cache_key, None)
try:
_webui_meta = Session.load_metadata_only(sid)
if _webui_meta and getattr(_webui_meta, 'title', None):
_title = _webui_meta.title
except Exception:
pass
_display_title = _title or f'{_source.title()} Session'
cli_sessions.append({
'session_id': sid,
'title': _display_title,
'workspace': str(get_last_workspace()),
'model': row['model'] or None,
'message_count': row['message_count'] or row['actual_message_count'] or 0,
'created_at': row['started_at'],
'updated_at': raw_ts,
'pinned': False,
'archived': False,
'project_id': _cron_pid() if is_cron_session(sid, _source) else None,
'profile': profile,
'source_tag': _source,
'raw_source': row.get('raw_source'),
'user_id': row.get('user_id'),
'chat_id': row.get('chat_id') or row.get('origin_chat_id'),
'chat_type': row.get('chat_type'),
'thread_id': row.get('thread_id'),
'session_key': row.get('session_key'),
'platform': row.get('platform'),
'session_source': row.get('session_source'),
'source_label': row.get('source_label'),
'parent_session_id': row.get('parent_session_id'),
'parent_title': row.get('parent_title'),
'parent_source': row.get('parent_source'),
'relationship_type': row.get('relationship_type'),
'_parent_lineage_root_id': row.get('_parent_lineage_root_id'),
'end_reason': row.get('end_reason'),
'actual_message_count': row.get('actual_message_count'),
'user_message_count': row.get('actual_user_message_count'),
'_lineage_root_id': row.get('_lineage_root_id'),
'_lineage_tip_id': row.get('_lineage_tip_id'),
'_compression_segment_count': row.get('_compression_segment_count'),
'is_cli_session': True,
})
sessions = _load_cli_sessions_uncached(hermes_home, db_path, cli_profile)
except Exception as _cli_err:
logger.warning(
"get_cli_sessions() failed — check state.db schema or path (%s): %s",
db_path, _cli_err,
)
return []
_CLI_SESSIONS_CACHE[cache_key] = (
time.monotonic() + ttl,
_copy_cli_sessions(sessions),
)
return _copy_cli_sessions(sessions)
try:
return _load_cli_sessions_uncached(hermes_home, db_path, cli_profile)
except Exception as _cli_err:
# DB schema changed, locked, or corrupted -- log warning so admins can diagnose.
# Still degrade gracefully (don't crash the WebUI).
import logging as _logging
_logging.getLogger(__name__).warning(
logger.warning(
"get_cli_sessions() failed — check state.db schema or path (%s): %s",
db_path, _cli_err,
)
return []
return cli_sessions
def _json_loads_if_string(value):
if not isinstance(value, str):

51
api/routes.py
View file

@ -1525,18 +1525,6 @@ def _lookup_cli_session_metadata(session_id: str) -> dict:
return {}
def _needs_cli_session_metadata(session) -> bool:
"""Return true when /api/session should pay for Agent/CLI metadata lookup."""
if not session:
return False
is_cli = (
bool(session.get("is_cli_session"))
if isinstance(session, dict)
else bool(getattr(session, "is_cli_session", False))
)
return is_cli or _is_messaging_session_record(session)
def _messaging_session_identity(session: dict, raw_source: str) -> str:
metadata = _lookup_gateway_session_identity(session.get("session_id"))
session_key = _safe_first(
@ -1684,6 +1672,43 @@ def _messages_include_tool_metadata(messages) -> bool:
return False
def _session_requires_cli_metadata_lookup(session) -> bool:
"""Return True when a sidecar/session row still needs CLI metadata.
Legacy imported sidecars may predate the ``read_only`` field and therefore
load with ``read_only=False``. They still persist ``is_cli_session`` and/or
source metadata from import time, so those markers intentionally keep them
on the CLI lookup path while ordinary WebUI-native sessions take the fast
path.
Supersedes the simpler is-cli-or-messaging gate from PR #1822 — the new
gate is strictly more inclusive (also covers ``read_only=True`` sidecars,
``session_source`` markers, and source_tag/raw_source/platform metadata)
so all sessions that previously took the slow path still do, plus a few
more legacy shapes.
"""
if not session:
return False
def _field(name):
return session.get(name) if isinstance(session, dict) else getattr(session, name, None)
if _is_messaging_session_record(session):
return True
if bool(_field("is_cli_session")) or bool(_field("read_only")):
return True
session_source = _normalize_messaging_source(_safe_first(_field("session_source")))
if session_source in {"messaging", "external_agent", "external-agent"}:
return True
return bool(_safe_first(
_field("source_tag"),
_field("raw_source"),
_field("source"),
_field("source_label"),
_field("platform"),
))
def _is_messaging_session_id(sid: str) -> bool:
"""Detect messaging-backed sessions from WebUI metadata or Agent rows."""
try:
@ -3252,7 +3277,7 @@ def handle_get(handler, parsed) -> bool:
_t1 = _time.monotonic()
s = get_session(sid, metadata_only=(not load_messages))
_clear_stale_stream_state(s)
cli_meta = _lookup_cli_session_metadata(sid) if _needs_cli_session_metadata(s) else {}
cli_meta = _lookup_cli_session_metadata(sid) if _session_requires_cli_metadata_lookup(s) else {}
is_messaging_session = _is_messaging_session_record(s) or _is_messaging_session_record(cli_meta)
cli_messages = []
if is_messaging_session:

83
tests/test_claude_code_session_import.py
View file

@ -86,6 +86,89 @@ def test_claude_code_scan_skips_symlinks_and_oversized_files(tmp_path):
assert models.get_claude_code_sessions(projects_dir=root_link) == []
def test_get_cli_sessions_reuses_short_ttl_cache(monkeypatch, tmp_path):
import api.models as models
import api.profiles as profiles
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: str(hermes_home))
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default")
monkeypatch.setattr(models, "_CLI_SESSIONS_CACHE_TTL_SECONDS", 60.0, raising=False)
models.clear_cli_sessions_cache()
calls = 0
def fake_claude_code_sessions():
nonlocal calls
calls += 1
return [
{
"session_id": "claude_code_cached",
"title": "Cached Claude Code",
"updated_at": calls,
"message_count": 1,
"source_tag": "claude_code",
"is_cli_session": True,
}
]
monkeypatch.setattr(models, "get_claude_code_sessions", fake_claude_code_sessions)
first = models.get_cli_sessions()
first[0]["title"] = "mutated by caller"
second = models.get_cli_sessions()
assert calls == 1
assert second[0]["title"] == "Cached Claude Code"
assert second[0]["updated_at"] == 1
def test_get_cli_sessions_cache_invalidates_when_sqlite_wal_changes(monkeypatch, tmp_path):
import api.models as models
import api.profiles as profiles
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
db_path = hermes_home / "state.db"
db_path.write_text("initial", encoding="utf-8")
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: str(hermes_home))
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default")
monkeypatch.setattr(models, "_CLI_SESSIONS_CACHE_TTL_SECONDS", 60.0, raising=False)
monkeypatch.setattr(models, "get_claude_code_sessions", lambda: [])
models.clear_cli_sessions_cache()
calls = 0
def fake_rows(_db_path, **_kwargs):
nonlocal calls
calls += 1
return [
{
"id": "cli_cached_state_db",
"title": "State DB Session",
"model": "test-model",
"source": "cli",
"raw_source": "cli",
"message_count": calls,
"actual_message_count": calls,
"actual_user_message_count": 1,
"last_activity": float(calls),
"started_at": 1.0,
}
]
monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_rows)
first = models.get_cli_sessions()
Path(f"{db_path}-wal").write_text("new wal contents", encoding="utf-8")
second = models.get_cli_sessions()
assert calls == 2
assert first[0]["message_count"] == 1
assert second[0]["message_count"] == 2
def test_session_import_cli_returns_read_only_claude_code_payload(monkeypatch, tmp_path):
import api.routes as routes

109
tests/test_session_cli_scan_fast_path.py Normal file
View file

@ -0,0 +1,109 @@
from urllib.parse import urlparse
def test_webui_session_metadata_load_skips_cli_metadata_scan(monkeypatch):
"""Opening a normal WebUI session should not scan imported CLI sessions."""
import api.routes as routes
from api.models import Session
session = Session(
session_id="webui_normal",
title="Normal WebUI chat",
messages=[{"role": "user", "content": "hello"}],
)
monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: session)
monkeypatch.setattr(routes, "_clear_stale_stream_state", lambda _session: None)
monkeypatch.setattr(routes, "redact_session_data", lambda payload: payload)
monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload)
monkeypatch.setattr(
routes,
"_lookup_cli_session_metadata",
lambda _sid: (_ for _ in ()).throw(AssertionError("normal WebUI loads should not scan CLI sessions")),
)
response = routes.handle_get(
object(),
urlparse("/api/session?session_id=webui_normal&messages=0&resolve_model=0"),
)
assert response["session"]["session_id"] == "webui_normal"
assert response["session"]["messages"] == []
def test_read_only_session_metadata_load_preserves_cli_metadata_lookup(monkeypatch):
"""Read-only imported sidecars still need CLI metadata for source identity."""
import api.routes as routes
from api.models import Session
session = Session(
session_id="readonly_sidecar",
title="Imported chat",
messages=[{"role": "user", "content": "hello"}],
read_only=True,
)
looked_up = []
monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: session)
monkeypatch.setattr(routes, "_clear_stale_stream_state", lambda _session: None)
monkeypatch.setattr(routes, "get_cli_session_messages", lambda _sid: [])
monkeypatch.setattr(routes, "redact_session_data", lambda payload: payload)
monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload)
def fake_lookup(sid):
looked_up.append(sid)
return {
"session_id": sid,
"read_only": True,
"source_label": "External Agent",
}
monkeypatch.setattr(routes, "_lookup_cli_session_metadata", fake_lookup)
response = routes.handle_get(
object(),
urlparse("/api/session?session_id=readonly_sidecar&messages=0&resolve_model=0"),
)
assert looked_up == ["readonly_sidecar"]
assert response["session"]["read_only"] is True
def test_messaging_session_metadata_load_preserves_cli_metadata_lookup(monkeypatch):
"""Messaging/imported sidecars still need CLI metadata for source identity."""
import api.routes as routes
from api.models import Session
session = Session(
session_id="messaging_sidecar",
title="Telegram chat",
messages=[{"role": "user", "content": "hello"}],
session_source="messaging",
raw_source="telegram",
)
looked_up = []
monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: session)
monkeypatch.setattr(routes, "_clear_stale_stream_state", lambda _session: None)
monkeypatch.setattr(routes, "get_cli_session_messages", lambda _sid: [])
monkeypatch.setattr(routes, "redact_session_data", lambda payload: payload)
monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload)
def fake_lookup(sid):
looked_up.append(sid)
return {
"session_id": sid,
"session_source": "messaging",
"raw_source": "telegram",
"source_label": "Telegram",
}
monkeypatch.setattr(routes, "_lookup_cli_session_metadata", fake_lookup)
response = routes.handle_get(
object(),
urlparse("/api/session?session_id=messaging_sidecar&messages=0&resolve_model=0"),
)
assert looked_up == ["messaging_sidecar"]
assert response["session"]["source_label"] == "Telegram"

47
tests/test_session_index.py
View file

@ -123,6 +123,53 @@ def test_all_sessions_backfills_last_message_at_for_legacy_index_rows():
assert persisted[0].get("last_message_at") == 100.0
def test_all_sessions_prune_reuses_in_memory_id_snapshot(monkeypatch):
"""Index pruning should not reacquire the session lock for every row."""
index_file = models.SESSION_INDEX_FILE
entries = [
{
"session_id": "sess_a",
"title": "Alpha",
"updated_at": 200.0,
"last_message_at": 200.0,
"workspace": "/tmp",
"model": "test",
"message_count": 1,
"created_at": 100.0,
"pinned": False,
"archived": False,
},
{
"session_id": "sess_b",
"title": "Bravo",
"updated_at": 150.0,
"last_message_at": 150.0,
"workspace": "/tmp",
"model": "test",
"message_count": 1,
"created_at": 90.0,
"pinned": False,
"archived": False,
},
]
_write_index_file(index_file, entries)
seen = []
def _assert_snapshot_used(session_id, in_memory_ids=None):
assert in_memory_ids is not None, "all_sessions should snapshot SESSIONS once before pruning"
seen.append(session_id)
return True
monkeypatch.setattr(models, "_index_entry_exists", _assert_snapshot_used)
monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
rows = models.all_sessions()
assert [row["session_id"] for row in rows] == ["sess_a", "sess_b"]
assert seen == ["sess_a", "sess_b"]
# ── 6. test_incremental_patch_correctness ─────────────────────────────────
def test_incremental_patch_correctness():