FN-7091: recover messaging index corruption on send

Repair the messaging send path when SQLite reports corrupt messages-table indexes despite quick_check passing.

- Detect SQLite corruption errors from codes, names, and messages instead of only FTS5 text.
- Add a scoped REINDEX messages repair and single retry around message inserts.
- Return actionable repair guidance when reindexing or the retry still fails.
- Cover send recovery and corruption classification with regression tests, plus a patch changeset.

Files changed:
 .changeset/fn-7091-messaging-corruption-recovery.md       |   7 ++
 packages/core/src/__tests__/db.test.ts             |  38 ++++++++
 packages/core/src/__tests__/message-store.test.ts  | 100 +++++++++++++++++++++
 packages/core/src/db.ts                            |  43 +++++++--
 packages/core/src/message-store.ts                 |  81 ++++++++++++++---
 5 files changed, 251 insertions(+), 18 deletions(-)

Fusion-Task-Id: FN-7091

Fusion-Task-Lineage: 47bcbe17-b428-4702-9835-59cb9e6ec164

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>

.changeset/fn-7091-messaging-corruption-recovery.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Recover corrupt messaging indexes during send or report the exact repair command.
+category: fix
+dev: MessageStore now runs a scoped REINDEX messages retry on SQLite corruption during send.

packages/core/src/__tests__/db.test.ts

@@ -2,6 +2,7 @@ import { describe, it, expect, beforeAll, beforeEach, afterEach, afterAll, vi }
 import {
   Database,
   createDatabase,
+  isSqliteCorruptionError,
   quickCheckSqliteFile,
   integrityCheckSqliteFileAsync,
   toJson,
@@ -255,6 +256,43 @@ describe("Database", () => {
     await cleanupTmpDirsAsync();
   });
 
+  describe("SQLite corruption recovery helpers", () => {
+    it("classifies SQLite corruption errors without matching ordinary SQLite errors", () => {
+      const corruptByCode = Object.assign(new Error("constraint failed"), { code: "SQLITE_CORRUPT" });
+
+      expect(isSqliteCorruptionError(corruptByCode)).toBe(true);
+      expect(isSqliteCorruptionError(new Error("database disk image is malformed"))).toBe(true);
+      expect(isSqliteCorruptionError(new Error("corruption found reading blob from fts5 table"))).toBe(true);
+      expect(isSqliteCorruptionError(new Error("fts5 segment is corrupt"))).toBe(true);
+      expect(isSqliteCorruptionError(Object.assign(new Error("database is locked"), { code: "SQLITE_BUSY" }))).toBe(false);
+      expect(isSqliteCorruptionError(new Error("plain application failure"))).toBe(false);
+    });
+
+    it("reindexes messages indexes for populated disk and in-memory databases", () => {
+      db.prepare(`
+        INSERT INTO messages (id, fromId, fromType, toId, toType, content, type, read, metadata, createdAt, updatedAt)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+      `).run("msg-disk", "agent-1", "agent", "user-1", "user", "hello", "agent-to-user", 0, null, "2026-06-26T00:00:00.000Z", "2026-06-26T00:00:00.000Z");
+
+      expect(() => db.reindexMessages()).not.toThrow();
+
+      const memDb = new Database(fusionDir, { inMemory: true });
+      try {
+        memDb.init();
+        memDb.prepare(`
+          INSERT INTO messages (id, fromId, fromType, toId, toType, content, type, read, metadata, createdAt, updatedAt)
+          VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        `).run("msg-memory", "agent-1", "agent", "user-1", "user", "hello", "agent-to-user", 0, null, "2026-06-26T00:00:00.000Z", "2026-06-26T00:00:00.000Z");
+
+        expect(() => memDb.reindexMessages()).not.toThrow();
+        memDb.close();
+        expect(() => memDb.reindexMessages()).not.toThrow();
+      } finally {
+        memDb.close();
+      }
+    });
+  });
+
   describe("initialization", () => {
     it("creates the database file", () => {
       expect(existsSync(join(fusionDir, "fusion.db"))).toBe(true);

packages/core/src/__tests__/message-store.test.ts

@@ -7,6 +7,10 @@ import { MessageStore } from "../message-store.js";
 import { DASHBOARD_USER_ID } from "../types.js";
 import type { Message, Mailbox } from "../types.js";
 
+function makeSqliteCorruptError(): Error & { code: string } {
+  return Object.assign(new Error("database disk image is malformed"), { code: "SQLITE_CORRUPT" });
+}
+
 describe("MessageStore", () => {
   let store: MessageStore;
   let db: Database;
@@ -21,6 +25,7 @@ describe("MessageStore", () => {
   });
 
   afterEach(() => {
+    vi.restoreAllMocks();
     db.close();
     try {
       rmSync(tempDir, { recursive: true, force: true });
@@ -150,6 +155,101 @@ describe("MessageStore", () => {
       }).toThrow("metadata.replyTo.messageId must be a non-empty string");
     });
 
+    it("reindexes messages indexes once and retries when an insert reports SQLite corruption", () => {
+      const privateStore = store as unknown as { stmtInsert: { run: (...args: unknown[]) => unknown } };
+      const originalInsertRun = privateStore.stmtInsert.run.bind(privateStore.stmtInsert);
+      let attempts = 0;
+      privateStore.stmtInsert = {
+        run: vi.fn((...args: unknown[]) => {
+          attempts += 1;
+          if (attempts === 1) {
+            throw makeSqliteCorruptError();
+          }
+          return originalInsertRun(...args);
+        }),
+      };
+      const reindexMessages = vi.spyOn(db, "reindexMessages");
+
+      const message = store.sendMessage({
+        fromId: "agent-1",
+        fromType: "agent",
+        toId: "user-1",
+        toType: "user",
+        content: "Recovered send",
+        type: "agent-to-user",
+      });
+
+      expect(reindexMessages).toHaveBeenCalledTimes(1);
+      expect(attempts).toBe(2);
+      expect(message.id).toMatch(/^msg-/);
+      expect(store.getMessage(message.id)).toEqual(message);
+    });
+
+    it("throws a repair-specific remediation error when REINDEX itself reports corruption", () => {
+      const privateStore = store as unknown as { stmtInsert: { run: (...args: unknown[]) => unknown } };
+      privateStore.stmtInsert = {
+        run: vi.fn(() => {
+          throw makeSqliteCorruptError();
+        }),
+      };
+      const reindexMessages = vi.spyOn(db, "reindexMessages").mockImplementation(() => {
+        throw makeSqliteCorruptError();
+      });
+
+      let thrown: unknown;
+      try {
+        store.sendMessage({
+          fromId: "agent-1",
+          fromType: "agent",
+          toId: "user-1",
+          toType: "user",
+          content: "Reindex fails",
+          type: "agent-to-user",
+        });
+      } catch (error) {
+        thrown = error;
+      }
+
+      expect(reindexMessages).toHaveBeenCalledTimes(1);
+      expect(thrown).toBeInstanceOf(Error);
+      const message = (thrown as Error).message;
+      expect(message).toContain("Messages store index repair failed (table=messages, db=:memory:)");
+      expect(message).toContain('run "fn db --vacuum" and inspect with "PRAGMA integrity_check"');
+      expect(message).not.toBe("database disk image is malformed");
+    });
+
+    it("throws a table/database remediation error when corruption persists after successful reindex", () => {
+      const privateStore = store as unknown as { stmtInsert: { run: (...args: unknown[]) => unknown } };
+      privateStore.stmtInsert = {
+        run: vi.fn(() => {
+          throw makeSqliteCorruptError();
+        }),
+      };
+      const reindexMessages = vi.spyOn(db, "reindexMessages");
+
+      let thrown: unknown;
+      try {
+        store.sendMessage({
+          fromId: "agent-1",
+          fromType: "agent",
+          toId: "user-1",
+          toType: "user",
+          content: "Still corrupt",
+          type: "agent-to-user",
+        });
+      } catch (error) {
+        thrown = error;
+      }
+
+      expect(reindexMessages).toHaveBeenCalledTimes(1);
+      expect(thrown).toBeInstanceOf(Error);
+      const message = (thrown as Error).message;
+      expect(message).toContain("Messages store table/database corruption after REINDEX (table=messages, db=:memory:)");
+      expect(message).toContain('run "fn db --vacuum" and inspect with "PRAGMA integrity_check"');
+      expect(message).not.toContain('run "REINDEX messages" or "fn db --vacuum" to repair');
+      expect(message).not.toBe("database disk image is malformed");
+    });
+
     it("returns null for non-existent message", () => {
       const result = store.getMessage("msg-nonexistent");
       expect(result).toBeNull();

packages/core/src/db.ts

@@ -152,17 +152,33 @@ export function probeFts5(db: DatabaseSync): boolean {
   }
 }
 
+function getSqliteErrorText(error: unknown): string {
+  const err = error as { message?: unknown; code?: unknown; name?: unknown } | null | undefined;
+  return [err?.code, err?.name, err?.message, error]
+    .map((part) => (typeof part === "string" ? part : ""))
+    .filter(Boolean)
+    .join(" ")
+    .toLowerCase();
+}
+
+/**
+ * Check whether an error appears to be a SQLite corruption/integrity failure.
+ */
+export function isSqliteCorruptionError(error: unknown): boolean {
+  const text = getSqliteErrorText(error);
+  return (
+    text.includes("sqlite_corrupt") ||
+    text.includes("database disk image is malformed") ||
+    text.includes("corruption found reading blob") ||
+    (text.includes("fts5") && text.includes("corrupt"))
+  );
+}
+
 /**
  * Check whether an error appears to be an FTS5 corruption/integrity failure.
  */
 export function isFts5CorruptionError(error: unknown): boolean {
-  const message = error instanceof Error ? error.message : String(error ?? "");
+  return isSqliteCorruptionError(error);
-  const lower = message.toLowerCase();
-  return (
-    lower.includes("corruption found reading blob") ||
-    lower.includes("database disk image is malformed") ||
-    (lower.includes("fts5") && lower.includes("corrupt"))
-  );
 }
 
 // ── Schema Definition ────────────────────────────────────────────────
@@ -5546,6 +5562,19 @@ export class Database {
     return isFts5CorruptionError(error);
   }
 
+  /**
+   * Rebuild every SQLite index attached to the messages table.
+   *
+   * FNXC:Database 2026-06-26-00:00:
+   * `PRAGMA quick_check` can report ok while a messages-table index is out of sync with its table; INSERT/UPDATE/DELETE then fail because SQLite must maintain idxMessagesTo, idxMessagesFrom, and idxMessagesCreatedAt. A scoped `REINDEX messages` repairs only the messaging lookup indexes without invoking whole-file recovery from the send path.
+   */
+  reindexMessages(): void {
+    if (this.closed) {
+      return;
+    }
+    this.db.exec("REINDEX messages");
+  }
+
   /**
    * Read the declared columns for a table.
    */

packages/core/src/message-store.ts

@@ -13,7 +13,7 @@
 import { EventEmitter } from "node:events";
 import { randomUUID } from "node:crypto";
 import type { Database } from "./db.js";
-import { fromJson, toJsonNullable } from "./db.js";
+import { fromJson, isSqliteCorruptionError, toJsonNullable } from "./db.js";
 import { createLogger } from "./logger.js";
 import { DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, type Message, type MessageCreateInput, type MessageFilter, type MessageType, type Mailbox, type ParticipantType } from "./types.js";
 
@@ -151,6 +151,57 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
       updatedAt: now,
     };
 
+    this.runInsertWithMessagesIndexRecovery(message);
+
+    this.db.bumpLastModified();
+    messageStoreLog.log(`MessageStore emitting message:sent id=${message.id} type=${message.type} fromId=${message.fromId} toId=${message.toId}`);
+    this.emit("message:sent", message);
+    this.emit("message:received", message);
+
+    if (message.toType === "agent" && this.onMessageToAgent) {
+      this.onMessageToAgent(message);
+    }
+
+    return message;
+  }
+
+  /**
+   * Insert a message, repairing messages-table indexes once when SQLite reports corruption.
+   *
+   * FNXC:Messaging 2026-06-26-00:00:
+   * Sending a message must not leak a bare `database disk image is malformed` when only the `messages` lookup indexes are corrupt. The send path runs one scoped `REINDEX messages` repair and one retry, then distinguishes repair failure from post-repair table/database corruption so operator remediation targets the right store and database path.
+   */
+  private runInsertWithMessagesIndexRecovery(message: Message): void {
+    try {
+      this.runInsert(message);
+      return;
+    } catch (error) {
+      if (!isSqliteCorruptionError(error)) {
+        throw error;
+      }
+
+      try {
+        this.db.reindexMessages();
+      } catch (reindexError) {
+        if (isSqliteCorruptionError(reindexError)) {
+          throw this.createMessagesReindexFailureError(reindexError, error);
+        }
+        throw reindexError;
+      }
+
+      try {
+        this.runInsert(message);
+        return;
+      } catch (retryError) {
+        if (isSqliteCorruptionError(retryError)) {
+          throw this.createMessagesPostReindexCorruptionError(retryError, error);
+        }
+        throw retryError;
+      }
+    }
+  }
+
+  private runInsert(message: Message): void {
     this.stmtInsert.run(
       message.id,
       message.fromId,
@@ -164,17 +215,22 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
       message.createdAt,
       message.updatedAt,
     );
+  }
 
-    this.db.bumpLastModified();
+  private createMessagesReindexFailureError(error: unknown, originalError: unknown): Error {
-    messageStoreLog.log(`MessageStore emitting message:sent id=${message.id} type=${message.type} fromId=${message.fromId} toId=${message.toId}`);
+    const detail = error instanceof Error ? error.message : String(error);
-    this.emit("message:sent", message);
+    const original = originalError instanceof Error ? originalError.message : String(originalError);
-    this.emit("message:received", message);
+    return new Error(
+      `Messages store index repair failed (table=messages, db=${this.db.getPath()}) — REINDEX messages could not complete; run "fn db --vacuum" and inspect with "PRAGMA integrity_check" before retrying (original insert: ${original}; reindex: ${detail})`,
+    );
+  }
 
-    if (message.toType === "agent" && this.onMessageToAgent) {
+  private createMessagesPostReindexCorruptionError(error: unknown, originalError: unknown): Error {
-      this.onMessageToAgent(message);
+    const detail = error instanceof Error ? error.message : String(error);
-    }
+    const original = originalError instanceof Error ? originalError.message : String(originalError);
-
+    return new Error(
-    return message;
+      `Messages store table/database corruption after REINDEX (table=messages, db=${this.db.getPath()}) — REINDEX messages completed but sending still failed; run "fn db --vacuum" and inspect with "PRAGMA integrity_check" for messages table or database-file damage (original insert: ${original}; retry: ${detail})`,
+    );
   }
 
   /**
@@ -280,6 +336,7 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
     if (existing.read) return existing;
 
     const now = new Date().toISOString();
+    // Deferral: markAsRead updates idxMessagesTo but is not the reported fn_send_message failure path; the reusable recovery helper keeps this path ready for a follow-up without expanding this focused fix.
     this.stmtUpdateRead.run(now, messageId);
     this.db.bumpLastModified();
 
@@ -310,7 +367,7 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
     `).get(...participantIds, ownerType) as { count: number } | undefined;
     const count = unreadRow?.count ?? 0;
 
-    // Mark all as read
+    // Deferral: markAllAsRead updates idxMessagesTo, but sendMessage is the operator-blocking repro; leave bulk read-state recovery for a dedicated follow-up if observed.
     this.db.prepare(`
       UPDATE messages SET read = 1, updatedAt = ? WHERE ${toIdPredicate} AND toType = ? AND read = 0
     `).run(now, ...participantIds, ownerType);
@@ -331,6 +388,7 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
       throw new Error(`Message ${id} not found`);
     }
 
+    // Deferral: deleteMessage mutates messages indexes but is not on the fn_send_message delivery path; avoid adding extra retry semantics outside the scoped send repair.
     this.stmtDelete.run(id);
     this.db.bumpLastModified();
     this.emit("message:deleted", id);
@@ -349,6 +407,7 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
     const cutoff = new Date(Date.now() - maxAgeMs).toISOString();
 
     const deletedIds = this.db.transaction(() => {
+      // Deferral: cleanupOldMessages deletes through messages indexes in retention maintenance, not interactive messaging send; keep recovery limited to the reported INSERT path.
       const rows = this.db.prepare(`
         DELETE FROM messages
         WHERE updatedAt < ?