fix(db): generalize schema-drift detection to all M3 tables (slabs, slab_photos, sync_log)

Live host had stale M2 slabs table (no deleted_at/server_rev) -> GET /slabs 500.
Drift check now verifies sentinel (table,column) pairs across settings/slabs/
slab_photos/sync_log; on drift drops all app-owned M3 tables and rebuilds, stamps
schema_version. Drift test now seeds M2 settings + slabs and asserts both rebuilt.
pytest 23 passed; gates exit 0.

backend/db/__init__.py

@@ -49,13 +49,34 @@ def _has_schema_version(conn: sqlite3.Connection) -> bool:
     except sqlite3.OperationalError:
         return False
 
-def _has_settings_value_column(conn: sqlite3.Connection) -> bool:
-    try:
-        info = conn.execute("PRAGMA table_info(settings)").fetchall()
-        columns = {row[1]: row[2] for row in info}
-        return 'value' in columns
-    except sqlite3.OperationalError:
-        return False
+# (table, column) pairs that must exist for the DB to be a valid M3 schema.
+# A pre-M3 (M2) volume lacks these, so any missing sentinel signals drift.
+_SCHEMA_SENTINELS = [
+    ("settings", "value"),
+    ("slabs", "deleted_at"),
+    ("slabs", "server_rev"),
+    ("slab_photos", "original_blob"),
+    ("sync_log", "payload_hash"),
+]
+
+
+def _schema_matches(conn: sqlite3.Connection) -> bool:
+    for table, column in _SCHEMA_SENTINELS:
+        try:
+            cols = {r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()}
+        except sqlite3.OperationalError:
+            return False
+        if column not in cols:
+            return False
+    return True
+
+
+# Tables the app owns and that must be rebuilt together on drift.
+_M3_TABLES = [
+    "slabs", "slab_photos", "pricing_rules", "settings", "woo_categories",
+    "woo_attributes", "woo_attribute_terms", "woo_tags", "sync_log",
+    "inference_log", "prompts", "prompt_revisions",
+]
 
 
 def init_db(db_path: str | None = None) -> sqlite3.Connection:
@@ -69,15 +90,13 @@ def init_db(db_path: str | None = None) -> sqlite3.Connection:
         conn.executescript(schema)
         conn.commit()
     if not _has_schema_version(conn):
-        if not _has_settings_value_column(conn):
+        if not _schema_matches(conn):
             logger = logging.getLogger("slab.db")
-            logger.info("Detected pre-M3 DB, rebuilding schema")
-            conn.executescript("DROP TABLE IF EXISTS settings")
+            logger.info("Detected pre-M3 schema drift, rebuilding app tables")
+            for t in _M3_TABLES:
+                conn.execute(f"DROP TABLE IF EXISTS {t}")
             conn.executescript(schema)
-            conn.execute("INSERT OR REPLACE INTO schema_version (version) VALUES (1)")
-        else:
-            # M3-era tables present but no version stamped; stamp it.
-            conn.execute("INSERT OR REPLACE INTO schema_version (version) VALUES (1)")
+        conn.execute("INSERT OR REPLACE INTO schema_version (version) VALUES (1)")
         conn.commit()
 
     return conn

backend/tests/test_schema_drift.py

@@ -25,8 +25,12 @@ def _setup_temp_db_with_m2_schema(tmp_dir: str) -> str:
     # M2-era settings shape: NO `value` column (this is the drift signature).
     # Use a raw connection (DELETE journal) exactly as an old M2 db would be on disk.
     conn = sqlite3.connect(db_path)
+    # M2-era settings: NO `value` column (drift signature).
     conn.execute("CREATE TABLE settings (key TEXT PRIMARY KEY, value_enc BLOB, updated_at TEXT)")
     conn.execute("INSERT INTO settings (key, value_enc, updated_at) VALUES ('woo_wp_username', X'00', '2026-01-01')")
+    # M2-era slabs: NO deleted_at / server_rev (drift signature).
+    conn.execute("CREATE TABLE slabs (id TEXT PRIMARY KEY, sku TEXT, status TEXT)")
+    conn.execute("INSERT INTO slabs (id, sku, status) VALUES ('old', 'OLD-1', 'calibrated')")
     conn.commit()
     conn.close()
     # Force any journal sidecars to merge before the WAL-mode factory opens the file.
@@ -50,6 +54,9 @@ def test_drift_detection_and_rebuild(monkeypatch, drift_db_dir):
     cols = {row[1] for row in info}
     assert "value" in cols
     assert "value_enc" in cols
+    # slabs must also be rebuilt to the M3 shape (deleted_at present)
+    slab_cols = {row[1] for row in conn.execute("PRAGMA table_info(slabs)").fetchall()}
+    assert "deleted_at" in slab_cols and "server_rev" in slab_cols
     # M2 rows are disposable (M3 supersedes M2 on this dev volume)
-    rows = conn.execute("SELECT key FROM settings").fetchall()
-    assert rows == []
+    assert conn.execute("SELECT key FROM settings").fetchall() == []
+    assert conn.execute("SELECT id FROM slabs").fetchall() == []