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

* 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).

* review: address CodeRabbit feedback on PR #30 — tighten schema gates

Six fixes from the CodeRabbit pass on the initial typed-models commit,
plus matching regression tests so each can't silently regress later.

1. api/composers.py: validate the top-level JSON payload is a dict before
   key/list checks. A non-object decoded value previously bypassed the
   SchemaError path and was swallowed by the surrounding except.

2. models/{cli_session,conversation,workspace,export}.py: same isinstance
   guard at the top of every from_dict, so a non-dict raw (list, str, None)
   surfaces as SchemaError instead of AttributeError.

3. models/conversation.py: drop the `raw.get(...) or []` fallback on
   fullConversationHeadersOnly. The `or []` idiom coerced None/"" / 0
   into [] and made the isinstance gate unreachable.

4. models/export.py: replace str(raw[...]) coercion of required string
   fields with an isinstance(value, str) gate. Coercion masked drift —
   an int log_id would silently turn into a string.

5. models/conversation.py: re-gate createdAt as required per issue #24.
   A live workspaceStorage scan confirmed 17/17 composers carry it,
   so a missing value is real schema drift, not benign older-record
   absence. Two existing test fixtures in test_search_exclusion_filtering
   that lacked createdAt updated; the prior tolerated test replaced with
   a spec-correct missing-raises test.

6. utils/cli_chat_reader.py: broaden the traverse_blobs catch to
   (SchemaError, ValueError, UnicodeDecodeError, TypeError). Malformed
   hex, non-UTF-8 bytes, and invalid JSON in the meta blob previously
   escaped to the caller; now all four routes return [] uniformly.

Regression tests added:
  - tests/test_models.py: non_dict_payload (5 models × 5 bad types),
    headers_falsy_non_list, headers_empty_list_is_valid,
    export_entry_non_string_required, missing_created_at_raises
  - tests/test_cli_chat_reader.py: malformed_hex, non_utf8, invalid_json,
    non_dict meta payloads each return [] via traverse_blobs

201 tests pass (was 197); mypy clean on models/; real Cursor data
(17 composers) still loads with zero Schema drift warnings.

* 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.

* review: log non-schema read failures in list_composers (#30)

CodeRabbit flagged that the bare ``except Exception: pass`` adjacent to
the new SchemaError path swallowed decode / SQLite / non-schema errors
silently and returned a partial composer list with no signal. Replace
with a print that surfaces the exception and ``db_path``, matching the
``Schema drift in ...`` line a few lines up. No behaviour change beyond
observability — the loop still continues to the next workspace.

* docs: drop module docstrings and trim class docstrings to one-liners (#30)

Internal cleanup on the typed-models package — file location and class
name carry the intent; the multi-paragraph rationales were rehashing the
PR description. Function bodies are unchanged.

  - models/__init__.py:        removed 8-line module docstring
  - models/errors.py:          dropped module docstring + SchemaError
                               class docstring trimmed to one line
  - models/conversation.py:    dropped module docstring; Composer,
                               WorkspaceLocalComposer, Bubble docstrings
                               trimmed to one line each
  - models/cli_session.py:     dropped module docstring; CliSessionMeta
                               docstring trimmed
  - models/workspace.py:       dropped module docstring; Workspace
                               docstring trimmed
  - models/export.py:          dropped module docstring; ExportEntry
                               docstring trimmed
  - tests/test_models.py:      dropped module docstring; removed
                               now-unused ``from pathlib import Path``

204 tests pass.

* docs: drop task-reference comments and section banners (#30)

Per the review nit: stop narrating in code. Removes:

  - tests/test_models.py: 3 ``# --- section ---`` banners and 10
    inline comments referencing issue #24 / CodeRabbit findings.
  - tests/test_cli_chat_reader.py: 4 ``"""Regression: ..."""``
    docstrings and 4 line-comments restating what the test exercises.
  - utils/cli_chat_reader.py: 5-line narration on the broadened
    except arm — the tuple of exception types is the documentation.

Test names and code already describe the WHAT; the WHY for these cases
is in the PR description and the commit history, not the source.

204 tests still pass.

* review: tighten ID + createdAt type gates across the model layer (#30)

Three related CodeRabbit findings, addressed together:

  - Composer.composer_id, Bubble.bubble_id, Workspace.workspace_id now
    require ``isinstance(..., str) and value != ""``. Truthiness alone
    let non-string IDs (e.g. ``123``) slip past the boundary; aligned
    with the pattern already in WorkspaceLocalComposer.

  - Composer.createdAt is now type-gated. Presence was already enforced
    but any value type (e.g. ``"2026-05-12"``) silently became the
    timestamp. Required values must be ``int`` or ``float`` and explicitly
    not ``bool`` (since ``bool`` is a subclass of ``int`` in Python and
    a boolean is never a real timestamp).

Regression tests pin each new gate:

  - test_non_string_composer_id_raises
  - test_non_string_bubble_id_raises
  - test_non_string_workspace_id_raises
  - test_non_numeric_created_at_raises
  - test_numeric_created_at_passes

209 tests pass (was 204; +5 new).

* 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

* review: log schema drift + harden get_composer envelope (#30)

CodeRabbit follow-ups to the round-2 wiring pass:

1. The three bubble-loading loops (api/workspaces.py:583, 1025;
   api/search.py:149) caught SchemaError, json.JSONDecodeError and
   ValueError together and dropped the row silently. That broke the
   read-boundary contract — a drifted bubble would simply vanish with
   no operator signal. Split the catches: SchemaError now logs
   `Schema drift in bubble <bid>: <reason>`; JSONDecodeError /
   ValueError still pass silently because that's a parser-noise issue,
   not a schema-contract issue.

2. get_composer() at api/composers.py:125 trusted that the per-workspace
   blob was a dict with a list-valued allComposers. A drifted local row
   like `[]` or `"…"` raised AttributeError on data.get(...) and turned
   schema drift into a 500. Mirrored the three envelope guards
   list_composers() already applies at line 60–74 (data is dict,
   allComposers present, allComposers is list) and wrapped the outer
   try with `except SchemaError` so drift logs the db_path and falls
   through to the global fallback instead of 500'ing.

3. scripts/ lacked __init__.py while every other top-level package
   directory (api/, models/, utils/, tests/) has one. The new
   `from scripts import export` calls in tests resolve at runtime via
   the sys.path insert in the test header, but mypy was reporting
   inconsistent module resolution on scripts.export — a CI typecheck
   risk. Added an empty scripts/__init__.py to match convention.

3 new regression tests in tests/test_models_wired_at_read_sites.py:
  - test_bubble_schema_drift_is_logged_not_swallowed_silently —
    seeds a malformed bubble, captures stdout, asserts the
    `Schema drift in bubble <bid>` line is present.
  - test_get_composer_handles_non_dict_envelope_via_fallback —
    replaces composer.composerData with a list payload, asserts
    the endpoint returns 200 via the global fallback (was 500
    before the envelope guards).
  - test_get_composer_handles_non_list_all_composers_via_fallback —
    same shape for non-list allComposers.

Verified locally:
  - python -m unittest discover tests: 223 passed (was 220)
  - python -m unittest tests.test_models_wired_at_read_sites: 13/13
  - mypy resolves scripts.export consistently

* review: 2 CodeRabbit findings on PR #30 — response-shape parity + alias-vote drift guard

* api/composers.py — get_composer() returned three different shapes for
  the same conceptual `conversation` field depending on which branch
  resolved the composer:
    - list_composers (line 89): `c["conversation"] = c.get("conversation") or []`
    - get_composer per-workspace (line 157):   `return jsonify(local.raw)` — raw, no normalisation
    - get_composer global fallback (line 185): `payload.setdefault("conversation", [])` — fills missing but not None
  All three sites now use the line-89 idiom, so the endpoint family
  returns an identical shape regardless of source.

* api/workspaces.py — `_infer_invalid_workspace_aliases()` was reparsing
  raw composer_rows with `json.loads` + bare `except Exception` *before*
  the per-row Composer.from_dict gates in list_workspaces() /
  get_workspace_tabs() ran. That meant schema-drifted composers could
  still cast votes in the workspace-alias inference and misassign
  otherwise-valid composers to the wrong workspace. Now gates via
  `Composer.from_dict(...)` with a narrow
  `(SchemaError, JSONDecodeError, TypeError, ValueError)` catch.

  Existing test_majority_vote_alias_selection fixture predated the
  strict createdAt gate from issue #24; added `createdAt` to each of
  the three composer rows so the fixture reflects real production shape.

  New test_drifted_composer_does_not_skew_vote pins the regression:
  seeds two well-formed boost-ws votes plus one drifted row that would
  vote for team-ws if counted, asserts the alias resolves to boost-ws.

Verified:
  - mypy --ignore-missing-imports --no-strict-optional . → Success
  - ruff check api/ utils/ scripts/export.py app.py → All checks passed
  - unittest discover tests → 224 / 224 OK (was 223 + 1 new regression test)

Author: timon0305

api/composers.py

@@ -14,6 +14,7 @@
 
 from utils.workspace_path import resolve_workspace_path
 from utils.path_helpers import to_epoch_ms
+from models import Composer, SchemaError, Workspace, WorkspaceLocalComposer
 
 bp = Blueprint("composers", __name__)
 _logger = logging.getLogger(__name__)
@@ -42,9 +43,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:
@@ -56,18 +59,44 @@ 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
-
-        composers.sort(key=lambda c: to_epoch_ms(c.get("lastUpdatedAt")), reverse=True)
-        return jsonify(composers)
+                    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:
+                            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((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 pair: to_epoch_ms(pair[0].last_updated_at), reverse=True)
+        return jsonify([c for _, c in composers])
 
     except Exception:
         _logger.exception("Failed to get composers")
@@ -97,10 +126,45 @@ 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:
+                    # Mirror the envelope guards list_composers() applies at line 60–74
+                    # so a drifted local row (data not a dict, or allComposers missing
+                    # / non-list) surfaces as a logged SchemaError, not a 500.
+                    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 not isinstance(all_composers, list):
+                        raise SchemaError(
+                            "WorkspaceComposers",
+                            "allComposers",
+                            hint=f"expected list, got {type(all_composers).__name__}",
+                        )
+                    for c in all_composers:
+                        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
+                            # Match list_composers() at line 89 and the global
+                            # fallback below: `conversation` is normalised to []
+                            # whether it's absent or None, so the response shape
+                            # is identical regardless of which branch resolved
+                            # the composer (CodeRabbit on PR #30).
+                            payload = dict(local.raw)
+                            payload["conversation"] = payload.get("conversation") or []
+                            return jsonify(payload)
+            except SchemaError as e:
+                print(f"Schema drift in {db_path}: {e}")
+            except (OSError, sqlite3.Error, json.JSONDecodeError, ValueError):
                 pass
 
         # Fallback: global storage
@@ -116,10 +180,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["conversation"] = payload.get("conversation") or []
+                    return jsonify(payload)
+            except (OSError, sqlite3.Error, json.JSONDecodeError, ValueError):
                 pass
 
         return jsonify({"error": "Composer not found"}), 404

api/search.py

@@ -19,6 +19,7 @@
 from utils.path_helpers import 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 Bubble, Composer, SchemaError
 
 bp = Blueprint("search", __name__)
 _logger = logging.getLogger(__name__)
@@ -148,11 +149,15 @@ 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 as e:
+                            # Drift logged so the operator can see why a chat dropped
+                            # out of search results; bad row still skipped so search
+                            # keeps returning results from the well-formed ones.
+                            print(f"Schema drift in bubble {bid}: {e}")
+                        except (json.JSONDecodeError, ValueError):
                             pass
 
                 # Search through composerData
@@ -163,17 +168,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
 
@@ -245,7 +257,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

@@ -35,6 +35,7 @@
 from utils.text_extract import extract_text_from_bubble, format_tool_action
 from utils.tool_parser import parse_tool_call as _parse_tool_call
 from utils.exclusion_rules import build_searchable_text, is_excluded_by_rules
+from models import Bubble, Composer, SchemaError, Workspace
 
 bp = Blueprint("workspaces", __name__)
 _logger = logging.getLogger(__name__)
@@ -53,11 +54,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
 
@@ -446,8 +447,15 @@ def _infer_invalid_workspace_aliases(
         if mapped not in invalid_workspace_ids:
             continue
         try:
-            cd = json.loads(row["value"])
-        except Exception:
+            # Validate via Composer.from_dict so a schema-drifted row can't
+            # steer the alias vote and misassign otherwise-valid composers to
+            # the wrong workspace. The downstream per-row loops in
+            # list_workspaces() / get_workspace_tabs() already drop drift,
+            # but this helper runs BEFORE that loop and its votes shape
+            # invalid_workspace_aliases for every other composer.
+            composer = Composer.from_dict(json.loads(row["value"]), composer_id=cid)
+            cd = composer.raw
+        except (SchemaError, json.JSONDecodeError, TypeError, ValueError):
             continue
         inferred = _determine_project_for_conversation(
             cd,
@@ -583,10 +591,14 @@ 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 as e:
+                                # Drift surfaces in logs so an operator sees disappearing
+                                # bubbles instead of guessing. The row is still skipped —
+                                # one bad bubble must not 500 the endpoint.
+                                print(f"Schema drift in bubble {bid}: {e}")
+                            except (json.JSONDecodeError, ValueError):
                                 pass
 
                     # Process each composer
@@ -603,9 +615,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
                             )
@@ -614,16 +632,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
@@ -1013,10 +1031,14 @@ 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 as e:
+                        # Drift surfaces in logs so an operator can chase disappearing
+                        # bubbles instead of guessing. Bad row still skipped so the
+                        # tabs endpoint can't 500 on one malformed bubble.
+                        print(f"Schema drift in bubble {bid}: {e}")
+                    except (json.JSONDecodeError, ValueError):
                         pass
 
             # Load codeBlockDiffs
@@ -1091,8 +1113,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,
@@ -1103,11 +1134,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/__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",
+]