review: wire typed models load-bearing across read sites (#30)

Closes five blocking findings from Brad's review pass:

1. Bubble / Workspace / ExportEntry were defined and unit-tested but had
   zero production callers — the PR's own "Used at" table didn't match
   the code. Wired each at the cited sites:
     * Bubble in api/workspaces.py (both bubble loaders, line 584 + 1023)
     * Bubble in api/search.py:150
     * Workspace in api/composers.py:44 and api/workspaces.py:54
     * ExportEntry reader in scripts/export.py:66 — paired with writer
       updates so the on-disk shape matches what the model accepts.

2. get_workspace_tabs() in api/workspaces.py bypassed Composer.from_dict
   while list_workspaces() in the same file validated. The two primary
   conversation-browsing paths now agree on what counts as a valid row.

3. get_composer() returned the raw dict from either per-workspace DB or
   global storage with no validation. Now goes through
   WorkspaceLocalComposer.from_dict (per-workspace, drift logged + skipped)
   and Composer.from_dict (global fallback, drift -> 404 + log).

4. WorkspaceLocalComposer.from_dict result was discarded after the gate.
   The typed object is now load-bearing: the sort key reads
   local.last_updated_at and the JSON's composerId/lastUpdatedAt are
   written from the validated fields, not the raw dict.

5. SchemaError message always read "missing required field" even on
   type/shape mismatches. Now reads "invalid field" when hint is set
   and "missing required field" only when hint is None, so log grepping
   can tell the two drift modes apart.

Plus one forward-looking note: added a WHY comment at the numeric-only
createdAt check in models/conversation.py pointing at the 17/17 live
scan that justifies the strict isinstance(int, float) gate.

11 new regression tests:
  - tests/test_models_wired_at_read_sites.py (new file, 10 tests)
    pinning each from_dict call site with spy + side_effect patches so
    a future refactor that drops the typed model fails loudly.
  - tests/test_models.py
    one new test asserting the SchemaError wording distinction holds.

Verified locally:
  - python -m unittest discover tests: 220 passed, OK (was 209)
  - python -m unittest tests.test_models_wired_at_read_sites: 10/10 pass
  - python app.py boots clean, curl / returns 200

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, WorkspaceLocalComposer
+from models import Composer, SchemaError, Workspace, WorkspaceLocalComposer
 
 bp = Blueprint("composers", __name__)
 
@@ -41,9 +41,11 @@ def list_composers():
 
             workspace_folder = None
             try:
-                wd = _read_json_file(wj_path)
-                workspace_folder = wd.get("folder")
-            except Exception:
+                workspace = Workspace.from_dict(_read_json_file(wj_path), workspace_id=name)
+                workspace_folder = workspace.folder
+            except (SchemaError, OSError, ValueError):
+                # Missing / malformed workspace.json is non-fatal — the row still
+                # contributes its composer data, just without a folder hint.
                 pass
 
             try:
@@ -72,21 +74,27 @@ def list_composers():
                         )
                     for c in all_composers:
                         try:
-                            WorkspaceLocalComposer.from_dict(c)
+                            local = WorkspaceLocalComposer.from_dict(c)
                         except SchemaError as e:
                             print(f"Schema drift in {db_path}: {e}")
                             continue
+                        # Use the typed view downstream so the dataclass is
+                        # load-bearing, not just a filter (Brad's review): the
+                        # sort key and the JSON's composerId both read off the
+                        # validated values, not the raw dict.
+                        c["composerId"] = local.composer_id
+                        c["lastUpdatedAt"] = local.last_updated_at
                         c["conversation"] = c.get("conversation") or []
                         c["workspaceId"] = name
                         c["workspaceFolder"] = workspace_folder
-                        composers.append(c)
+                        composers.append((local, 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)
+        composers.sort(key=lambda pair: to_epoch_ms(pair[0].last_updated_at), reverse=True)
+        return jsonify([c for _, c in composers])
 
     except Exception as e:
         print(f"Failed to get composers: {e}")
@@ -117,9 +125,17 @@ def get_composer(composer_id):
                 if row and row[0]:
                     data = json.loads(row[0])
                     for c in (data.get("allComposers") or []):
-                        if c.get("composerId") == composer_id:
-                            return jsonify(c)
-            except Exception:
+                        if isinstance(c, dict) and c.get("composerId") == composer_id:
+                            try:
+                                local = WorkspaceLocalComposer.from_dict(c)
+                            except SchemaError as e:
+                                # Same drift list_composers() logs and skips at line ~78,
+                                # so a single-composer fetch can't silently return malformed
+                                # JSON the list endpoint hid.
+                                print(f"Schema drift in workspace-local composer {composer_id}: {e}")
+                                continue
+                            return jsonify(local.raw)
+            except (OSError, sqlite3.Error, json.JSONDecodeError, ValueError):
                 pass
 
         # Fallback: global storage
@@ -135,10 +151,18 @@ def get_composer(composer_id):
 
                 if row and row[0]:
                     raw = row[0] if isinstance(row[0], str) else row[0].decode("utf-8")
-                    composer = json.loads(raw)
-                    composer.setdefault("conversation", [])
-                    return jsonify(composer)
-            except Exception:
+                    try:
+                        composer = Composer.from_dict(json.loads(raw), composer_id=composer_id)
+                    except SchemaError as e:
+                        # Don't return malformed JSON to the client — surface the drift
+                        # as a 404 + log, matching the silent-skip behaviour of the
+                        # list endpoints for the same row.
+                        print(f"Schema drift in composer {composer_id}: {e}")
+                        return jsonify({"error": "Composer schema drift"}), 404
+                    payload = dict(composer.raw)
+                    payload.setdefault("conversation", [])
+                    return jsonify(payload)
+            except (OSError, sqlite3.Error, json.JSONDecodeError, ValueError):
                 pass
 
         return jsonify({"error": "Composer not found"}), 404

api/search.py

@@ -18,7 +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
+from models import Bubble, Composer, SchemaError
 
 bp = Blueprint("search", __name__)
 
@@ -147,11 +147,12 @@ def search():
                     if len(parts) >= 3:
                         bid = parts[2]
                         try:
-                            b = json.loads(row["value"])
-                            if isinstance(b, dict):
-                                text = extract_text_from_bubble(b)
-                                bubble_map[bid] = {"text": text, "raw": b}
-                        except Exception:
+                            bubble = Bubble.from_dict(json.loads(row["value"]), bubble_id=bid)
+                            text = extract_text_from_bubble(bubble.raw)
+                            bubble_map[bid] = {"text": text, "raw": bubble.raw}
+                        except (SchemaError, json.JSONDecodeError, ValueError):
+                            # Skip malformed bubble rows — search must keep returning
+                            # results from the well-formed ones.
                             pass
 
                 # Search through composerData

api/workspaces.py

@@ -33,7 +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
+from models import Bubble, Composer, SchemaError, Workspace
 
 bp = Blueprint("workspaces", __name__)
 
@@ -51,11 +51,11 @@ def _get_workspace_display_name(workspace_path: str, workspace_id: str) -> str:
         return "Other chats"
     wj_path = os.path.join(workspace_path, workspace_id, "workspace.json")
     try:
-        wd = _read_json_file(wj_path)
-        name = get_workspace_display_name(wd)
+        workspace = Workspace.from_dict(_read_json_file(wj_path), workspace_id=workspace_id)
+        name = get_workspace_display_name(workspace.raw)
         if name:
             return name
-    except Exception:
+    except (SchemaError, OSError, ValueError):
         pass
     return workspace_id
 
@@ -581,10 +581,11 @@ def list_workspaces():
                         if len(parts) >= 3:
                             bid = parts[2]
                             try:
-                                b = json.loads(row["value"])
-                                if isinstance(b, dict):
-                                    bubble_map[bid] = b
-                            except Exception:
+                                bubble = Bubble.from_dict(json.loads(row["value"]), bubble_id=bid)
+                                bubble_map[bid] = bubble.raw
+                            except (SchemaError, json.JSONDecodeError, ValueError):
+                                # Skip malformed bubble rows — read-many path, one bad row
+                                # must not 500 the endpoint.
                                 pass
 
                     # Process each composer
@@ -1019,10 +1020,10 @@ def get_workspace_tabs(workspace_id):
                 if len(parts) >= 3:
                     bid = parts[2]
                     try:
-                        b = json.loads(row["value"])
-                        if isinstance(b, dict):
-                            bubble_map[bid] = b
-                    except Exception:
+                        bubble = Bubble.from_dict(json.loads(row["value"]), bubble_id=bid)
+                        bubble_map[bid] = bubble.raw
+                    except (SchemaError, json.JSONDecodeError, ValueError):
+                        # Skip malformed rows — one bad bubble must not 500 the tabs endpoint.
                         pass
 
             # Load codeBlockDiffs
@@ -1097,8 +1098,17 @@ def get_workspace_tabs(workspace_id):
             for row in composer_rows:
                 composer_id = row["key"].split(":")[1]
                 try:
-                    cd = json.loads(row["value"])
-    
+                    composer = Composer.from_dict(json.loads(row["value"]), composer_id=composer_id)
+                except SchemaError as e:
+                    # Skip the same drift list_workspaces() drops at line 605 so the two
+                    # primary conversation paths agree on what counts as a valid composer.
+                    print(f"Schema drift in composer {composer_id}: {e}")
+                    continue
+                except (json.JSONDecodeError, TypeError, ValueError):
+                    continue
+                try:
+                    cd = composer.raw
+
                     # Determine project
                     pid = _determine_project_for_conversation(
                         cd, composer_id, project_layouts_map,
@@ -1109,11 +1119,11 @@ def get_workspace_tabs(workspace_id):
                     if not pid and mapped_ws in invalid_workspace_ids:
                         pid = invalid_workspace_aliases.get(mapped_ws)
                     assigned = pid if pid else "global"
-    
+
                     if assigned not in matching_ws_ids:
                         continue
-    
-                    headers = cd.get("fullConversationHeadersOnly") or []
+
+                    headers = composer.full_conversation_headers_only
 
                     # Build bubbles
                     bubbles = []

models/conversation.py

@@ -38,6 +38,10 @@ def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer":
             raise SchemaError("Composer", "createdAt")
 
         created_at = raw.get("createdAt")
+        # Numeric-only on purpose: a 2026-05 scan of 17/17 live composers on
+        # disk stored createdAt as int milliseconds. If Cursor ever switches
+        # to ISO strings, those rows would disappear from list/search via a
+        # drift warning — relax the check at that point, don't silently coerce.
         if not isinstance(created_at, (int, float)) or isinstance(created_at, bool):
             raise SchemaError(
                 "Composer",

models/errors.py

@@ -8,7 +8,12 @@ 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}'"
+        # Distinguish "absent" from "present-but-wrong-shape" so log grepping can
+        # tell missing-key drift apart from type-mismatch drift. Hint is only
+        # populated for shape mismatches (e.g. "expected list, got dict"), so its
+        # presence is the signal.
         if hint:
-            message = f"{message} ({hint})"
+            message = f"{model}: invalid field '{field}' ({hint})"
+        else:
+            message = f"{model}: missing required field '{field}'"
         super().__init__(message)

scripts/export.py

@@ -38,6 +38,7 @@
     aggregate_session_stats,
 )
 from utils.cursor_md_exporter import cursor_cli_session_to_markdown
+from models import ExportEntry, SchemaError
 
 _logger = logging.getLogger(__name__)
 
@@ -62,13 +63,13 @@ def _load_manifest_entries(manifest_path: str) -> dict:
                 if not line:
                     continue
                 try:
-                    entry = json.loads(line)
-                    log_id = entry.get("log_id")
-                    if log_id:
-                        existing[log_id] = entry
-                except Exception as e:
-                    _logger.debug("Skipping malformed manifest line in %s: %s", manifest_path, e)
-    except Exception as e:
+                    entry = ExportEntry.from_dict(json.loads(line))
+                    existing[entry.log_id] = entry.raw
+                except (SchemaError, json.JSONDecodeError, ValueError) as e:
+                    # Pre-PR-30 manifests lack title/workspace — skip them so the
+                    # next export rebuilds the entry under the new schema.
+                    _logger.debug("Skipping manifest line in %s: %s", manifest_path, e)
+    except OSError as e:
         _logger.debug("Failed to read manifest %s: %s", manifest_path, e)
     return existing
 
@@ -818,7 +819,8 @@ def assign_workspace(cd, cid):
 
         rel_path = os.path.join(today, ws_slug, "chat", filename)
         exported.append({"id": composer_id, "rel_path": rel_path, "content": md,
-                         "out_path": out_path, "updatedAt": updated_at})
+                         "out_path": out_path, "updatedAt": updated_at,
+                         "title": title, "workspace": ws_display_name})
         count += 1
 
     # --- Cursor CLI sessions ---
@@ -915,6 +917,8 @@ def assign_workspace(cd, cid):
                 "content": md,
                 "out_path": out_path,
                 "updatedAt": updated_ms,
+                "title": title,
+                "workspace": ws_name,
             })
             count += 1
 
@@ -947,6 +951,8 @@ def assign_workspace(cd, cid):
         for e in exported:
             existing[e["id"]] = {
                 "log_id": e["id"],
+                "title": e["title"],
+                "workspace": e["workspace"],
                 "path": os.path.relpath(e["out_path"], out_dir),
                 "updated_at": datetime.fromtimestamp(e["updatedAt"] / 1000).isoformat() if e["updatedAt"] else datetime.now().isoformat(),
             }
@@ -960,6 +966,8 @@ def assign_workspace(cd, cid):
         for e in exported:
             global_existing[e["id"]] = {
                 "log_id": e["id"],
+                "title": e["title"],
+                "workspace": e["workspace"],
                 "path": e["out_path"],
                 "updated_at": datetime.fromtimestamp(e["updatedAt"] / 1000).isoformat() if e["updatedAt"] else datetime.now().isoformat(),
             }

tests/test_models.py

@@ -191,6 +191,22 @@ def test_schema_error_inherits_value_error(self) -> None:
             return
         self.fail("SchemaError did not propagate as ValueError")
 
+    def test_schema_error_message_distinguishes_missing_from_invalid(self) -> None:
+        # Operators grep logs for drift. "missing required field" must only fire
+        # when a key is absent; type / shape mismatches must read "invalid field"
+        # so the two failure modes can be told apart.
+        missing = SchemaError("Composer", "createdAt")
+        self.assertIn("missing required field", str(missing))
+        self.assertNotIn("invalid field", str(missing))
+
+        type_mismatch = SchemaError(
+            "Composer", "createdAt",
+            hint="expected timestamp number, got str",
+        )
+        self.assertIn("invalid field", str(type_mismatch))
+        self.assertNotIn("missing required field", str(type_mismatch))
+        self.assertIn("expected timestamp number, got str", str(type_mismatch))
+
     def test_non_dict_payload_raises_schema_error(self) -> None:
         bad_payloads: tuple[object, ...] = ([], "not a dict", 42, None, ("a", "b"))
         for bad in bad_payloads: