fix(persistence): enforce PRAGMA integrity_check, wipe corrupt bucket database on preflight, and auto-fallback to cloud vault

entrypoint.sh

@@ -166,32 +166,24 @@ preflight_persistence_gate() {
     fi
     rm -f "$_PROBE" 2>/dev/null || true
 
-    # Cloud Vault Restore: check private HF dataset if local database is missing
-    if [ ! -f "$DATA_DIR/storage.sqlite" ] || [ ! -s "$DATA_DIR/storage.sqlite" ]; then
-        if [ -f "/vault_sync.py" ]; then
-            log_info "PREFLIGHT" "Local storage.sqlite absent; checking cloud vault for existing database snapshot..."
-            python3 /vault_sync.py restore || true
-        fi
-    fi
-
-    # SQLite integrity check if database already exists
-    if [ -f "$DATA_DIR/storage.sqlite" ] && [ -s "$DATA_DIR/storage.sqlite" ]; then
-        if command -v sqlite3 >/dev/null 2>&1; then
-            _CHK=$(sqlite3 "$DATA_DIR/storage.sqlite" "PRAGMA quick_check;" 2>/dev/null || echo "failed")
-            if [ "$_CHK" = "ok" ]; then
-                log_info "PREFLIGHT" "Existing storage.sqlite integrity verified: PRAGMA quick_check passed."
-            else
-                log_warn "PREFLIGHT" "storage.sqlite failed quick_check: $_CHK"
-                mv "$DATA_DIR/storage.sqlite" "$DATA_DIR/storage.sqlite.corrupt-$(date +%s)" 2>/dev/null || true
-                rm -f "$DATA_DIR/storage.sqlite-wal" "$DATA_DIR/storage.sqlite-shm" 2>/dev/null || true
-                if [ -f "/vault_sync.py" ]; then
-                    log_warn "PREFLIGHT" "Attempting cloud vault restore as fallback..."
-                    python3 /vault_sync.py restore --force || true
-                fi
+    # SQLite Database Health Invariant & Cloud Vault Sync
+    if [ -f "$DATA_DIR/storage.sqlite" ]; then
+        _CHK=$(sqlite3 "$DATA_DIR/storage.sqlite" "PRAGMA integrity_check;" 2>/dev/null || echo "failed")
+        if [ "$_CHK" = "ok" ]; then
+            log_info "PREFLIGHT" "Existing storage.sqlite verified 100% healthy: PRAGMA integrity_check passed."
+        else
+            log_warn "PREFLIGHT" "storage.sqlite failed PRAGMA integrity_check: $_CHK"
+            log_warn "PREFLIGHT" "Wiping corrupt volume DB and fetching verified cloud vault snapshot..."
+            rm -rf "$DATA_DIR/storage.sqlite"* "$DATA_DIR/backups/last-known-good.sqlite"* 2>/dev/null || true
+            if [ -f "/vault_sync.py" ]; then
+                python3 /vault_sync.py restore --force || true
             fi
         fi
     else
-        log_info "PREFLIGHT" "Fresh start: storage.sqlite will be created directly by OmniRoute in $DATA_DIR"
+        log_info "PREFLIGHT" "Local storage.sqlite absent; fetching verified snapshot from cloud vault..."
+        if [ -f "/vault_sync.py" ]; then
+            python3 /vault_sync.py restore || true
+        fi
     fi
 
     log_info "PREFLIGHT" "Persistence preflight passed successfully. Single source of truth: $DATA_DIR"

persistence_manager.py

@@ -74,29 +74,13 @@ def sqlite_status(path: Path, deep: bool = False) -> dict:
         quick = connection.execute("PRAGMA quick_check").fetchone()
         result["quick_check"] = quick[0] if quick else "empty result"
         if result["quick_check"] != "ok":
-            # Attempt auto-reindex for index discrepancies
-            if "index" in str(result["quick_check"]).lower():
-                try:
-                    connection.close()
-                    connection = None
-                    rep = sqlite3.connect(str(Path(path).resolve()), timeout=5)
-                    rep.execute("REINDEX;")
-                    rep.commit()
-                    rep.close()
-                    connection = sqlite3.connect(str(Path(path).resolve()), timeout=5)
-                    quick = connection.execute("PRAGMA quick_check").fetchone()
-                    result["quick_check"] = quick[0] if quick else "empty result"
-                except Exception:
-                    pass
-            if result["quick_check"] != "ok":
-                result["status"] = "CORRUPT"
-                return result
-        if deep:
-            integrity = connection.execute("PRAGMA integrity_check").fetchone()
-            result["integrity_check"] = integrity[0] if integrity else "empty result"
-            if result["integrity_check"] != "ok":
-                result["status"] = "CORRUPT"
-                return result
+            result["status"] = "CORRUPT"
+            return result
+        integrity = connection.execute("PRAGMA integrity_check").fetchone()
+        result["integrity_check"] = integrity[0] if integrity else "empty result"
+        if result["integrity_check"] != "ok":
+            result["status"] = "CORRUPT"
+            return result
         result["journal_mode"] = connection.execute("PRAGMA journal_mode").fetchone()[0]
         result["page_count"] = connection.execute("PRAGMA page_count").fetchone()[0]
         result["status"] = "HEALTHY"
@@ -215,6 +199,19 @@ def recover() -> dict:
             _sync_dir(RUNTIME_DB.parent)
             log.info("Recovered runtime DB from verified snapshot %s", candidate)
             return {"action": "restored", "source": str(candidate), "status": "HEALTHY"}
+        # Try cloud vault restore if all local snapshots failed
+        log.warning("No verified local snapshot exists; attempting emergency cloud vault restore...")
+        try:
+            sys.path.insert(0, "/")
+            import vault_sync
+            vault_sync.restore_from_vault(force=True)
+            if PERSISTENT_DB.exists() and sqlite_status(PERSISTENT_DB)["status"] == "HEALTHY":
+                sqlite_backup(PERSISTENT_DB, RUNTIME_DB)
+                log.info("Successfully recovered runtime DB from cloud vault snapshot")
+                return {"action": "vault_restored", "source": str(PERSISTENT_DB), "status": "HEALTHY"}
+        except Exception as e:
+            log.error("Emergency cloud vault restore failed: %s", e)
+
         if current["status"] == "MISSING":
             sqlite3.connect(RUNTIME_DB).close()
             log.warning("No verified snapshot exists; initialized an empty runtime database")