Session list performance: disk summary cache, display-name cache, and count alignment

api/projects.py

@@ -5,31 +5,23 @@
 from api._flask_types import FlaskReturn, json_response
 from api.error_codes import ErrorCode, error_response
 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__)
 
 
-def _session_row_ok(s: SessionListItemDict, parsed: SessionDict) -> ProjectSessionRowDict:
-    meta = parsed["metadata"]
-    models = meta.get("models_used", [])
-    return {
-        "id": s["id"],
-        "path": s["path"],
-        "size_bytes": s["size_bytes"],
-        "modified": s["modified"],
-        "title": parsed["title"],
-        "models": sorted(models) if isinstance(models, set) else list(models),
-        "tokens": meta["total_input_tokens"] + meta["total_output_tokens"],
-        "tool_calls": meta["total_tool_calls"],
-        "first_timestamp": meta["first_timestamp"],
-        "last_timestamp": meta["last_timestamp"],
-    }
-
-
 def _session_row_error(s: SessionListItemDict) -> ProjectSessionRowDict:
     return {
         "id": s["id"],
@@ -41,30 +33,44 @@ 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:
+                current_app.logger.exception(
+                    "Failed to peek session summary for project %s",
+                    project["name"],
+                )
                 titled_count += 1
         project["session_count"] = titled_count
         if latest_ts:
@@ -82,21 +88,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))
+            result.append(session_row_from_summary(s, row))
         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,39 @@ 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,177 @@
+"""Unit tests for utils.session_summary_cache."""
+
+from __future__ import annotations
+
+import os
+import shutil
+from collections.abc import Iterator
+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) -> Iterator[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_max_cache_rows_invalid_env_falls_back(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SUMMARY_CACHE_MAX_ROWS", "not-a-number")
+    from utils.session_summary_cache import DEFAULT_MAX_ROWS, max_cache_rows
+
+    assert max_cache_rows() == DEFAULT_MAX_ROWS
+
+
+def test_max_cache_rows_valid_env_enforces_minimum(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SUMMARY_CACHE_MAX_ROWS", "0")
+    from utils.session_summary_cache import max_cache_rows
+
+    assert max_cache_rows() == 1
+
+
+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))),
+        )
+
+    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_put_summary_drops_stale_mtime_rows(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
+    put_summary(path, new_mtime, "none", summary_from_session(parsed, is_excluded=False))
+    assert get_summary(path, mtime, "none") is None
+    assert get_summary(path, new_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