feat: typed models + schema validation at DB read boundaries (closes #24)

api/composers.py

@@ -13,6 +13,7 @@
 
 from utils.workspace_path import resolve_workspace_path
 from utils.path_helpers import to_epoch_ms
+from models import SchemaError, WorkspaceLocalComposer
 
 bp = Blueprint("composers", __name__)
 
@@ -54,15 +55,35 @@ def list_composers():
 
                 if row and row[0]:
                     data = json.loads(row[0])
+                    if not isinstance(data, dict):
+                        raise SchemaError(
+                            "WorkspaceComposers",
+                            "composer.composerData",
+                            hint=f"expected object, got {type(data).__name__}",
+                        )
+                    if "allComposers" not in data:
+                        raise SchemaError("WorkspaceComposers", "allComposers")
                     all_composers = data.get("allComposers")
-                    if isinstance(all_composers, list):
-                        for c in all_composers:
-                            c["conversation"] = c.get("conversation") or []
-                            c["workspaceId"] = name
-                            c["workspaceFolder"] = workspace_folder
-                            composers.append(c)
-            except Exception:
-                pass
+                    if not isinstance(all_composers, list):
+                        raise SchemaError(
+                            "WorkspaceComposers",
+                            "allComposers",
+                            hint=f"expected list, got {type(all_composers).__name__}",
+                        )
+                    for c in all_composers:
+                        try:
+                            WorkspaceLocalComposer.from_dict(c)
+                        except SchemaError as e:
+                            print(f"Schema drift in {db_path}: {e}")
+                            continue
+                        c["conversation"] = c.get("conversation") or []
+                        c["workspaceId"] = name
+                        c["workspaceFolder"] = workspace_folder
+                        composers.append(c)
+            except SchemaError as e:
+                print(f"Schema drift in {db_path}: {e}")
+            except Exception as e:
+                print(f"Failed reading composers from {db_path}: {e}")
 
         composers.sort(key=lambda c: to_epoch_ms(c.get("lastUpdatedAt")), reverse=True)
         return jsonify(composers)

api/search.py

@@ -18,6 +18,7 @@
 from utils.path_helpers import normalize_file_path, get_workspace_folder_paths, to_epoch_ms
 from utils.text_extract import extract_text_from_bubble
 from utils.cli_chat_reader import list_cli_projects, traverse_blobs, messages_to_bubbles
+from models import Composer, SchemaError
 
 bp = Blueprint("search", __name__)
 
@@ -161,17 +162,24 @@ def search():
                 for row in composer_rows:
                     composer_id = row["key"].split(":")[1]
                     try:
-                        cd = json.loads(row["value"])
-                        headers = cd.get("fullConversationHeadersOnly") or []
+                        composer = Composer.from_dict(json.loads(row["value"]), composer_id=composer_id)
+                    except SchemaError as e:
+                        print(f"Schema drift in composer {composer_id}: {e}")
+                        continue
+                    except (json.JSONDecodeError, TypeError, ValueError):
+                        continue
+                    try:
+                        cd = composer.raw
+                        headers = composer.full_conversation_headers_only
                         if not headers:
                             continue
 
-                        title = cd.get("name") or ""
+                        title = composer.name or ""
                         ws_id = composer_id_to_ws.get(composer_id, "global")
                         ws_name = ws_id_to_name.get(ws_id)
                         project_name = ws_name or ("Other chats" if ws_id == "global" else ws_id)
 
-                        model_config = cd.get("modelConfig") or {}
+                        model_config = composer.model_config
                         model_name = model_config.get("modelName")
                         model_names = [model_name] if model_name and model_name != "default" else None
 
@@ -243,7 +251,7 @@ def search():
                                 "workspaceFolder": ws_name,
                                 "chatId": composer_id,
                                 "chatTitle": title,
-                                "timestamp": to_epoch_ms(cd.get("lastUpdatedAt")) or to_epoch_ms(cd.get("createdAt")) or int(datetime.now().timestamp() * 1000),
+                                "timestamp": to_epoch_ms(composer.last_updated_at) or to_epoch_ms(composer.created_at) or int(datetime.now().timestamp() * 1000),
                                 "matchingText": matching_text,
                                 "type": "composer",
                             })

api/workspaces.py

@@ -33,6 +33,7 @@
 )
 from utils.text_extract import extract_text_from_bubble, format_tool_action
 from utils.exclusion_rules import build_searchable_text, is_excluded_by_rules
+from models import Composer, SchemaError
 
 bp = Blueprint("workspaces", __name__)
 
@@ -600,9 +601,15 @@ def list_workspaces():
                     for row in composer_rows:
                         cid = row["key"].split(":")[1]
                         try:
-                            cd = json.loads(row["value"])
+                            composer = Composer.from_dict(json.loads(row["value"]), composer_id=cid)
+                        except SchemaError as e:
+                            print(f"Schema drift in composer {cid}: {e}")
+                            continue
+                        except (json.JSONDecodeError, TypeError, ValueError):
+                            continue
+                        try:
                             pid = _determine_project_for_conversation(
-                                cd, cid, project_layouts_map,
+                                composer.raw, cid, project_layouts_map,
                                 project_name_map, workspace_path_map,
                                 workspace_entries, bubble_map, composer_id_to_ws, invalid_workspace_ids
                             )
@@ -611,16 +618,16 @@ def list_workspaces():
                                 pid = invalid_workspace_aliases.get(mapped_ws)
                             assigned = pid if pid else "global"
 
-                            headers = cd.get("fullConversationHeadersOnly") or []
+                            headers = composer.full_conversation_headers_only
                             has_bubbles = any(bubble_map.get(h.get("bubbleId")) for h in headers)
                             if not has_bubbles:
                                 continue
 
                             conversation_map.setdefault(assigned, []).append({
                                 "composerId": cid,
-                                "name": cd.get("name") or f"Conversation {cid[:8]}",
-                                "lastUpdatedAt": to_epoch_ms(cd.get("lastUpdatedAt")) or to_epoch_ms(cd.get("createdAt")) or 0,
-                                "createdAt": to_epoch_ms(cd.get("createdAt")) or 0,
+                                "name": composer.name or f"Conversation {cid[:8]}",
+                                "lastUpdatedAt": to_epoch_ms(composer.last_updated_at) or to_epoch_ms(composer.created_at) or 0,
+                                "createdAt": to_epoch_ms(composer.created_at) or 0,
                             })
                         except Exception:
                             pass

models/__init__.py

@@ -0,0 +1,15 @@
+from models.cli_session import CliSessionMeta
+from models.conversation import Bubble, Composer, WorkspaceLocalComposer
+from models.errors import SchemaError
+from models.export import ExportEntry
+from models.workspace import Workspace
+
+__all__ = [
+    "Bubble",
+    "CliSessionMeta",
+    "Composer",
+    "ExportEntry",
+    "SchemaError",
+    "Workspace",
+    "WorkspaceLocalComposer",
+]

models/cli_session.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class CliSessionMeta:
+    """CLI session meta blob; latestRootBlobId is the conversation entry point and the only required field."""
+
+    latest_root_blob_id: str
+    created_at: Any = None
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any]) -> "CliSessionMeta":
+        if not isinstance(raw, dict):
+            raise SchemaError(
+                "CliSessionMeta",
+                "meta",
+                hint=f"expected object, got {type(raw).__name__}",
+            )
+        latest = raw.get("latestRootBlobId")
+        if not latest:
+            raise SchemaError("CliSessionMeta", "latestRootBlobId")
+        if not isinstance(latest, str):
+            raise SchemaError(
+                "CliSessionMeta",
+                "latestRootBlobId",
+                hint=f"expected str, got {type(latest).__name__}",
+            )
+        return cls(
+            latest_root_blob_id=latest,
+            created_at=raw.get("createdAt"),
+            raw=raw,
+        )

models/conversation.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class Composer:
+    """Cursor conversation row from globalStorage cursorDiskKV; requires fullConversationHeadersOnly + createdAt."""
+
+    composer_id: str
+    full_conversation_headers_only: list[dict[str, Any]]
+    created_at: Any
+    name: str | None = None
+    last_updated_at: Any = None
+    model_config: dict[str, Any] = field(default_factory=dict)
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer":
+        if not isinstance(raw, dict):
+            raise SchemaError(
+                "Composer",
+                "composerData",
+                hint=f"expected object, got {type(raw).__name__}",
+            )
+        if not isinstance(composer_id, str) or not composer_id:
+            raise SchemaError(
+                "Composer",
+                "composerId",
+                hint=f"expected non-empty str, got {type(composer_id).__name__}",
+            )
+        if "fullConversationHeadersOnly" not in raw:
+            raise SchemaError("Composer", "fullConversationHeadersOnly")
+        if "createdAt" not in raw:
+            raise SchemaError("Composer", "createdAt")
+
+        created_at = raw.get("createdAt")
+        if not isinstance(created_at, (int, float)) or isinstance(created_at, bool):
+            raise SchemaError(
+                "Composer",
+                "createdAt",
+                hint=f"expected timestamp number, got {type(created_at).__name__}",
+            )
+
+        headers = raw.get("fullConversationHeadersOnly")
+        if not isinstance(headers, list):
+            raise SchemaError(
+                "Composer",
+                "fullConversationHeadersOnly",
+                hint=f"expected list, got {type(headers).__name__}",
+            )
+
+        model_config = raw.get("modelConfig") or {}
+        if not isinstance(model_config, dict):
+            model_config = {}
+
+        return cls(
+            composer_id=composer_id,
+            full_conversation_headers_only=headers,
+            created_at=created_at,
+            name=raw.get("name"),
+            last_updated_at=raw.get("lastUpdatedAt"),
+            model_config=model_config,
+            raw=raw,
+        )
+
+
+@dataclass(frozen=True)
+class WorkspaceLocalComposer:
+    """Summary composer row from per-workspace state.vscdb ItemTable; only composerId is required."""
+
+    composer_id: str
+    last_updated_at: Any = None
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any]) -> "WorkspaceLocalComposer":
+        if not isinstance(raw, dict):
+            raise SchemaError(
+                "WorkspaceLocalComposer",
+                "composer",
+                hint=f"expected object, got {type(raw).__name__}",
+            )
+        composer_id = raw.get("composerId")
+        if not isinstance(composer_id, str) or not composer_id:
+            raise SchemaError(
+                "WorkspaceLocalComposer",
+                "composerId",
+                hint=f"expected non-empty str, got {type(composer_id).__name__}",
+            )
+        return cls(
+            composer_id=composer_id,
+            last_updated_at=raw.get("lastUpdatedAt"),
+            raw=raw,
+        )
+
+
+@dataclass(frozen=True)
+class Bubble:
+    """One message in a composer; bubble_id comes from the row key, not the JSON value."""
+
+    bubble_id: str
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any], *, bubble_id: str) -> "Bubble":
+        if not isinstance(raw, dict):
+            raise SchemaError(
+                "Bubble",
+                "bubble",
+                hint=f"expected object, got {type(raw).__name__}",
+            )
+        if not isinstance(bubble_id, str) or not bubble_id:
+            raise SchemaError(
+                "Bubble",
+                "bubbleId",
+                hint=f"expected non-empty str, got {type(bubble_id).__name__}",
+            )
+        return cls(bubble_id=bubble_id, raw=raw)

models/errors.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+
+class SchemaError(ValueError):
+    """Raised when a required Cursor schema field is missing or malformed."""
+
+    def __init__(self, model: str, field: str, *, hint: str | None = None) -> None:
+        self.model = model
+        self.field = field
+        self.hint = hint
+        message = f"{model}: missing required field '{field}'"
+        if hint:
+            message = f"{message} ({hint})"
+        super().__init__(message)

models/export.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class ExportEntry:
+    """One line of manifest.jsonl; log_id / title / workspace required, timestamps optional."""
+
+    log_id: str
+    title: str
+    workspace: str
+    created_at: Any = None
+    updated_at: Any = None
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any]) -> "ExportEntry":
+        if not isinstance(raw, dict):
+            raise SchemaError(
+                "ExportEntry",
+                "entry",
+                hint=f"expected object, got {type(raw).__name__}",
+            )
+        for required in ("log_id", "title", "workspace"):
+            value = raw.get(required)
+            if not isinstance(value, str) or value == "":
+                raise SchemaError(
+                    "ExportEntry",
+                    required,
+                    hint=f"expected non-empty str, got {type(value).__name__}",
+                )
+        return cls(
+            log_id=raw["log_id"],
+            title=raw["title"],
+            workspace=raw["workspace"],
+            created_at=raw.get("created_at"),
+            updated_at=raw.get("updated_at"),
+            raw=raw,
+        )