refactor(models): tighten Bubble boundary with typed accessors

Author: clean6378-max-it

api/composers.py

@@ -201,7 +201,7 @@ def get_composer(composer_id: str) -> tuple[Response, int] | Response:
                             # 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 = local.cursor_storage_payload()
                             payload["conversation"] = payload.get("conversation") or []
                             return json_response(payload)
             except SchemaError as e:
@@ -240,7 +240,7 @@ def get_composer(composer_id: str) -> tuple[Response, int] | Response:
                             type(e).__name__,
                         )
                         return json_response({"error": "Composer schema drift"}, 404)
-                    payload = dict(composer.raw)
+                    payload = composer.cursor_storage_payload()
                     payload["conversation"] = payload.get("conversation") or []
                     return json_response(payload)
             except (OSError, sqlite3.Error, json.JSONDecodeError, ValueError):

models/conversation.py

@@ -4,6 +4,17 @@
 from dataclasses import dataclass, field
 from typing import Any, cast
 
+from models.conversation_types import (
+    BubbleContextDict,
+    BubbleMetadataDict,
+    ContextWindowStatusDict,
+    FileUriDict,
+    ModelInfoDict,
+    ThinkingDict,
+    TokenCountDict,
+    ToolFormerDataDict,
+    ToolResultEntry,
+)
 from models.errors import SchemaError
 from models.from_dict_validation import (
     require_dict,
@@ -26,7 +37,7 @@ class Composer:
     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)
+    _raw: dict[str, Any] = field(default_factory=dict, repr=False)
 
     @classmethod
     def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer":
@@ -79,12 +90,19 @@ def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer":
             name=raw.get("name"),
             last_updated_at=raw.get("lastUpdatedAt"),
             model_config=model_config,
-            raw=raw,
+            _raw=raw,
         )
 
+    def cursor_storage_payload(self) -> dict[str, Any]:
+        """Shallow copy of stored Cursor JSON for API passthrough.
+
+        Prefer typed accessors for field reads.
+        """
+        return dict(self._raw)
+
     @property
     def newly_created_files(self) -> list[Any]:
-        value = self.raw.get("newlyCreatedFiles")
+        value = self._raw.get("newlyCreatedFiles")
         if value is None:
             return []
         if not isinstance(value, list):
@@ -98,7 +116,7 @@ def newly_created_files(self) -> list[Any]:
 
     @property
     def code_block_data(self) -> dict[str, Any] | None:
-        value = self.raw.get("codeBlockData")
+        value = self._raw.get("codeBlockData")
         if value is None:
             return None
         if not isinstance(value, dict):
@@ -113,7 +131,7 @@ def code_block_data(self) -> dict[str, Any] | None:
     @property
     def usage_data(self) -> dict[str, Any]:
         """Composer cost rollup; empty dict when absent (common)."""
-        value = self.raw.get("usageData")
+        value = self._raw.get("usageData")
         if value is None:
             return {}
         if not isinstance(value, dict):
@@ -127,9 +145,9 @@ def usage_data(self) -> dict[str, Any]:
         return value
 
     def _optional_counter(self, key: str) -> int | float:
-        value = self.raw.get(key, 0)
+        value = self._raw.get(key, 0)
         if isinstance(value, bool) or not isinstance(value, (int, float)):
-            if key in self.raw:
+            if key in self._raw:
                 suffix = f" {self.composer_id}" if self.composer_id else ""
                 _logger.warning(
                     "Schema drift in Composer%s: invalid type for %s (expected number, got %s)",
@@ -172,7 +190,7 @@ class WorkspaceLocalComposer:
 
     composer_id: str
     last_updated_at: Any = None
-    raw: dict[str, Any] = field(default_factory=dict)
+    _raw: dict[str, Any] = field(default_factory=dict, repr=False)
 
     @classmethod
     def from_dict(cls, raw: dict[str, Any]) -> "WorkspaceLocalComposer":
@@ -194,9 +212,13 @@ def from_dict(cls, raw: dict[str, Any]) -> "WorkspaceLocalComposer":
         return cls(
             composer_id=composer_id,
             last_updated_at=raw.get("lastUpdatedAt"),
-            raw=raw,
+            _raw=raw,
         )
 
+    def cursor_storage_payload(self) -> dict[str, Any]:
+        """Shallow copy of stored Cursor JSON for API passthrough."""
+        return dict(self._raw)
+
 
 @dataclass(frozen=True)
 class Bubble:
@@ -206,7 +228,7 @@ class Bubble:
     """
 
     bubble_id: str
-    raw: dict[str, Any] = field(default_factory=dict)
+    _raw: dict[str, Any] = field(default_factory=dict, repr=False)
 
     @classmethod
     def from_dict(cls, raw: dict[str, Any], *, bubble_id: str) -> "Bubble":
@@ -224,17 +246,24 @@ def from_dict(cls, raw: dict[str, Any], *, bubble_id: str) -> "Bubble":
         """
         raw = require_dict(raw, model="Bubble", field="bubble")
         require_non_empty_str(bubble_id, model="Bubble", field="bubbleId")
-        return cls(bubble_id=bubble_id, raw=raw)
+        return cls(bubble_id=bubble_id, _raw=raw)
+
+    def cursor_storage_payload(self) -> dict[str, Any]:
+        """Shallow copy of stored Cursor JSON for API passthrough.
+
+        Prefer typed accessors for field reads.
+        """
+        return dict(self._raw)
 
     @property
     def text(self) -> str | None:
         """Plain ``text`` field; richText is handled by :func:`extract_text_from_bubble`."""
-        value = self.raw.get("text")
+        value = self._raw.get("text")
         return value if isinstance(value, str) else None
 
     @property
-    def metadata(self) -> dict[str, Any]:
-        value = self.raw.get("metadata")
+    def metadata(self) -> BubbleMetadataDict:
+        value = self._raw.get("metadata")
         if value is None:
             return {}
         if not isinstance(value, dict):
@@ -244,11 +273,11 @@ def metadata(self) -> dict[str, Any]:
                 type(value).__name__,
             )
             return {}
-        return value
+        return cast(BubbleMetadataDict, value)
 
     @property
-    def relevant_files(self) -> list[Any]:
-        value = self.raw.get("relevantFiles")
+    def relevant_files(self) -> list[str]:
+        value = self._raw.get("relevantFiles")
         if value is None:
             return []
         if not isinstance(value, list):
@@ -258,11 +287,11 @@ def relevant_files(self) -> list[Any]:
                 type(value).__name__,
             )
             return []
-        return value
+        return cast(list[str], value)
 
     @property
-    def attached_file_code_chunks_uris(self) -> list[Any]:
-        value = self.raw.get("attachedFileCodeChunksUris")
+    def attached_file_code_chunks_uris(self) -> list[FileUriDict]:
+        value = self._raw.get("attachedFileCodeChunksUris")
         if value is None:
             return []
         if not isinstance(value, list):
@@ -272,11 +301,11 @@ def attached_file_code_chunks_uris(self) -> list[Any]:
                 type(value).__name__,
             )
             return []
-        return value
+        return cast(list[FileUriDict], value)
 
     @property
-    def context(self) -> dict[str, Any]:
-        value = self.raw.get("context")
+    def context(self) -> BubbleContextDict:
+        value = self._raw.get("context")
         if value is None:
             return {}
         if not isinstance(value, dict):
@@ -286,11 +315,11 @@ def context(self) -> dict[str, Any]:
                 type(value).__name__,
             )
             return {}
-        return value
+        return cast(BubbleContextDict, value)
 
     @property
-    def token_count(self) -> dict[str, Any] | None:
-        value = self.raw.get("tokenCount")
+    def token_count(self) -> TokenCountDict | None:
+        value = self._raw.get("tokenCount")
         if value is None:
             return None
         if not isinstance(value, dict):
@@ -300,11 +329,11 @@ def token_count(self) -> dict[str, Any] | None:
                 type(value).__name__,
             )
             return None
-        return value
+        return cast(TokenCountDict, value)
 
     @property
-    def tool_former_data(self) -> dict[str, Any] | None:
-        value = self.raw.get("toolFormerData")
+    def tool_former_data(self) -> ToolFormerDataDict | None:
+        value = self._raw.get("toolFormerData")
         if value is None:
             return None
         if not isinstance(value, dict):
@@ -314,11 +343,11 @@ def tool_former_data(self) -> dict[str, Any] | None:
                 type(value).__name__,
             )
             return None
-        return value
+        return cast(ToolFormerDataDict, value)
 
     @property
-    def model_info(self) -> dict[str, Any]:
-        value = self.raw.get("modelInfo")
+    def model_info(self) -> ModelInfoDict:
+        value = self._raw.get("modelInfo")
         if value is None:
             return {}
         if not isinstance(value, dict):
@@ -328,15 +357,15 @@ def model_info(self) -> dict[str, Any]:
                 type(value).__name__,
             )
             return {}
-        return value
+        return cast(ModelInfoDict, value)
 
     @property
-    def thinking(self) -> str | dict[str, Any] | None:
-        value = self.raw.get("thinking")
+    def thinking(self) -> str | ThinkingDict | None:
+        value = self._raw.get("thinking")
         if value is None:
             return None
         if isinstance(value, (str, dict)):
-            return value
+            return cast(str | ThinkingDict, value)
         _logger.warning(
             "Schema drift in Bubble %s: invalid type for thinking (expected str or dict, got %s)",
             self.bubble_id,
@@ -346,7 +375,7 @@ def thinking(self) -> str | dict[str, Any] | None:
 
     @property
     def thinking_duration_ms(self) -> int | float | None:
-        value = self.raw.get("thinkingDurationMs")
+        value = self._raw.get("thinkingDurationMs")
         if value is None:
             return None
         if isinstance(value, bool) or not isinstance(value, (int, float)):
@@ -359,8 +388,8 @@ def thinking_duration_ms(self) -> int | float | None:
         return cast(int | float, value)
 
     @property
-    def context_window_status_at_creation(self) -> dict[str, Any]:
-        value = self.raw.get("contextWindowStatusAtCreation")
+    def context_window_status_at_creation(self) -> ContextWindowStatusDict:
+        value = self._raw.get("contextWindowStatusAtCreation")
         if value is None:
             return {}
         if not isinstance(value, dict):
@@ -370,11 +399,11 @@ def context_window_status_at_creation(self) -> dict[str, Any]:
                 type(value).__name__,
             )
             return {}
-        return value
+        return cast(ContextWindowStatusDict, value)
 
     @property
-    def tool_results(self) -> list[Any] | None:
-        value = self.raw.get("toolResults")
+    def tool_results(self) -> list[ToolResultEntry] | None:
+        value = self._raw.get("toolResults")
         if value is None:
             return None
         if not isinstance(value, list):
@@ -384,12 +413,12 @@ def tool_results(self) -> list[Any] | None:
                 type(value).__name__,
             )
             return None
-        return value
+        return cast(list[ToolResultEntry], value)
 
     def bubble_timestamp_ms(self) -> int | float | None:
         """``createdAt`` or ``timestamp`` in milliseconds when present."""
         for key in ("createdAt", "timestamp"):
-            value = self.raw.get(key)
+            value = self._raw.get(key)
             if isinstance(value, (int, float)) and not isinstance(value, bool):
                 return value
         return None

models/conversation_types.py

@@ -0,0 +1,61 @@
+"""Typed shapes for Cursor JSON fields on :class:`models.conversation.Bubble`."""
+
+from __future__ import annotations
+
+from typing import Any, TypedDict
+
+
+class FileUriDict(TypedDict, total=False):
+    """URI object on attached-file or context entries."""
+
+    path: str
+
+
+class BubbleMetadataDict(TypedDict, total=False):
+    """Storage ``metadata`` blob on a bubble row."""
+
+    modelName: str
+
+
+class TokenCountDict(TypedDict, total=False):
+    inputTokens: int
+    outputTokens: int
+    cachedTokens: int
+
+
+class ModelInfoDict(TypedDict, total=False):
+    modelName: str
+
+
+class ContextWindowStatusDict(TypedDict, total=False):
+    percentageRemainingFloat: float
+    percentageRemaining: float
+    tokensUsed: int
+    tokenLimit: int
+
+
+class FileSelectionDict(TypedDict, total=False):
+    uri: FileUriDict
+
+
+class BubbleContextDict(TypedDict, total=False):
+    fileSelections: list[FileSelectionDict]
+
+
+class ToolFormerDataDict(TypedDict, total=False):
+    name: str
+    status: str
+    params: str | dict[str, Any]
+    rawArgs: str
+    result: str
+
+
+class ThinkingDict(TypedDict, total=False):
+    text: str
+
+
+class ToolResultEntry(TypedDict, total=False):
+    """One element of a bubble ``toolResults`` list."""
+
+    toolName: str
+    result: str | dict[str, Any]