Add disk-backed session summary cache and display-name cache

Session list and landing page re-parsed or re-read every .jsonl on each cold start and on every GET /api/projects call. Add SQLite summary cache (keyed by path, mtime, rules fingerprint) so complete rows survive restart; route get_projects and get_project_sessions through it. Cache project display names by directory max mtime. Align project card session_count with session list when no exclusion rules are active.

Author: clean6378-max-it

api/projects.py

@@ -7,8 +7,18 @@
 from models.project import ProjectSessionRowDict, SessionListItemDict
 from models.session import SessionDict
 from utils.exclusion_rules import is_session_excluded
+from utils.jsonl_parser import quick_session_info
 from utils.session_cache import get_cached_session
 from utils.session_path import get_claude_projects_dir, list_projects, list_sessions, safe_join
+from utils.session_summary_cache import (
+    SummaryCacheRowDict,
+    get_summary,
+    put_summary,
+    rules_fingerprint,
+    session_row_from_summary,
+    summary_from_peek,
+    summary_from_session,
+)
 
 projects_bp = Blueprint("projects", __name__)
 
@@ -41,27 +51,37 @@ def _session_row_error(s: SessionListItemDict) -> ProjectSessionRowDict:
     }
 
 
+def _peek_or_cache_summary(path: str, mtime: float, rules_fp: str) -> SummaryCacheRowDict:
+    """Return a cached summary row or peek the file and store a partial row."""
+    cached = get_summary(path, mtime, rules_fp)
+    if cached is not None:
+        return cached
+    info = quick_session_info(path)
+    row = summary_from_peek(info)
+    put_summary(path, mtime, rules_fp, row)
+    return row
+
+
 @projects_bp.route("/api/projects")
 def get_projects() -> FlaskReturn:
     base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
     projects = list_projects(base)
-
-    # Enrich each project with accurate titled-session count and latest timestamp
-    # so the landing page matches what the workspace page shows.
-    # Uses quick_session_info() which peeks at files without full parsing.
-    from utils.jsonl_parser import quick_session_info
+    rules = current_app.config.get("EXCLUSION_RULES") or []
+    rules_fp = rules_fingerprint(rules)
 
     for project in projects:
         sessions = list_sessions(project["path"])
         titled_count = 0
         latest_ts = None
         for s in sessions:
             try:
-                info = quick_session_info(s["path"])
-                if info["title"] == "Untitled Session":
+                row = _peek_or_cache_summary(s["path"], s["modified"], rules_fp)
+                if row["is_untitled"]:
+                    continue
+                if row["is_complete"] and row["is_excluded"]:
                     continue
                 titled_count += 1
-                ts = info.get("last_timestamp") or info.get("first_timestamp")
+                ts = row.get("last_timestamp") or row.get("first_timestamp")
                 if ts and (latest_ts is None or ts > latest_ts):
                     latest_ts = ts
             except Exception:
@@ -82,21 +102,25 @@ def get_project_sessions(project_name: str) -> FlaskReturn:
         return error_response(ErrorCode.INVALID_PATH, "Invalid path", 400)
     sessions = list_sessions(project_dir)
     rules = current_app.config.get("EXCLUSION_RULES") or []
+    rules_fp = rules_fingerprint(rules)
     result: list[ProjectSessionRowDict] = []
     for s in sessions:
         try:
-            parsed = get_cached_session(s["path"])
-            # Skip untitled sessions (no real conversation)
-            if parsed["title"] == "Untitled Session":
+            cached = get_summary(s["path"], s["modified"], rules_fp)
+            if cached is not None and cached["is_complete"]:
+                if cached["is_untitled"] or cached["is_excluded"]:
+                    continue
+                result.append(session_row_from_summary(s, cached))
                 continue
-            if is_session_excluded(rules, parsed, project_name):
+
+            parsed = get_cached_session(s["path"])
+            excluded = is_session_excluded(rules, parsed, project_name)
+            row = summary_from_session(parsed, is_excluded=excluded)
+            put_summary(s["path"], s["modified"], rules_fp, row)
+            if row["is_untitled"] or excluded:
                 continue
             result.append(_session_row_ok(s, parsed))
         except Exception:
-            # Full detail (class, message, traceback) to the server log via
-            # logger.exception. The per-session card carries only `error: True`
-            # — the class-name+message string was a leak (issue #25). The
-            # operator looks at the server log for triage.
             current_app.logger.exception("Failed to parse session %s", s["id"])
             result.append(_session_row_error(s))
     return json_response(result)

tests/test_api_integration.py

@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+import pytest
+
 from app import CSP_POLICY
 from tests.conftest import assert_error_response as _assert_error_shape
 
@@ -124,3 +126,42 @@ def test_search_valid_limit(client):
     results = resp.get_json()
     assert isinstance(results, list)
     assert len(results) <= 5
+
+
+# --- session summary cache (disk) ---
+
+
+@pytest.fixture
+def summary_cache_db(tmp_path, monkeypatch):
+    from utils.session_summary_cache import clear_cache, reset_connection_for_tests
+
+    db = tmp_path / "session_summary_cache.sqlite"
+    reset_connection_for_tests(db)
+    yield db
+    clear_cache()
+
+
+def test_project_session_count_matches_list(client, summary_cache_db):
+    projects = client.get("/api/projects").get_json()
+    project = next(p for p in projects if p["name"] == "test-project")
+    sessions = client.get("/api/projects/test-project/sessions").get_json()
+    assert project["session_count"] == len(sessions)
+
+
+def test_project_sessions_uses_disk_cache_on_second_request(
+    client, summary_cache_db, monkeypatch
+):
+    client.get("/api/projects/test-project/sessions")
+    calls = 0
+
+    def counting_get_cached(path: str):
+        nonlocal calls
+        calls += 1
+        from utils.session_cache import get_cached_session as real_get
+
+        return real_get(path)
+
+    monkeypatch.setattr("api.projects.get_cached_session", counting_get_cached)
+    client.get("/api/projects/test-project/sessions")
+    assert calls == 0
+

tests/test_session_path.py

@@ -39,3 +39,30 @@ def test_get_claude_projects_dir_on_windows_runner(
     got = session_path.get_claude_projects_dir()
     expected = os.path.join(str(profile), ".claude", "projects")
     assert got == expected
+
+
+def test_display_name_cache_avoids_repeat_file_reads(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    session_path.clear_display_name_cache()
+    project_dir = tmp_path / "proj-hash"
+    project_dir.mkdir()
+    jsonl = project_dir / "session.jsonl"
+    jsonl.write_text(
+        '{"type":"user","cwd":"/home/user/MyProject","timestamp":"2026-01-01T00:00:00Z"}\n',
+        encoding="utf-8",
+    )
+    calls = 0
+    real_get = session_path._get_display_name
+
+    def counting_get_display_name(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return real_get(*args, **kwargs)
+
+    monkeypatch.setattr(session_path, "_get_display_name", counting_get_display_name)
+    session_path.list_projects(str(tmp_path))
+    first_calls = calls
+    session_path.list_projects(str(tmp_path))
+    assert first_calls > 0
+    assert calls == first_calls

tests/test_session_summary_cache.py

@@ -0,0 +1,146 @@
+"""Unit tests for utils.session_summary_cache."""
+
+from __future__ import annotations
+
+import os
+import shutil
+from pathlib import Path
+
+import pytest
+
+from utils.jsonl_parser import parse_session, quick_session_info
+from utils.session_summary_cache import (
+    clear_cache,
+    get_summary,
+    put_summary,
+    reset_connection_for_tests,
+    rules_fingerprint,
+    summary_from_peek,
+    summary_from_session,
+)
+
+FIXTURES = Path(__file__).resolve().parent / "fixtures"
+SAMPLE_SESSION = FIXTURES / "session_with_tools.jsonl"
+
+
+@pytest.fixture
+def sample_session(tmp_path: Path) -> Path:
+    dest = tmp_path / "session.jsonl"
+    shutil.copy(SAMPLE_SESSION, dest)
+    return dest
+
+
+@pytest.fixture
+def cache_db(tmp_path: Path) -> Path:
+    db = tmp_path / "summary.sqlite"
+    reset_connection_for_tests(db)
+    yield db
+    clear_cache()
+
+
+def test_rules_fingerprint_empty() -> None:
+    assert rules_fingerprint([]) == "none"
+
+
+def test_rules_fingerprint_stable() -> None:
+    rules = [[("word", "secret")]]
+    assert rules_fingerprint(rules) == rules_fingerprint(rules)
+
+
+def test_cache_miss_returns_none(sample_session: Path, cache_db: Path) -> None:
+    path = str(sample_session)
+    mtime = sample_session.stat().st_mtime
+    assert get_summary(path, mtime, "none") is None
+
+
+def test_cache_hit_round_trip(sample_session: Path, cache_db: Path) -> None:
+    path = str(sample_session)
+    mtime = sample_session.stat().st_mtime
+    parsed = parse_session(path)
+    row = summary_from_session(parsed, is_excluded=False)
+    put_summary(path, mtime, "none", row)
+    hit = get_summary(path, mtime, "none")
+    assert hit is not None
+    assert hit["title"] == row["title"]
+    assert hit["tokens"] == row["tokens"]
+    assert hit["is_complete"] is True
+
+
+def test_cache_invalidates_on_mtime_change(sample_session: Path, cache_db: Path) -> None:
+    path = str(sample_session)
+    mtime = sample_session.stat().st_mtime
+    parsed = parse_session(path)
+    put_summary(path, mtime, "none", summary_from_session(parsed, is_excluded=False))
+    stat = sample_session.stat()
+    os.utime(sample_session, (stat.st_mtime + 10, stat.st_mtime + 10))
+    new_mtime = sample_session.stat().st_mtime
+    assert new_mtime != mtime
+    assert get_summary(path, new_mtime, "none") is None
+
+
+def test_exclusion_key_separation(sample_session: Path, cache_db: Path) -> None:
+    path = str(sample_session)
+    mtime = sample_session.stat().st_mtime
+    parsed = parse_session(path)
+    put_summary(path, mtime, "none", summary_from_session(parsed, is_excluded=False))
+    put_summary(path, mtime, "rules_a", summary_from_session(parsed, is_excluded=True))
+    hit_none = get_summary(path, mtime, "none")
+    hit_rules = get_summary(path, mtime, "rules_a")
+    assert hit_none is not None and hit_none["is_excluded"] is False
+    assert hit_rules is not None and hit_rules["is_excluded"] is True
+
+
+def test_peek_partial_row(sample_session: Path, cache_db: Path) -> None:
+    path = str(sample_session)
+    mtime = sample_session.stat().st_mtime
+    row = summary_from_peek(quick_session_info(path))
+    put_summary(path, mtime, "none", row)
+    hit = get_summary(path, mtime, "none")
+    assert hit is not None
+    assert hit["is_complete"] is False
+    assert hit["tokens"] == 0
+
+
+def test_lru_eviction(
+    sample_session: Path, cache_db: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr("utils.session_summary_cache.DEFAULT_MAX_ROWS", 2)
+    content = sample_session.read_text(encoding="utf-8")
+    paths = []
+    for name in ("a.jsonl", "b.jsonl", "c.jsonl"):
+        p = sample_session.parent / name
+        p.write_text(content, encoding="utf-8")
+        paths.append(p)
+
+    for p in paths[:2]:
+        mtime = p.stat().st_mtime
+        put_summary(
+            str(p),
+            mtime,
+            "none",
+            summary_from_peek(quick_session_info(str(p))),
+        )
+        get_summary(str(p), mtime, "none")
+
+    third = paths[2]
+    third_mtime = third.stat().st_mtime
+    put_summary(
+        str(third),
+        third_mtime,
+        "none",
+        summary_from_peek(quick_session_info(str(third))),
+    )
+
+    first = paths[0]
+    assert get_summary(str(first), first.stat().st_mtime, "none") is None
+    assert get_summary(str(third), third_mtime, "none") is not None
+
+
+def test_clear_cache(sample_session: Path, cache_db: Path) -> None:
+    path = str(sample_session)
+    mtime = sample_session.stat().st_mtime
+    parsed = parse_session(path)
+    put_summary(path, mtime, "none", summary_from_session(parsed, is_excluded=False))
+    clear_cache()
+    reset_connection_for_tests(cache_db)
+    assert get_summary(path, mtime, "none") is None

utils/session_path.py

@@ -5,11 +5,15 @@
 import logging
 import os
 import platform
+import threading
 
 from models.project import ProjectDict, SessionListItemDict
 
 _logger = logging.getLogger(__name__)
 
+_display_name_cache: dict[str, tuple[float, str]] = {}
+_display_name_lock = threading.Lock()
+
 
 def safe_join(base: str, *parts: str) -> str:
     """Join path components and verify the result stays under base.
@@ -31,6 +35,33 @@ def get_claude_projects_dir() -> str:
     return os.path.join(home, ".claude", "projects")
 
 
+def clear_display_name_cache() -> None:
+    """Clear the in-memory display-name cache (for tests)."""
+    with _display_name_lock:
+        _display_name_cache.clear()
+
+
+def _project_jsonl_max_mtime(project_dir: str, jsonl_files: list[str]) -> float:
+    return max(os.path.getmtime(os.path.join(project_dir, jf)) for jf in jsonl_files)
+
+
+def _resolve_display_name(project_dir: str, jsonl_files: list[str], fallback: str) -> str:
+    max_mtime = _project_jsonl_max_mtime(project_dir, jsonl_files)
+    with _display_name_lock:
+        hit = _display_name_cache.get(project_dir)
+        if hit is not None and hit[0] == max_mtime:
+            return hit[1]
+    display_name = fallback
+    for jf in jsonl_files:
+        candidate = _get_display_name(os.path.join(project_dir, jf), None)
+        if candidate is not None:
+            display_name = candidate
+            break
+    with _display_name_lock:
+        _display_name_cache[project_dir] = (max_mtime, display_name)
+    return display_name
+
+
 def list_projects(base_dir: str | None = None) -> list[ProjectDict]:
     """Scan the projects dir and return info for each one that has .jsonl files."""
     base = base_dir or get_claude_projects_dir()
@@ -52,13 +83,7 @@ def list_projects(base_dir: str | None = None) -> list[ProjectDict]:
             from datetime import datetime, timezone
 
             last_modified = datetime.fromtimestamp(latest_mtime, tz=timezone.utc).isoformat()
-            # Read cwd from sessions to get the real project path
-            display_name = name
-            for jf in jsonl_files:
-                candidate = _get_display_name(os.path.join(project_dir, jf), None)
-                if candidate is not None:
-                    display_name = candidate
-                    break
+            display_name = _resolve_display_name(project_dir, jsonl_files, name)
             projects.append(
                 {
                     "name": name,