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

Author: timon0305

api/composers.py

@@ -124,7 +124,25 @@ def get_composer(composer_id):
 
                 if row and row[0]:
                     data = json.loads(row[0])
-                    for c in (data.get("allComposers") or []):
+                    # 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)
@@ -135,6 +153,8 @@ def get_composer(composer_id):
                                 print(f"Schema drift in workspace-local composer {composer_id}: {e}")
                                 continue
                             return jsonify(local.raw)
+            except SchemaError as e:
+                print(f"Schema drift in {db_path}: {e}")
             except (OSError, sqlite3.Error, json.JSONDecodeError, ValueError):
                 pass
 

api/search.py

@@ -150,9 +150,12 @@ def search():
                             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.
+                        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

api/workspaces.py

@@ -583,9 +583,12 @@ def list_workspaces():
                             try:
                                 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.
+                            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
@@ -1022,8 +1025,12 @@ def get_workspace_tabs(workspace_id):
                     try:
                         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.
+                    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

scripts/__init__.py

@@ -128,6 +128,38 @@ def test_workspace_tabs_endpoint_calls_bubble_from_dict(self):
                     "model is defined but not wired at the production read site",
             )
 
+    def test_bubble_schema_drift_is_logged_not_swallowed_silently(self):
+        # CodeRabbit: SchemaError used to be lumped in with JSONDecodeError /
+        # ValueError and skipped silently. Schema drift must now print a
+        # `Schema drift in bubble <bid>` line so disappearing bubbles can be
+        # traced. The well-formed row still loads alongside.
+        from app import create_app
+        # Seed a deliberately-malformed bubble row that will trip
+        # Bubble.from_dict's "expected non-empty str" gate on the bubble_id by
+        # putting a non-dict at the value slot.
+        global_db = os.path.join(self._tmp.name, "globalStorage", "state.vscdb")
+        with closing(sqlite3.connect(global_db)) as conn:
+            conn.execute(
+                "INSERT INTO cursorDiskKV ([key], value) VALUES (?, ?)",
+                (f"bubbleId:{COMPOSER_ID}:bub-bad", json.dumps("not-a-dict")),
+            )
+            conn.commit()
+        app = create_app()
+        app.config["TESTING"] = True
+        app.config["EXCLUSION_RULES"] = []
+        import io
+        from contextlib import redirect_stdout
+        captured = io.StringIO()
+        with redirect_stdout(captured):
+            client = app.test_client()
+            response = client.get("/api/search?q=sentinel-wired")
+            self.assertEqual(response.status_code, 200)
+        out = captured.getvalue()
+        self.assertIn("Schema drift in bubble", out,
+            msg=f"expected drift log line, got stdout:\n{out!r}")
+        self.assertIn("bub-bad", out,
+            msg="drift log must include the offending bubble id")
+
     def test_workspace_tabs_endpoint_calls_composer_from_dict(self):
         # Brad's most-important finding: list_workspaces() at api/workspaces.py:605
         # validates each composer with Composer.from_dict, but get_workspace_tabs()
@@ -289,6 +321,47 @@ def test_get_composer_calls_workspace_local_composer_from_dict(self):
                     "schema validation that list_composers performs.",
             )
 
+    def test_get_composer_handles_non_dict_envelope_via_fallback(self):
+        # CodeRabbit: data.get(...) used to crash with AttributeError when the
+        # per-workspace blob isn't a dict. Now must be caught as SchemaError
+        # so the function falls through to the global fallback instead of 500.
+        from app import create_app
+        ws_db = os.path.join(self.workspace_path, WORKSPACE_ID, "state.vscdb")
+        with closing(sqlite3.connect(ws_db)) as conn:
+            # Replace the dict envelope with a list — would have raised
+            # AttributeError on data.get(...) before the guards landed.
+            conn.execute(
+                "UPDATE ItemTable SET value = ? WHERE [key] = 'composer.composerData'",
+                (json.dumps(["not", "a", "dict"]),),
+            )
+            conn.commit()
+        app = create_app()
+        app.config["TESTING"] = True
+        app.config["EXCLUSION_RULES"] = []
+        client = app.test_client()
+        response = client.get(f"/api/composers/{COMPOSER_ID}")
+        # Global fallback still has the composer seeded, so this must return
+        # 200, not 500. The per-workspace drift gets logged and skipped.
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.get_json().get("name"), "Wired conversation")
+
+    def test_get_composer_handles_non_list_all_composers_via_fallback(self):
+        from app import create_app
+        ws_db = os.path.join(self.workspace_path, WORKSPACE_ID, "state.vscdb")
+        with closing(sqlite3.connect(ws_db)) as conn:
+            conn.execute(
+                "UPDATE ItemTable SET value = ? WHERE [key] = 'composer.composerData'",
+                (json.dumps({"allComposers": "should-be-list"}),),
+            )
+            conn.commit()
+        app = create_app()
+        app.config["TESTING"] = True
+        app.config["EXCLUSION_RULES"] = []
+        client = app.test_client()
+        response = client.get(f"/api/composers/{COMPOSER_ID}")
+        # Drift surfaces via global fallback, not as a 500.
+        self.assertEqual(response.status_code, 200)
+
     def test_get_composer_calls_composer_from_dict_on_global_fallback(self):
         # When the per-workspace path misses (composer only in globalStorage),
         # the fallback must validate via Composer.from_dict — not just decode