Merge origin/master into feat/split-api-workspaces-25

Integrates PR #30 (typed models + schema validation) with PR #31's
services-split structure. Conflicts in api/composers.py and
api/workspaces.py because both PRs touched the same read paths from
different angles: #30 added Bubble.from_dict / Composer.from_dict /
Workspace.from_dict gates inline in api/, #31 extracted those read
paths into services/ before #30's gates landed.

Resolution:

* api/composers.py — took master's version wholesale. Master already
  has both #31's simplification (uses utils.workspace_descriptor._read_json_file
  instead of an inline copy) AND #30's typed-model gates. Nothing to
  merge structurally; the file is what #30 produced over #31's earlier
  cleanup.

* api/workspaces.py — kept HEAD's 146-line services-split structure
  (the whole point of #31). Added two re-exports for the test harness:
  `from models import Bubble, Composer, Workspace` so that #30's
  spy-based regression tests in test_models_wired_at_read_sites.py
  can resolve `workspaces_mod.Bubble.from_dict` etc.; and
  `_get_workspace_display_name` re-exported from services.workspace_resolver
  so test_workspace_display_name_calls_workspace_from_dict still works.

Services rewiring (so #30's contract holds after the #31 structural
split — the from_dict gates have to live wherever the read paths
actually run):

* services/workspace_tabs.py:
  - bubble loader now goes through `Bubble.from_dict(...)` with
    `except SchemaError -> print drift line + skip`,
    `except (JSONDecodeError, ValueError) -> silent skip`. Matches
    the pattern api/workspaces.py used to have inline.
  - composer loader goes through `Composer.from_dict(...)` with the
    same try/except split.
  - renamed local `bubble = Bubble.from_dict(...)` to `bubble_obj`
    because the same function scope later has `bubble = bubble_map.get(...)`
    pulling raw dicts back out, and the shadowing tripped mypy.

* services/workspace_resolver.py:
  - `_get_workspace_display_name` now validates the workspace.json
    payload via `Workspace.from_dict(...)` with narrow
    `(SchemaError, OSError, ValueError)` catch instead of bare
    `except Exception`. Same shape as the api/workspaces.py inline
    version master has today.

Verified locally on the merged tree:
  - mypy --ignore-missing-imports --no-strict-optional . → Success
  - ruff check api/ utils/ scripts/export.py app.py services/ → All checks passed
  - unittest discover tests → 253 / 253 OK (including all 11 typed-model
    wiring regression tests from #30 — Bubble.from_dict / Composer.from_dict
    / Workspace.from_dict / WorkspaceLocalComposer.from_dict are now
    verifiably called from the services-side read paths)

Author: timon0305

api/composers.py

@@ -14,12 +14,17 @@
 
 from utils.workspace_path import resolve_workspace_path
 from utils.path_helpers import to_epoch_ms
-from utils.workspace_descriptor import _read_json_file
+from models import Composer, SchemaError, Workspace, WorkspaceLocalComposer
 
 bp = Blueprint("composers", __name__)
 _logger = logging.getLogger(__name__)
 
 
+def _read_json_file(path: str):
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
 @bp.route("/api/composers")
 def list_composers():
     try:
@@ -38,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:
@@ -52,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")
@@ -93,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
@@ -112,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

@@ -21,15 +21,23 @@
     _infer_workspace_name_from_context,
     # Re-exported for back-compat with existing tests that import from api.workspaces
     # directly (test_invalid_workspace_aliases, test_workspace_assignment_fallback,
-    # test_workspace_name_inference).  Production callers should import from
-    # services.workspace_resolver instead.
+    # test_workspace_name_inference, test_models_wired_at_read_sites).
+    # Production callers should import from services.workspace_resolver instead.
     _determine_project_for_conversation,  # noqa: F401
     _infer_invalid_workspace_aliases,  # noqa: F401
+    _get_workspace_display_name,  # noqa: F401
 )
 from services.cli_tabs import _get_cli_workspace_tabs
 from services.workspace_listing import list_workspace_projects
 from services.workspace_tabs import assemble_workspace_tabs
 
+# Re-exported for tests/test_models_wired_at_read_sites.py — the typed-model
+# spy harness patches `workspaces_mod.Bubble` / `.Composer` / `.Workspace` to
+# verify that production read paths actually call from_dict. The classes
+# themselves are wired inside the services modules now (post-#25 split);
+# importing them here keeps the spy resolution stable.
+from models import Bubble, Composer, Workspace  # noqa: F401
+
 bp = Blueprint("workspaces", __name__)
 _logger = logging.getLogger(__name__)
 

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,
+        )