fix: share workspace_storage_fingerprint and guard cache writes

Author: clean6378-max-it

services/export_engine.py

@@ -12,7 +12,7 @@
 
 from models import Bubble
 from models.export import CollectedExportEntry
-from services.summary_cache import fingerprint_workspace_storage, nocache_enabled
+from services.summary_cache import nocache_enabled
 from services.workspace_context import (
     WorkspaceContext,
     enrich_workspace_context_from_global_db,
@@ -22,7 +22,6 @@
 from services.workspace_db import (
     COMPOSER_ROWS_WITH_HEADERS_SQL,
     collect_workspace_entries,
-    global_storage_db_path,
     load_code_block_diff_map,
     open_global_db,
     safe_fetchall,
@@ -84,7 +83,6 @@ class WorkspaceOrchestration:
 
     workspace_path: str
     workspace_entries: list[dict[str, Any]]
-    fingerprint: dict[str, Any]
     ctx: WorkspaceContext
     workspace_id_to_display_name: dict[str, str]
     workspace_id_to_slug: dict[str, str]
@@ -135,21 +133,12 @@ def prepare_workspace_orchestration(
     nocache: bool = False,
     workspace_entries: list[dict[str, Any]] | None = None,
 ) -> WorkspaceOrchestration:
-    """Scan workspace storage and resolve maps (with summary-cache fingerprint)."""
+    """Scan workspace storage and resolve maps for listing and export."""
     entries = (
         workspace_entries
         if workspace_entries is not None
         else collect_workspace_entries(workspace_path)
     )
-    gdb = global_storage_db_path(workspace_path)
-    cli_path = get_cli_chats_path()
-    fingerprint = fingerprint_workspace_storage(
-        workspace_path,
-        entries,
-        global_db_path=gdb if os.path.isfile(gdb) else None,
-        rules=rules,
-        cli_chats_path=cli_path if os.path.isdir(cli_path) else None,
-    )
     ctx = resolve_workspace_context_cached(
         workspace_path,
         rules,
@@ -160,7 +149,6 @@ def prepare_workspace_orchestration(
     return WorkspaceOrchestration(
         workspace_path=workspace_path,
         workspace_entries=entries,
-        fingerprint=fingerprint,
         ctx=ctx,
         workspace_id_to_display_name=display_name,
         workspace_id_to_slug=slug_map,

services/search_index.py

@@ -29,8 +29,8 @@
 )
 from services.summary_cache import (
     CACHE_DIR,
-    fingerprint_workspace_storage,
     nocache_enabled,
+    workspace_storage_fingerprint,
 )
 from services.workspace_db import (
     COMPOSER_ROWS_WITH_HEADERS_SQL,
@@ -40,7 +40,6 @@
 )
 from utils.path_helpers import to_epoch_ms
 from utils.exclusion_rules import RuleTokens
-from utils.workspace_path import get_cli_chats_path
 
 __all__ = [
     "SEARCH_INDEX_FILE",
@@ -130,15 +129,7 @@ def index_search_enabled() -> bool:
 
 def _storage_fingerprint(workspace_path: str, rules: list[RuleTokens]) -> dict[str, Any]:
     entries = collect_workspace_entries(workspace_path)
-    gdb = global_storage_db_path(workspace_path)
-    cli_path = get_cli_chats_path()
-    return fingerprint_workspace_storage(
-        workspace_path,
-        entries,
-        global_db_path=gdb if os.path.isfile(gdb) else None,
-        rules=rules,
-        cli_chats_path=cli_path if os.path.isdir(cli_path) else None,
-    )
+    return workspace_storage_fingerprint(workspace_path, entries, rules)
 
 
 def _open_index_db(*, readonly: bool = True) -> sqlite3.Connection | None:

services/summary_cache.py

@@ -115,7 +115,7 @@ def _entry_mtimes(entry: dict[str, Any]) -> list[list[str | int]]:
     }
 
 
-def _workspace_storage_fingerprint(
+def workspace_storage_fingerprint(
     workspace_path: str,
     workspace_entries: list[dict[str, Any]],
     rules: list[RuleTokens],
@@ -260,20 +260,25 @@ def _get_or_build_cached(
     should_cache: Callable[[T], bool] | None = None,
 ) -> T:
     with _summary_cache_lock:
-        fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules)
-        hit = get_unlocked(fingerprint)
+        pre_fingerprint = workspace_storage_fingerprint(
+            workspace_path, workspace_entries, rules,
+        )
+        hit = get_unlocked(pre_fingerprint)
         if hit is not None:
             return hit
 
     built = build_fn()
 
     with _summary_cache_lock:
-        fingerprint = _workspace_storage_fingerprint(workspace_path, workspace_entries, rules)
-        hit = get_unlocked(fingerprint)
+        post_fingerprint = workspace_storage_fingerprint(
+            workspace_path, workspace_entries, rules,
+        )
+        hit = get_unlocked(post_fingerprint)
         if hit is not None:
             return hit
         if should_cache is None or should_cache(built):
-            set_unlocked(fingerprint, built)
+            if _fingerprint_equal(pre_fingerprint, post_fingerprint):
+                set_unlocked(pre_fingerprint, built)
     return built
 
 

tests/test_export_engine.py

@@ -64,7 +64,6 @@ def _minimal_orch(
     return WorkspaceOrchestration(
         workspace_path=tmp_ws,
         workspace_entries=[],
-        fingerprint={},
         ctx=_minimal_ctx(),
         workspace_id_to_display_name=display_name or {},
         workspace_id_to_slug=slug_map or {},

tests/test_summary_cache_concurrency.py

@@ -142,7 +142,7 @@ def test_concurrent_get_or_build_returns_consistent_results(self):
         warm_projects = [
             {"id": "warm", "name": "Warm", "conversationCount": 2, "lastModified": "z"},
         ]
-        fingerprint = summary_cache._workspace_storage_fingerprint(
+        fingerprint = summary_cache.workspace_storage_fingerprint(
             workspace_path,
             workspace_entries,  # type: ignore[arg-type]
             [],