refactor(services): centralize workspace determination in resolve_workspace_context (#91)

api/export_api.py

@@ -22,17 +22,15 @@
 from utils.text_extract import extract_text_from_bubble, slug
 from utils.exclusion_rules import build_searchable_text, is_excluded_by_rules
 from utils.cursor_md_exporter import cursor_ide_chat_to_markdown
+from services.workspace_context import (
+    enrich_workspace_context_from_global_db,
+    resolve_workspace_context,
+)
 from services.workspace_db import (
-    build_composer_id_to_workspace_id,
-    collect_workspace_entries,
-    load_bubble_map,
     load_code_block_diff_map,
     open_global_db,
 )
-from services.workspace_resolver import (
-    create_project_name_to_workspace_id_map,
-    lookup_workspace_display_name,
-)
+from services.workspace_resolver import lookup_workspace_display_name
 
 bp = Blueprint("export_api", __name__)
 _logger = logging.getLogger(__name__)
@@ -102,9 +100,13 @@ def export_chats():
                 last_export_ms = to_epoch_ms(ts_str)
 
         # ── Workspace scanning via service layer ──────────────────────────────
-        workspace_entries = collect_workspace_entries(workspace_path)
-        composer_id_to_ws = build_composer_id_to_workspace_id(workspace_path, workspace_entries)
-        project_name_map = create_project_name_to_workspace_id_map(workspace_entries)
+        ctx = resolve_workspace_context(
+            workspace_path,
+            include_invalid_workspace_ids=False,
+            include_workspace_path_map=False,
+        )
+        workspace_entries = ctx.workspace_entries
+        composer_id_to_ws = ctx.composer_id_to_workspace_id
 
         # Build display-name and slug maps
         ws_id_to_slug: dict[str, str] = {}
@@ -124,7 +126,10 @@ def export_chats():
             if global_db is None:
                 return jsonify({"error": "Cursor global storage not found"}), 404
 
-            bubble_map = load_bubble_map(global_db)
+            ctx = enrich_workspace_context_from_global_db(
+                ctx, global_db, populate_bubble_map=True,
+            )
+            bubble_map = ctx.bubble_map
             code_block_diff_map = load_code_block_diff_map(global_db)
 
             try:

scripts/export.py

@@ -53,18 +53,15 @@
     cursor_ide_chat_to_markdown,
 )
 from models import ExportEntry, SchemaError  # noqa: E402
+from services.workspace_context import (  # noqa: E402
+    enrich_workspace_context_from_global_db,
+    resolve_workspace_context,
+)
 from services.workspace_db import (  # noqa: E402
-    build_composer_id_to_workspace_id,
-    collect_invalid_workspace_ids,
-    collect_workspace_entries,
-    load_bubble_map,
     load_code_block_diff_map,
-    load_project_layouts_map,
     open_global_db,
 )
 from services.workspace_resolver import (  # noqa: E402
-    create_project_name_to_workspace_id_map,
-    create_workspace_path_to_id_map,
     determine_project_for_conversation,
     infer_invalid_workspace_aliases,
     lookup_workspace_display_name,
@@ -203,11 +200,12 @@ def main():
             )
 
     # ── Workspace scanning via service layer ──────────────────────────────────
-    workspace_entries = collect_workspace_entries(workspace_path)
-    invalid_workspace_ids = collect_invalid_workspace_ids(workspace_entries)
-    project_name_map = create_project_name_to_workspace_id_map(workspace_entries)
-    workspace_path_map = create_workspace_path_to_id_map(workspace_entries)
-    composer_id_to_ws = build_composer_id_to_workspace_id(workspace_path, workspace_entries)
+    ctx = resolve_workspace_context(workspace_path)
+    workspace_entries = ctx.workspace_entries
+    invalid_workspace_ids = ctx.invalid_workspace_ids
+    project_name_map = ctx.project_name_to_workspace_id
+    workspace_path_map = ctx.workspace_path_to_id
+    composer_id_to_ws = ctx.composer_id_to_workspace_id
 
     # Build display-name and slug maps from workspace entries.
     # Entries whose workspace.json cannot be resolved are omitted so the
@@ -235,8 +233,14 @@ def main():
                 global_db_path,
             )
         else:
-            project_layouts_map = load_project_layouts_map(global_db)
-            bubble_map = load_bubble_map(global_db)
+            ctx = enrich_workspace_context_from_global_db(
+                ctx,
+                global_db,
+                populate_project_layouts=True,
+                populate_bubble_map=True,
+            )
+            project_layouts_map = ctx.project_layouts_map
+            bubble_map = ctx.bubble_map
             code_block_diff_map = load_code_block_diff_map(global_db)
 
             try:

services/workspace_context.py

@@ -0,0 +1,133 @@
+"""Workspace determination ceremony — single orchestrator for shared maps."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from typing import TYPE_CHECKING
+
+from services.workspace_db import (
+    build_composer_id_to_workspace_id,
+    build_composer_id_to_workspace_id_cached,
+    collect_invalid_workspace_ids,
+    collect_workspace_entries,
+    load_bubble_map,
+    load_project_layouts_map,
+)
+from services.workspace_resolver import (
+    create_project_name_to_workspace_id_map,
+    create_workspace_path_to_id_map,
+)
+
+if TYPE_CHECKING:
+    import sqlite3
+
+
+@dataclass(frozen=True)
+class WorkspaceContext:
+    """Precomputed workspace-resolution maps for conversation assignment."""
+
+    workspace_path: str
+    workspace_entries: list[dict]
+    invalid_workspace_ids: set[str]
+    composer_id_to_workspace_id: dict[str, str]
+    project_name_to_workspace_id: dict[str, str]
+    workspace_path_to_id: dict[str, str]
+    project_layouts_map: dict[str, list]
+    bubble_map: dict[str, dict]
+
+
+def resolve_workspace_context(
+    workspace_path: str,
+    *,
+    workspace_entries: list[dict] | None = None,
+    rules: list | None = None,
+    nocache: bool = False,
+    use_composer_cache: bool = False,
+    include_invalid_workspace_ids: bool = True,
+    include_workspace_path_map: bool = True,
+    global_db: sqlite3.Connection | None = None,
+    populate_project_layouts: bool = False,
+    populate_bubble_map: bool = False,
+) -> WorkspaceContext:
+    """Run the workspace-determination ceremony and return a typed context.
+
+    Always resolves ``workspace_entries`` (when not supplied), composer and
+    project-name maps. Optional pieces are controlled by flags so lightweight
+    consumers (e.g. HTTP export) can omit unused maps.
+
+    Args:
+        workspace_path: Cursor ``workspaceStorage`` root.
+        workspace_entries: Pre-collected entries; when ``None``, scanned from disk.
+        rules: Exclusion rules; required when ``use_composer_cache`` is ``True``.
+        nocache: Skip the mtime-keyed composer-map disk cache.
+        use_composer_cache: Use :func:`build_composer_id_to_workspace_id_cached`.
+        include_invalid_workspace_ids: When ``False``, ``invalid_workspace_ids`` is empty.
+        include_workspace_path_map: When ``False``, ``workspace_path_to_id`` is empty.
+        global_db: Open global ``state.vscdb`` connection for optional KV loads.
+        populate_project_layouts: Populate ``project_layouts_map`` from *global_db*.
+        populate_bubble_map: Populate ``bubble_map`` from *global_db*.
+
+    Returns:
+        :class:`WorkspaceContext` with all requested maps populated.
+    """
+    entries = (
+        workspace_entries
+        if workspace_entries is not None
+        else collect_workspace_entries(workspace_path)
+    )
+    invalid_ids = (
+        collect_invalid_workspace_ids(entries)
+        if include_invalid_workspace_ids
+        else set()
+    )
+    project_name_map = create_project_name_to_workspace_id_map(entries)
+    workspace_path_map = (
+        create_workspace_path_to_id_map(entries)
+        if include_workspace_path_map
+        else {}
+    )
+    if use_composer_cache:
+        if rules is None:
+            raise ValueError("rules is required when use_composer_cache=True")
+        composer_id_to_ws = build_composer_id_to_workspace_id_cached(
+            workspace_path, entries, rules, nocache=nocache,
+        )
+    else:
+        composer_id_to_ws = build_composer_id_to_workspace_id(workspace_path, entries)
+
+    project_layouts: dict[str, list] = {}
+    bubble_map: dict[str, dict] = {}
+    if global_db is not None:
+        if populate_project_layouts:
+            project_layouts = load_project_layouts_map(global_db)
+        if populate_bubble_map:
+            bubble_map = load_bubble_map(global_db)
+
+    return WorkspaceContext(
+        workspace_path=workspace_path,
+        workspace_entries=entries,
+        invalid_workspace_ids=invalid_ids,
+        composer_id_to_workspace_id=composer_id_to_ws,
+        project_name_to_workspace_id=project_name_map,
+        workspace_path_to_id=workspace_path_map,
+        project_layouts_map=project_layouts,
+        bubble_map=bubble_map,
+    )
+
+
+def enrich_workspace_context_from_global_db(
+    ctx: WorkspaceContext,
+    global_db: sqlite3.Connection,
+    *,
+    populate_project_layouts: bool = False,
+    populate_bubble_map: bool = False,
+) -> WorkspaceContext:
+    """Return *ctx* with global KV maps loaded from an open global DB connection."""
+    updates: dict = {}
+    if populate_project_layouts:
+        updates["project_layouts_map"] = load_project_layouts_map(global_db)
+    if populate_bubble_map:
+        updates["bubble_map"] = load_bubble_map(global_db)
+    if not updates:
+        return ctx
+    return replace(ctx, **updates)

services/workspace_listing.py

@@ -24,10 +24,9 @@
     nocache_enabled,
     set_cached_projects,
 )
+from services.workspace_context import resolve_workspace_context
 from services.workspace_db import (
     COMPOSER_ROWS_WITH_HEADERS_SQL,
-    build_composer_id_to_workspace_id_cached,
-    collect_invalid_workspace_ids,
     collect_workspace_entries,
     global_storage_db_path,
     load_project_layouts_for_composer,
@@ -36,8 +35,6 @@
 )
 from utils.workspace_path import get_cli_chats_path
 from services.workspace_resolver import (
-    create_project_name_to_workspace_id_map,
-    create_workspace_path_to_id_map,
     determine_project_for_conversation,
     infer_invalid_workspace_aliases,
     infer_workspace_name_from_context,
@@ -125,13 +122,17 @@ def _build_workspace_projects_uncached(
     nocache: bool,
 ) -> tuple[list[dict], list[dict]]:
     parse_warnings = ParseWarningCollector()
-    invalid_workspace_ids = collect_invalid_workspace_ids(workspace_entries)
-
-    project_name_map = create_project_name_to_workspace_id_map(workspace_entries)
-    workspace_path_map = create_workspace_path_to_id_map(workspace_entries)
-    composer_id_to_ws = build_composer_id_to_workspace_id_cached(
-        workspace_path, workspace_entries, rules, nocache=nocache,
+    ctx = resolve_workspace_context(
+        workspace_path,
+        workspace_entries=workspace_entries,
+        rules=rules,
+        nocache=nocache,
+        use_composer_cache=True,
     )
+    invalid_workspace_ids = ctx.invalid_workspace_ids
+    project_name_map = ctx.project_name_to_workspace_id
+    workspace_path_map = ctx.workspace_path_to_id
+    composer_id_to_ws = ctx.composer_id_to_workspace_id
 
     conversation_map: dict[str, list] = {}