refactor: decouple API handlers from _-prefixed service internals (closes #73)

api/export_api.py

@@ -21,15 +21,15 @@
 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_db import (
-    _build_composer_id_to_workspace_id,
-    _collect_workspace_entries,
+    build_composer_id_to_workspace_id,
+    collect_workspace_entries,
     load_bubble_map,
     load_code_block_diff_map,
-    _open_global_db,
+    open_global_db,
 )
 from services.workspace_resolver import (
-    _get_workspace_display_name,
-    _create_project_name_to_workspace_id_map,
+    create_project_name_to_workspace_id_map,
+    lookup_workspace_display_name,
 )
 
 bp = Blueprint("export_api", __name__)
@@ -96,15 +96,15 @@ 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)
+        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)
 
         # Build display-name and slug maps
         ws_id_to_slug: dict[str, str] = {}
         ws_id_to_display_name: dict[str, str] = {}
         for e in workspace_entries:
-            display = _get_workspace_display_name(workspace_path, e["name"])
+            display = lookup_workspace_display_name(workspace_path, e["name"])
             if display != e["name"]:
                 ws_id_to_display_name[e["name"]] = display
                 ws_id_to_slug[e["name"]] = slug(display)
@@ -114,7 +114,7 @@ def export_chats():
         rules = current_app.config.get("EXCLUSION_RULES") or []
 
         # ── Database reading via service layer ────────────────────────────────
-        with _open_global_db(workspace_path) as (global_db, global_db_path):
+        with open_global_db(workspace_path) as (global_db, global_db_path):
             if global_db is None:
                 return jsonify({"error": "Cursor global storage not found"}), 404
 

api/workspaces.py

@@ -22,19 +22,21 @@
 )
 from utils.workspace_descriptor import read_json_file
 from services.workspace_resolver import (
-    _infer_workspace_name_from_context,
-    # Re-exported for back-compat with existing tests that import from api.workspaces
-    # directly (test_invalid_workspace_aliases, test_workspace_assignment_fallback,
-    # test_workspace_name_inference, test_models_wired_at_read_sites).
-    # Production callers should import from services.workspace_resolver instead.
-    _determine_project_for_conversation,  # noqa: F401
-    _infer_invalid_workspace_aliases,  # noqa: F401
-    _get_workspace_display_name,  # noqa: F401
+    determine_project_for_conversation,
+    infer_invalid_workspace_aliases,
+    infer_workspace_name_from_context,
+    lookup_workspace_display_name,
 )
-from services.cli_tabs import _get_cli_workspace_tabs
+from services.cli_tabs import get_cli_workspace_tabs
 from services.workspace_listing import list_workspace_projects
 from services.workspace_tabs import assemble_workspace_tabs
 
+# Re-exported for back-compat with tests that import from api.workspaces directly.
+_determine_project_for_conversation = determine_project_for_conversation  # noqa: F401
+_infer_invalid_workspace_aliases = infer_invalid_workspace_aliases  # noqa: F401
+_get_workspace_display_name = lookup_workspace_display_name  # noqa: F401
+_infer_workspace_name_from_context = infer_workspace_name_from_context  # noqa: F401
+
 # Re-exported for tests/test_models_wired_at_read_sites.py — the typed-model
 # spy harness patches `workspaces_mod.Bubble` / `.Composer` / `.Workspace` to
 # verify that production read paths actually call from_dict. The classes
@@ -121,12 +123,12 @@ def get_workspace(workspace_id):
             if derived_name:
                 workspace_name = derived_name
             elif workspace_name == workspace_id:
-                inferred = _infer_workspace_name_from_context(workspace_path, workspace_id)
+                inferred = infer_workspace_name_from_context(workspace_path, workspace_id)
                 if inferred:
                     workspace_name = inferred
         except Exception as e:
             warn_workspace_json_read(_logger, workspace_id, e)
-            inferred = _infer_workspace_name_from_context(workspace_path, workspace_id)
+            inferred = infer_workspace_name_from_context(workspace_path, workspace_id)
             if inferred:
                 workspace_name = inferred
 
@@ -150,7 +152,7 @@ def get_workspace(workspace_id):
 @bp.route("/api/workspaces/<workspace_id>/tabs")
 def get_workspace_tabs(workspace_id):
     if workspace_id.startswith("cli:"):
-        return _get_cli_workspace_tabs(workspace_id)
+        return get_cli_workspace_tabs(workspace_id)
     try:
         workspace_path = resolve_workspace_path()
         rules = current_app.config.get("EXCLUSION_RULES") or []

services/cli_tabs.py

@@ -12,8 +12,16 @@
 from utils.workspace_path import get_cli_chats_path
 
 
-def _get_cli_workspace_tabs(workspace_id: str):
-    """Return tabs for a Cursor CLI project (workspace_id starts with "cli:")."""
+def get_cli_workspace_tabs(workspace_id: str):
+    """Return Flask JSON response with tabs for a Cursor CLI project.
+
+    Args:
+        workspace_id: Workspace id with ``cli:`` prefix (e.g. ``cli:proj-1``).
+
+    Returns:
+        ``(jsonify({...}), status)`` tuple suitable for a Flask route handler.
+        Status is 404 when the project is missing, 500 on unexpected errors.
+    """
     try:
         project_id = workspace_id[4:]
         cli_projects = list_cli_projects(get_cli_chats_path())
@@ -136,3 +144,7 @@ def _get_cli_workspace_tabs(workspace_id: str):
             exc_info=True,
         )
         return jsonify({"error": "Failed to get CLI workspace tabs"}), 500
+
+
+# Backward-compatible alias for tests and legacy imports.
+_get_cli_workspace_tabs = get_cli_workspace_tabs

services/workspace_db.py

@@ -15,7 +15,7 @@
 
 # ── Global-DB KV loaders ────────────────────────────────────────────────────
 # Each function accepts an already-opened sqlite3.Connection (row_factory must
-# be set to sqlite3.Row by the caller, as _open_global_db does) and returns
+# be set to sqlite3.Row by the caller, as open_global_db does) and returns
 # a populated dict.  sqlite3.Error is caught internally so a missing or
 # corrupt table cannot propagate to callers.
 
@@ -113,8 +113,16 @@ def load_code_block_diff_map(global_db) -> dict[str, list]:
     return diff_map
 
 
-def _collect_workspace_entries(workspace_path: str) -> list[dict]:
-    """Scan workspace directory and return entries with workspace.json."""
+def collect_workspace_entries(workspace_path: str) -> list[dict]:
+    """Scan workspace directory and return entries with workspace.json.
+
+    Args:
+        workspace_path: Cursor workspace storage root (parent of per-workspace folders).
+
+    Returns:
+        List of dicts with keys ``name`` (folder id) and ``workspaceJsonPath``.
+        Returns an empty list if ``workspace_path`` is missing or unreadable.
+    """
     entries = []
     try:
         for name in os.listdir(workspace_path):
@@ -131,8 +139,15 @@ def _collect_workspace_entries(workspace_path: str) -> list[dict]:
     return entries
 
 
-def _collect_invalid_workspace_ids(workspace_entries: list[dict]) -> set[str]:
-    """Workspace IDs whose descriptors have no resolvable folder paths."""
+def collect_invalid_workspace_ids(workspace_entries: list[dict]) -> set[str]:
+    """Return workspace IDs whose descriptors have no resolvable folder paths.
+
+    Args:
+        workspace_entries: Output of :func:`collect_workspace_entries`.
+
+    Returns:
+        Set of workspace folder names that cannot be mapped to a folder path.
+    """
     invalid: set[str] = set()
     for entry in workspace_entries:
         try:
@@ -149,8 +164,19 @@ def _collect_invalid_workspace_ids(workspace_entries: list[dict]) -> set[str]:
     return invalid
 
 
-def _build_composer_id_to_workspace_id(workspace_path: str, workspace_entries: list) -> dict:
-    """Build mapping: composerId -> workspaceId from per-workspace state.vscdb."""
+def build_composer_id_to_workspace_id(workspace_path: str, workspace_entries: list) -> dict:
+    """Build mapping from composer ID to workspace folder name.
+
+    Reads ``composer.composerData`` from each workspace's ``state.vscdb``.
+    Skips workspaces with missing databases or malformed JSON.
+
+    Args:
+        workspace_path: Cursor workspace storage root.
+        workspace_entries: Output of :func:`collect_workspace_entries`.
+
+    Returns:
+        Dict mapping ``composerId`` strings to workspace folder names.
+    """
     mapping: dict = {}
     for entry in workspace_entries:
         db_path = os.path.join(workspace_path, entry["name"], "state.vscdb")
@@ -187,8 +213,17 @@ def _build_composer_id_to_workspace_id(workspace_path: str, workspace_entries: l
 
 
 @contextmanager
-def _open_global_db(workspace_path: str):
-    """Yield (conn, path) for the global-storage SQLite db (read-only); (None, path) if the file is missing."""
+def open_global_db(workspace_path: str):
+    """Open Cursor global storage SQLite database read-only.
+
+    Args:
+        workspace_path: Cursor workspace storage root.
+
+    Yields:
+        ``(conn, path)`` where ``conn`` is a :class:`sqlite3.Connection` with
+        ``row_factory=sqlite3.Row``, or ``None`` if the database file is missing
+        or cannot be opened. ``path`` is always the resolved global DB path.
+    """
     global_db_path = os.path.join(workspace_path, "..", "globalStorage", "state.vscdb")
     global_db_path = os.path.normpath(global_db_path)
     if not os.path.isfile(global_db_path):
@@ -205,3 +240,10 @@ def _open_global_db(workspace_path: str):
         yield conn, global_db_path
     finally:
         conn.close()
+
+
+# Backward-compatible aliases for tests and legacy imports.
+_collect_workspace_entries = collect_workspace_entries
+_collect_invalid_workspace_ids = collect_invalid_workspace_ids
+_build_composer_id_to_workspace_id = build_composer_id_to_workspace_id
+_open_global_db = open_global_db

services/workspace_listing.py

@@ -20,37 +20,37 @@
 from utils.workspace_path import get_cli_chats_path
 from models import Composer, ParseWarningCollector, SchemaError
 from services.workspace_db import (
-    _build_composer_id_to_workspace_id,
-    _collect_invalid_workspace_ids,
-    _collect_workspace_entries,
+    build_composer_id_to_workspace_id,
+    collect_invalid_workspace_ids,
+    collect_workspace_entries,
     load_bubble_map,
     load_project_layouts_map,
-    _open_global_db,
+    open_global_db,
 )
 from services.workspace_resolver import (
-    _create_project_name_to_workspace_id_map,
-    _create_workspace_path_to_id_map,
-    _determine_project_for_conversation,
-    _get_workspace_display_name,
-    _infer_invalid_workspace_aliases,
-    _infer_workspace_name_from_context,
+    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,
+    lookup_workspace_display_name,
 )
 
 
 def list_workspace_projects(workspace_path: str, rules: list) -> tuple[list[dict], list[dict]]:
     """Return (projects, warnings) for GET /api/workspaces."""
     parse_warnings = ParseWarningCollector()
-    workspace_entries = _collect_workspace_entries(workspace_path)
-    invalid_workspace_ids = _collect_invalid_workspace_ids(workspace_entries)
+    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)
+    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)
 
     conversation_map: dict[str, list] = {}
 
     # closing semantics now baked into the context manager (issue #17).
-    with _open_global_db(workspace_path) as (global_db, _):
+    with open_global_db(workspace_path) as (global_db, _):
         if global_db:
             def _safe_fetchall(query: str, params: tuple = ()) -> list:
                 try:
@@ -65,7 +65,7 @@ def _safe_fetchall(query: str, params: tuple = ()) -> list:
                 project_layouts_map: dict[str, list] = load_project_layouts_map(global_db)
                 bubble_map: dict[str, dict] = load_bubble_map(global_db)
 
-                invalid_workspace_aliases = _infer_invalid_workspace_aliases(
+                invalid_workspace_aliases = infer_invalid_workspace_aliases(
                     composer_rows=composer_rows,
                     project_layouts_map=project_layouts_map,
                     project_name_map=project_name_map,
@@ -107,7 +107,7 @@ def _safe_fetchall(query: str, params: tuple = ()) -> list:
                         continue
                     cd = composer.raw
                     try:
-                        pid = _determine_project_for_conversation(
+                        pid = determine_project_for_conversation(
                             cd, cid, project_layouts_map,
                             project_name_map, workspace_path_map,
                             workspace_entries, bubble_map, composer_id_to_ws, invalid_workspace_ids,
@@ -192,9 +192,9 @@ def _safe_fetchall(query: str, params: tuple = ()) -> list:
             )
             mtime = 0
 
-        workspace_name = _get_workspace_display_name(workspace_path, primary["name"])
+        workspace_name = lookup_workspace_display_name(workspace_path, primary["name"])
         if workspace_name == primary["name"]:
-            inferred = _infer_workspace_name_from_context(workspace_path, primary["name"])
+            inferred = infer_workspace_name_from_context(workspace_path, primary["name"])
             workspace_name = inferred or f"Project {primary['name'][:8]}"
 
         if is_excluded_by_rules(rules, workspace_name):