review: route per-row workspace-local composer drift through SchemaError (#30)

CodeRabbit flagged that the inline ``isinstance(c, dict) or not c.get(...)``
skip in api/composers.py masked row-level drift instead of surfacing it the
way the rest of the boundary does. The exact suggested fix
(Composer.from_dict(c)) would have rejected every workspace-local record:
those carry composerId + lastUpdatedAt but not fullConversationHeadersOnly
or createdAt, which only exist on the global cursorDiskKV rows.

Add WorkspaceLocalComposer to models/conversation.py — a slim sibling
model that validates the workspace-local shape (composerId required as a
non-empty string; non-dict raw raises). api/composers.py:list_composers
now wraps each entry in WorkspaceLocalComposer.from_dict, so per-row
drift logs a Schema drift warning and skips, matching every other
read-site in this PR.

Regression coverage in tests/test_models.py: parses good shape;
raises SchemaError for missing/empty/non-string composerId and for
non-dict payloads.

204 tests pass (was 201; +3); mypy clean on models/; real Cursor smoke
test shows zero drift warnings.

Author: timon0305

api/composers.py

@@ -13,7 +13,7 @@
 
 from utils.workspace_path import resolve_workspace_path
 from utils.path_helpers import to_epoch_ms
-from models import SchemaError
+from models import SchemaError, WorkspaceLocalComposer
 
 bp = Blueprint("composers", __name__)
 
@@ -71,7 +71,10 @@ def list_composers():
                             hint=f"expected list, got {type(all_composers).__name__}",
                         )
                     for c in all_composers:
-                        if not isinstance(c, dict) or not c.get("composerId"):
+                        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

models/__init__.py

@@ -9,7 +9,7 @@
 """
 
 from models.cli_session import CliSessionMeta
-from models.conversation import Bubble, Composer
+from models.conversation import Bubble, Composer, WorkspaceLocalComposer
 from models.errors import SchemaError
 from models.export import ExportEntry
 from models.workspace import Workspace
@@ -21,4 +21,5 @@
     "ExportEntry",
     "SchemaError",
     "Workspace",
+    "WorkspaceLocalComposer",
 ]

models/conversation.py

@@ -70,6 +70,46 @@ def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer":
         )
 
 
+@dataclass(frozen=True)
+class WorkspaceLocalComposer:
+    """A composer entry from ``composer.composerData`` ItemTable rows.
+
+    These are summary records that live in each per-workspace ``state.vscdb``.
+    They share ``composerId`` and ``lastUpdatedAt`` with the global composer
+    schema, but they do **not** carry ``fullConversationHeadersOnly`` or
+    ``createdAt`` — those only exist on the global ``cursorDiskKV`` rows that
+    ``Composer.from_dict`` validates. Treating both shapes through the same
+    model would reject every workspace-local entry, so this slim model
+    exists to keep schema-drift detection at the boundary without conflating
+    the two storage paths.
+    """
+
+    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:
     """A single message bubble within a composer.

tests/test_models.py

@@ -28,6 +28,7 @@
     ExportEntry,
     SchemaError,
     Workspace,
+    WorkspaceLocalComposer,
 )
 from utils.cli_chat_reader import _extract_blob_refs
 
@@ -85,6 +86,17 @@ def test_workspace_parses_with_optional_folder(self) -> None:
         ws_no_folder = Workspace.from_dict({}, workspace_id="cli-only")
         self.assertEqual(ws_no_folder.folder, None)
 
+    def test_workspace_local_composer_parses(self) -> None:
+        # Workspace-local composers (composer.composerData ItemTable rows)
+        # carry composerId + lastUpdatedAt but not the global-storage fields.
+        c = WorkspaceLocalComposer.from_dict({
+            "composerId": "cid-local-1",
+            "lastUpdatedAt": 1_715_000_500_000,
+            "conversation": [],
+        })
+        self.assertEqual(c.composer_id, "cid-local-1")
+        self.assertEqual(c.last_updated_at, 1_715_000_500_000)
+
     def test_export_entry_parses(self) -> None:
         entry = ExportEntry.from_dict({
             "log_id": "L1",
@@ -149,6 +161,25 @@ def test_bubble_empty_id_raises(self) -> None:
         with self.assertRaises(SchemaError):
             Bubble.from_dict({"text": "hi"}, bubble_id="")
 
+    def test_workspace_local_composer_missing_id_raises(self) -> None:
+        # Regression for CodeRabbit's per-row drift comment on api/composers.py:
+        # previously a row missing composerId was silently skipped. Now drift
+        # raises SchemaError so api/composers.py can log + skip explicitly.
+        for bad in (
+            {"lastUpdatedAt": 0},           # composerId absent
+            {"composerId": ""},             # empty string
+            {"composerId": None},           # None
+            {"composerId": 123},            # wrong type
+        ):
+            with self.assertRaises(SchemaError, msg=f"failed for {bad!r}") as cm:
+                WorkspaceLocalComposer.from_dict(bad)
+            self.assertEqual(cm.exception.field, "composerId")
+
+    def test_workspace_local_composer_non_dict_raises(self) -> None:
+        for bad in (None, [], "str", 42):
+            with self.assertRaises(SchemaError):
+                WorkspaceLocalComposer.from_dict(bad)  # type: ignore[arg-type]
+
     def test_export_entry_missing_required_raises(self) -> None:
         with self.assertRaises(SchemaError) as cm:
             ExportEntry.from_dict({"title": "x", "workspace": "w"})