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

Add models/ package with @DataClass definitions for Workspace, Composer,
Bubble, CliSessionMeta, ExportEntry. Each model has a from_dict() classmethod
that raises SchemaError when critical fields are missing, replacing silent
dict.get() fallbacks at the four database read boundaries identified by
the eval (Section 5.1, Schema Fragility).

Wire into:
  - api/workspaces.py:601   — composerData rows in workspace listing
  - api/composers.py:57-63  — allComposers envelope in per-workspace fetch
  - api/search.py:163-164   — composerData rows during search
  - utils/cli_chat_reader.py:93-98 — CLI session meta blob in traverse_blobs

Schema drift surfaces as a printed warning + skipped row, not a silent
empty result. Existing call sites preserve their JSON response shapes
(behavior-preserving wire-in).

Author: timon0305

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
 
 bp = Blueprint("composers", __name__)
 
@@ -54,13 +55,24 @@ def list_composers():
 
                 if row and row[0]:
                     data = json.loads(row[0])
+                    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)
+                    if not isinstance(all_composers, list):
+                        raise SchemaError(
+                            "WorkspaceComposers",
+                            "allComposers",
+                            hint=f"expected list, got {type(all_composers).__name__}",
+                        )
+                    for c in all_composers:
+                        if not isinstance(c, dict) or not c.get("composerId"):
+                            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:
                 pass
 

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,24 @@
+"""Typed domain models for Cursor schema (closes #24).
+
+Cursor's on-disk JSON shapes are not versioned, so silent renames of fields
+like ``composerData`` or ``latestRootBlobId`` would otherwise pass through
+``dict.get(...)`` with a fallback default and produce empty conversations
+with no error raised. The models here add a schema-validation boundary at
+database read sites: ``from_dict`` classmethods raise ``SchemaError`` when
+critical fields are missing, so drift becomes loud instead of silent.
+"""
+
+from models.cli_session import CliSessionMeta
+from models.conversation import Bubble, Composer
+from models.errors import SchemaError
+from models.export import ExportEntry
+from models.workspace import Workspace
+
+__all__ = [
+    "Bubble",
+    "CliSessionMeta",
+    "Composer",
+    "ExportEntry",
+    "SchemaError",
+    "Workspace",
+]

models/cli_session.py

@@ -0,0 +1,43 @@
+"""CliSessionMeta — typed model for the Cursor CLI ``meta`` blob."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class CliSessionMeta:
+    """The ``meta`` blob at the head of a Cursor CLI ``store.db`` blob graph.
+
+    ``latestRootBlobId`` is the entry point for the conversation reconstruction
+    BFS in ``utils/cli_chat_reader.traverse_blobs``; without it, the entire
+    conversation is unreachable. ``createdAt`` is documented as part of the
+    meta-blob schema (see ``utils/cli_chat_reader`` module docstring) and is
+    captured here, but it is not gated on — only ``latestRootBlobId`` is the
+    hard requirement, since that is the only field whose absence prevents
+    conversation reconstruction.
+    """
+
+    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":
+        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,83 @@
+"""Composer (conversation) and Bubble (message) typed models."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class Composer:
+    """A Cursor conversation (a.k.a. "composer") row.
+
+    Required fields per the schema-validation contract:
+      - ``fullConversationHeadersOnly`` — without this, a composer cannot be
+        rendered (no message order is recoverable). This is the only hard
+        requirement: real Cursor data legitimately omits ``createdAt`` for
+        older composers (the existing call sites already fall back to
+        ``lastUpdatedAt`` and then to epoch zero), so it is captured but
+        not gated on.
+
+    The composer ID is intentionally passed in as a constructor argument
+    rather than read from ``raw`` because Cursor stores it in the row key
+    (``composerData:<id>``) rather than in the JSON value.
+    """
+
+    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 composer_id:
+            raise SchemaError("Composer", "composerId", hint="empty composer ID")
+        if "fullConversationHeadersOnly" not in raw:
+            raise SchemaError("Composer", "fullConversationHeadersOnly")
+
+        headers = raw.get("fullConversationHeadersOnly") or []
+        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=raw.get("createdAt"),
+            name=raw.get("name"),
+            last_updated_at=raw.get("lastUpdatedAt"),
+            model_config=model_config,
+            raw=raw,
+        )
+
+
+@dataclass(frozen=True)
+class Bubble:
+    """A single message bubble within a composer.
+
+    The bubble ID lives in the row key (``bubbleId:<composer_id>:<bubble_id>``)
+    rather than the JSON value, so it is passed in explicitly. The raw dict
+    is preserved to keep downstream rendering code (which still walks the
+    untyped shape) working without modification.
+    """
+
+    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 bubble_id:
+            raise SchemaError("Bubble", "bubbleId", hint="empty bubble ID")
+        return cls(bubble_id=bubble_id, raw=raw)

models/errors.py

@@ -0,0 +1,22 @@
+"""Exception types for the typed-model schema-validation layer."""
+
+from __future__ import annotations
+
+
+class SchemaError(ValueError):
+    """Raised when a required Cursor schema field is missing or malformed.
+
+    Inherits from ``ValueError`` so call sites that already catch generic
+    deserialisation errors (e.g. ``json.JSONDecodeError`` is a subclass of
+    ``ValueError``) also catch schema drift without needing a separate
+    ``except`` clause. New code should catch ``SchemaError`` explicitly.
+    """
+
+    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,40 @@
+"""ExportEntry — typed model for an export manifest record (JSONL line)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class ExportEntry:
+    """A single record in the export manifest (one line in ``manifest.jsonl``).
+
+    Required fields are the YAML-frontmatter keys that downstream tooling
+    indexes against: a missing ``log_id`` makes the entry unaddressable, and
+    a missing ``title`` produces unreadable output. Timestamps are optional —
+    not every Cursor conversation has both a creation and update time.
+    """
+
+    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":
+        for required in ("log_id", "title", "workspace"):
+            if required not in raw or raw[required] in (None, ""):
+                raise SchemaError("ExportEntry", required)
+        return cls(
+            log_id=str(raw["log_id"]),
+            title=str(raw["title"]),
+            workspace=str(raw["workspace"]),
+            created_at=raw.get("created_at"),
+            updated_at=raw.get("updated_at"),
+            raw=raw,
+        )

models/workspace.py

@@ -0,0 +1,37 @@
+"""Workspace — typed model for a single Cursor workspace folder."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from models.errors import SchemaError
+
+
+@dataclass(frozen=True)
+class Workspace:
+    """A Cursor workspace entry.
+
+    The workspace ID is the directory name on disk (Cursor uses random
+    short hashes as workspace IDs) and is passed in explicitly. ``folder``
+    is the absolute path of the project the workspace targets, read from
+    ``workspace.json``; it may legitimately be ``None`` for a CLI-only
+    workspace, so missing-folder is not a schema error.
+    """
+
+    workspace_id: str
+    folder: str | None = None
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any], *, workspace_id: str) -> "Workspace":
+        if not workspace_id:
+            raise SchemaError("Workspace", "workspaceId", hint="empty workspace ID")
+        folder = raw.get("folder")
+        if folder is not None and not isinstance(folder, str):
+            raise SchemaError(
+                "Workspace",
+                "folder",
+                hint=f"expected str or None, got {type(folder).__name__}",
+            )
+        return cls(workspace_id=workspace_id, folder=folder, raw=raw)