Merge pull request #30 from timothyjlaurent/fix/sqlite-corruption-wal-pragmas

fix: prevent SQLite B-tree corruption with WAL tuning and batched agent logs

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

@@ -2,7 +2,7 @@ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
 import { Database, createDatabase, toJson, toJsonNullable, fromJson, normalizeTaskComments } from "../db.js";
 import { DEFAULT_PROJECT_SETTINGS } from "../types.js";
 import { TaskStore } from "../store.js";
-import { mkdtempSync, existsSync, readFileSync } from "node:fs";
+import { mkdtempSync, existsSync, readFileSync, rmSync } from "node:fs";
 import { join, dirname } from "node:path";
 import { tmpdir } from "node:os";
 import { fileURLToPath } from "node:url";
@@ -58,7 +58,7 @@ describe("Database", () => {
       const journalSizeLimit = db.prepare("PRAGMA journal_size_limit").get() as { journal_size_limit: number };
 
       expect(synchronous.synchronous).toBe(1); // NORMAL
-      expect(autoCheckpoint.wal_autocheckpoint).toBe(1000);
+      expect(autoCheckpoint.wal_autocheckpoint).toBe(100);
       expect(journalSizeLimit.journal_size_limit).toBe(4_194_304);
     });
 
@@ -189,14 +189,67 @@ describe("Database", () => {
     it("does not overwrite existing config on re-init", () => {
       // Update the config
       db.prepare("UPDATE config SET nextId = 42 WHERE id = 1").run();
-      
+
       // Re-init
       db.init();
-      
+
       // Should keep updated value
       const row = db.prepare("SELECT nextId FROM config WHERE id = 1").get() as any;
       expect(row.nextId).toBe(42);
     });
+
+    it("sets wal_autocheckpoint to 100", () => {
+      const row = db.prepare("PRAGMA wal_autocheckpoint").get() as { wal_autocheckpoint: number };
+      expect(row.wal_autocheckpoint).toBe(100);
+    });
+
+    it("sets journal_size_limit to 4 MB", () => {
+      const row = db.prepare("PRAGMA journal_size_limit").get() as { journal_size_limit: number };
+      expect(row.journal_size_limit).toBe(4194304);
+    });
+
+    it("sets synchronous to NORMAL (1)", () => {
+      const row = db.prepare("PRAGMA synchronous").get() as { synchronous: number };
+      expect(row.synchronous).toBe(1); // NORMAL = 1
+    });
+
+    it("sets busy_timeout to 5000ms", () => {
+      const row = db.prepare("PRAGMA busy_timeout").get() as Record<string, number>;
+      // node:sqlite returns PRAGMA results as objects; the key name varies
+      const value = Object.values(row)[0];
+      expect(value).toBe(5000);
+    });
+
+    it("skips WAL PRAGMAs for in-memory databases", () => {
+      const memDb = new Database(":memory:", { inMemory: true });
+      memDb.init();
+      // journal_mode for :memory: is "memory", not "wal"
+      const row = memDb.prepare("PRAGMA journal_mode").get() as { journal_mode: string };
+      expect(row.journal_mode).toBe("memory");
+      memDb.close();
+    });
+  });
+
+  describe("startup integrity check", () => {
+    it("passes silently on a healthy database", () => {
+      // db was already init'd in beforeEach — no warning means pass
+      const result = db.prepare("PRAGMA integrity_check").get() as { integrity_check: string };
+      expect(result.integrity_check).toBe("ok");
+    });
+
+    it("init completes without throwing even on a fresh database", () => {
+      const freshDir = makeTmpDir();
+      const freshFusionDir = join(freshDir, ".fusion");
+      const freshDb = new Database(freshFusionDir);
+
+      try {
+        // init includes the integrity check — should not throw
+        expect(() => freshDb.init()).not.toThrow();
+      } finally {
+        freshDb.close();
+        rmSync(freshDir, { recursive: true, force: true });
+      }
+    });
   });
 
   describe("change detection", () => {

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

@@ -4376,6 +4376,7 @@ Task with acceptance criteria
 
       await store.appendAgentLog(task.id, "Hello world", "text");
       await store.appendAgentLog(task.id, "Read", "tool");
+      (store as any).flushAgentLogBuffer();
 
       const rows = (store as any).db.prepare(`
         SELECT taskId, text, type FROM agentLogEntries
@@ -4863,6 +4864,7 @@ Task with acceptance criteria
     it("deleting a task cascades agent log entry deletion", async () => {
       const task = await createTestTask();
       await store.appendAgentLog(task.id, "cascade me", "text");
+      (store as any).flushAgentLogBuffer();
 
       const before = (store as any).db.prepare(
         "SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
@@ -4962,6 +4964,125 @@ Task with acceptance criteria
       ).get("agentLogLegacyFileImportVersion") as { value: string } | undefined;
       expect(migrationRow?.value).toBe("1");
     });
+
+    describe("agent log buffering", () => {
+      it("buffers entries and flushes in a single transaction when buffer is full", async () => {
+        const task = await createTestTask();
+
+        // Fill the buffer to its max size (50)
+        for (let i = 0; i < 50; i++) {
+          await store.appendAgentLog(task.id, `entry ${i}`, "text");
+        }
+
+        // Validate DB persistence without invoking read-path auto-flush helpers.
+        const row = (store as any).db
+          .prepare("SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?")
+          .get(task.id) as { count: number };
+        expect(row.count).toBe(50);
+      });
+
+      it("auto-flushes buffered entries when getAgentLogs is called", async () => {
+        const task = await createTestTask();
+
+        // Write fewer than BUFFER_SIZE entries — these stay buffered
+        await store.appendAgentLog(task.id, "buffered 1", "text");
+        await store.appendAgentLog(task.id, "buffered 2", "text");
+
+        // getAgentLogs triggers a flush
+        const logs = await store.getAgentLogs(task.id);
+        expect(logs).toHaveLength(2);
+        expect(logs[0].text).toBe("buffered 1");
+        expect(logs[1].text).toBe("buffered 2");
+      });
+
+      it("auto-flushes buffered entries when getAgentLogCount is called", async () => {
+        const task = await createTestTask();
+
+        await store.appendAgentLog(task.id, "counted", "text");
+        const count = await store.getAgentLogCount(task.id);
+        expect(count).toBe(1);
+      });
+
+      it("auto-flushes before deleteTask so FK cascade finds the rows", async () => {
+        const task = await createTestTask();
+
+        await store.appendAgentLog(task.id, "to be cascaded", "text");
+        // Prove flush happens before delete
+        const flushSpy = vi.spyOn(store as any, "flushAgentLogBuffer");
+        await store.deleteTask(task.id);
+        expect(flushSpy).toHaveBeenCalled();
+        flushSpy.mockRestore();
+
+        const after = (store as any).db.prepare(
+          "SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
+        ).get(task.id) as { count: number };
+        expect(after.count).toBe(0);
+      });
+
+      it("flushes remaining entries on close without throwing", async () => {
+        // Disk-backed store required — in-memory data doesn't survive close+reopen
+        store.close();
+        store = new TaskStore(rootDir, globalDir); // no inMemoryDb
+        await store.init();
+
+        const task = await createTestTask();
+
+        await store.appendAgentLog(task.id, "flush on close", "text");
+        // close() should flush the buffer gracefully
+        expect(() => store.close()).not.toThrow();
+
+        // Re-open and verify the entry was persisted
+        store = new TaskStore(rootDir, globalDir);
+        await store.init();
+        const logs = await store.getAgentLogs(task.id);
+        expect(logs).toHaveLength(1);
+        expect(logs[0].text).toBe("flush on close");
+      });
+
+      it("close does not throw when flushing entries for already-deleted tasks", async () => {
+        const task = await createTestTask();
+
+        await store.appendAgentLog(task.id, "orphaned entry", "text");
+        // Flush so the entry is in the DB, then delete the task
+        (store as any).flushAgentLogBuffer();
+        await store.deleteTask(task.id);
+
+        // Now buffer another entry for the deleted task
+        await store.appendAgentLog(task.id, "ghost entry", "text");
+        // close() should not throw despite FK constraint violation on flush
+        expect(() => store.close()).not.toThrow();
+      });
+
+      it("emits agent:log event immediately even when buffered", async () => {
+        const task = await createTestTask();
+        const events: any[] = [];
+        store.on("agent:log", (entry) => events.push(entry));
+
+        await store.appendAgentLog(task.id, "immediate event", "text");
+
+        // Event fires immediately, even though DB write is deferred
+        expect(events).toHaveLength(1);
+        expect(events[0].text).toBe("immediate event");
+        expect(events[0].taskId).toBe(task.id);
+      });
+
+      it("flushes interleaved entries from multiple tasks correctly", async () => {
+        const taskA = await createTestTask();
+        const taskB = await store.createTask({ description: "Task B" });
+
+        // Interleave entries for two tasks
+        for (let i = 0; i < 25; i++) {
+          await store.appendAgentLog(taskA.id, `A-${i}`, "text");
+          await store.appendAgentLog(taskB.id, `B-${i}`, "text");
+        }
+        // 50 total = buffer full, triggers flush
+
+        const countA = await store.getAgentLogCount(taskA.id);
+        const countB = await store.getAgentLogCount(taskB.id);
+        expect(countA).toBe(25);
+        expect(countB).toBe(25);
+      });
+    });
   });
 
   describe("task comments", () => {

packages/core/src/db.ts

@@ -735,11 +735,14 @@ export class Database {
   private db: DatabaseSync;
   private readonly dbPath: string;
   private readonly inMemory: boolean;
+  /** Returns the database file path (or ":memory:" for in-memory databases). */
+  get path(): string { return this.dbPath; }
   corruptionDetected = false;
   /** Tracks transaction nesting depth for savepoint-based nested transactions. */
   private transactionDepth = 0;
   private readonly _fts5Available: boolean;
 
+
   constructor(fusionDir: string, options?: { inMemory?: boolean }) {
     // In-memory mode is a test-only fast path that swaps the on-disk
     // SQLite file for SQLite's `:memory:` connection. Schema + data live
@@ -792,9 +795,10 @@ export class Database {
       this.db.exec("PRAGMA busy_timeout = 5000");
       // In WAL mode NORMAL is nearly as durable as FULL with much lower fsync cost.
       this.db.exec("PRAGMA synchronous = NORMAL");
-      // Let WAL grow to roughly the journal size limit before auto-checkpointing.
-      // This avoids frequent synchronous checkpoints on log-heavy workloads.
-      this.db.exec("PRAGMA wal_autocheckpoint = 1000");
+      // Checkpoint every 100 pages (~400 KB) to keep WAL small and reduce
+      // corruption risk. More aggressive than the default 1000, but paired
+      // with journal_size_limit to prevent WAL bloat.
+      this.db.exec("PRAGMA wal_autocheckpoint = 100");
       // Bound WAL growth between checkpoints/maintenance cycles.
       this.db.exec("PRAGMA journal_size_limit = 4194304");
     } else {
@@ -943,6 +947,47 @@ export class Database {
    * and seed meta values.
    */
   init(): void {
+    // Startup integrity check — run BEFORE any writes to avoid
+    // compounding corruption. Attempts WAL checkpoint recovery on failure.
+    const integrity = this.integrityCheck();
+    if (!integrity.ok) {
+      this.corruptionDetected = true;
+      console.warn(`[fusion:db] Database integrity check FAILED for ${this.dbPath} — corruption detected`);
+      // Attempt WAL checkpoint recovery
+      try {
+        this.db.exec("PRAGMA wal_checkpoint(TRUNCATE)");
+        const recheck = this.integrityCheck();
+        if (recheck.ok) {
+          this.corruptionDetected = false;
+          console.warn(`[fusion:db] Database recovered via WAL checkpoint: ${this.dbPath}`);
+        } else {
+          const recheckMsg = ("errors" in recheck && Array.isArray(recheck.errors))
+            ? recheck.errors.slice(0, 3).join(" | ")
+            : "unknown";
+          console.error(
+            `[fusion:db] Database is corrupted and could not be auto-recovered. ` +
+            `Run: sqlite3 ${this.dbPath} ".recover" | sqlite3 ${this.dbPath}.recovered`,
+          );
+          throw new Error(
+            `[fusion:db] Refusing to initialize corrupted database at ${this.dbPath}. Integrity errors: ${recheckMsg}`,
+          );
+        }
+      } catch (err) {
+        // Re-throw our own abort error; wrap others
+        if (err instanceof Error && err.message.startsWith("[fusion:db] Refusing")) {
+          throw err;
+        }
+        const errMsg = err instanceof Error ? err.message : String(err);
+        console.error(
+          `[fusion:db] Database corruption detected for ${this.dbPath} and checkpoint recovery failed: ${errMsg}. ` +
+          "Manual recovery required.",
+        );
+        throw new Error(
+          `[fusion:db] Refusing to initialize corrupted database at ${this.dbPath}. Recovery error: ${errMsg}`,
+        );
+      }
+    }
+
     this.db.exec(SCHEMA_SQL);
 
     // Seed schemaVersion and lastModified idempotently
@@ -964,12 +1009,6 @@ export class Database {
     this.db.exec(
       `INSERT OR IGNORE INTO config (id, nextId, nextWorkflowStepId, settings, workflowSteps, updatedAt) VALUES (1, 1, 1, '${JSON.stringify(DEFAULT_PROJECT_SETTINGS)}', '[]', '${configNow}')`,
     );
-
-    const integrity = this.integrityCheck();
-    if (!integrity.ok) {
-      this.corruptionDetected = true;
-      console.warn("[fusion:db] Database integrity check FAILED — corruption detected");
-    }
   }
 
   /**

packages/core/src/store.ts

@@ -511,6 +511,24 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
   /** Cached TodoStore instance */
   private todoStore: TodoStore | null = null;
 
+  /** Buffer for batching agent log writes to reduce WAL pressure. */
+  private agentLogBuffer: Array<{
+    taskId: string;
+    timestamp: string;
+    text: string;
+    type: string;
+    detail: string | null;
+    agent: string | null;
+  }> = [];
+  /** Timer for flushing the agent log buffer. */
+  private agentLogFlushTimer: ReturnType<typeof setTimeout> | null = null;
+  /** Maximum buffer size before forced flush. */
+  private static readonly AGENT_LOG_BUFFER_SIZE = 50;
+  /** Flush interval in milliseconds. */
+  private static readonly AGENT_LOG_FLUSH_MS = 2000;
+  /** Absolute backlog cap — oldest entries are dropped when flushes keep failing. */
+  private static readonly MAX_AGENT_LOG_BACKLOG = 5_000;
+
   // Test-only: when true, both fusion.db and archive.db open as `:memory:`
   // SQLite connections instead of disk-backed files. Production code never
   // sets this; it's gated through an opt-in TaskStoreOptions field below.
@@ -541,8 +559,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
    */
   private get db(): Database {
     if (!this._db) {
-      this._db = new Database(this.fusionDir, { inMemory: this.inMemoryDb });
-      this._db.init();
+      const db = new Database(this.fusionDir, { inMemory: this.inMemoryDb });
+      try {
+        db.init();
+      } catch (error) {
+        db.close();
+        throw error;
+      }
+      this._db = db;
       // Auto-migrate legacy data if needed
       if (detectLegacyData(this.fusionDir)) {
         // Note: migrateFromLegacy is async but we need sync access.
@@ -555,8 +579,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
 
   private get archiveDb(): ArchiveDatabase {
     if (!this._archiveDb) {
-      this._archiveDb = new ArchiveDatabase(this.fusionDir, { inMemory: this.inMemoryDb });
-      this._archiveDb.init();
+      const db = new ArchiveDatabase(this.fusionDir, { inMemory: this.inMemoryDb });
+      try {
+        db.init();
+      } catch (error) {
+        db.close();
+        throw error;
+      }
+      this._archiveDb = db;
       this.migrateLegacyArchiveEntriesToArchiveDb();
     }
     return this._archiveDb;
@@ -567,8 +597,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     
     // Initialize SQLite database
     if (!this._db) {
-      this._db = new Database(this.fusionDir, { inMemory: this.inMemoryDb });
-      this._db.init();
+      const db = new Database(this.fusionDir, { inMemory: this.inMemoryDb });
+      try {
+        db.init();
+      } catch (error) {
+        db.close();
+        throw error;
+      }
+      this._db = db;
     }
     
     // Auto-migrate from legacy file-based storage
@@ -3789,6 +3825,9 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
 
   async deleteTask(id: string, options?: { removeDependencyReferences?: boolean }): Promise<Task> {
     return this.withTaskLock(id, async () => {
+      // Flush buffered agent logs inside the lock so no new appends for this
+      // task can sneak in between flush and DELETE.
+      this.flushAgentLogBuffer();
       const task = this.readTaskFromDb(id);
       if (!task) {
         throw new Error(`Task ${id} not found`);
@@ -4723,13 +4762,114 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
       ...(agent !== undefined && { agent }),
     };
 
-    this.db.prepare(`
-      INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
-      VALUES (?, ?, ?, ?, ?, ?)
-    `).run(taskId, timestamp, text, type, normalizedDetail ?? null, agent ?? null);
-
-    this.db.bumpLastModified();
+    // Buffer the entry for batched insertion to reduce WAL pressure.
+    // Drop oldest entries if backlog exceeds hard cap (prolonged outage).
+    if (this.agentLogBuffer.length >= TaskStore.MAX_AGENT_LOG_BACKLOG) {
+      const dropCount = this.agentLogBuffer.length - TaskStore.MAX_AGENT_LOG_BACKLOG + 1;
+      this.agentLogBuffer.splice(0, dropCount);
+      console.warn(
+        `[fusion] Dropped ${dropCount} buffered agent log entries — backlog cap reached (${this.db.path})`,
+      );
+    }
+    this.agentLogBuffer.push({
+      taskId,
+      timestamp,
+      text,
+      type,
+      detail: normalizedDetail ?? null,
+      agent: agent ?? null,
+    });
     this.emit("agent:log", entry);
+
+    if (this.agentLogBuffer.length >= TaskStore.AGENT_LOG_BUFFER_SIZE) {
+      try {
+        this.flushAgentLogBuffer();
+      } catch (err) {
+        // Size-triggered flush failed — log but don't crash the caller.
+        console.error(`[fusion] Size-triggered agent log flush failed (${this.db.path}):`, err);
+      }
+    } else if (!this.agentLogFlushTimer) {
+      this.agentLogFlushTimer = setTimeout(
+        () => {
+          try {
+            this.flushAgentLogBuffer();
+          } catch (err) {
+            // Timer-triggered flush failed — log but don't crash the process.
+            console.error(`[fusion] Timer-triggered agent log flush failed (${this.db.path}):`, err);
+          }
+        },
+        TaskStore.AGENT_LOG_FLUSH_MS,
+      );
+      this.agentLogFlushTimer.unref();
+    }
+  }
+
+  /**
+   * Flush all buffered agent log entries in a single transaction.
+   * Called when the buffer is full or on a timer.
+   */
+  private flushAgentLogBuffer(): void {
+    if (this.agentLogFlushTimer) {
+      clearTimeout(this.agentLogFlushTimer);
+      this.agentLogFlushTimer = null;
+    }
+    if (this.agentLogBuffer.length === 0) return;
+
+    // Snapshot the entries to flush. New entries appended during the
+    // synchronous transaction will appear past batch.length in
+    // this.agentLogBuffer, so we splice only the flushed count.
+    const batch = this.agentLogBuffer.slice();
+    const flushCount = batch.length;
+
+    let validEntries = batch;
+    let flushSucceeded = false;
+    try {
+      this.db.transaction(() => {
+        // Query live task IDs inside the transaction so the check is
+        // atomic with the inserts (prevents TOCTOU FK violations).
+        const liveTaskIds = new Set(
+          (this.db.prepare("SELECT id FROM tasks").all() as Array<{ id: string }>).map((r) => r.id),
+        );
+        validEntries = batch.filter((e) => liveTaskIds.has(e.taskId));
+        const dropped = batch.length - validEntries.length;
+        if (dropped > 0) {
+          console.warn(
+            `[fusion] Dropped ${dropped} buffered agent log entries for deleted tasks (${this.db.path})`,
+          );
+        }
+
+        if (validEntries.length > 0) {
+          const stmt = this.db.prepare(`
+            INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
+            VALUES (?, ?, ?, ?, ?, ?)
+          `);
+          for (const entry of validEntries) {
+            stmt.run(entry.taskId, entry.timestamp, entry.text, entry.type, entry.detail, entry.agent);
+          }
+          this.db.bumpLastModified();
+        }
+      });
+      flushSucceeded = true;
+    } finally {
+      // Always drain the original slice from the buffer.
+      this.agentLogBuffer.splice(0, flushCount);
+      // On transient failures (busy/IO), requeue valid entries for retry.
+      // Stale rows were already filtered out above.
+      if (!flushSucceeded && validEntries.length > 0) {
+        this.agentLogBuffer.unshift(...validEntries);
+        // Re-arm the flush timer so retried entries don't sit in memory forever.
+        if (!this.agentLogFlushTimer) {
+          this.agentLogFlushTimer = setTimeout(() => {
+            try {
+              this.flushAgentLogBuffer();
+            } catch (err) {
+              console.error(`[fusion] Retry agent log flush failed (${this.db.path}):`, err);
+            }
+          }, TaskStore.AGENT_LOG_FLUSH_MS);
+          this.agentLogFlushTimer.unref();
+        }
+      }
+    }
   }
 
   async appendAgentLogBatch(
@@ -4745,6 +4885,10 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
       return;
     }
 
+    // Flush buffered single-entry appends so they land before batch entries,
+    // preserving insertion order (same-timestamp entries are ordered by rowid).
+    this.flushAgentLogBuffer();
+
     const timestamp = new Date().toISOString();
     const normalizedEntries = entries.map((entry) => ({
       ...entry,
@@ -4766,9 +4910,9 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
           entry.agent ?? null,
         );
       }
+      this.db.bumpLastModified();
     });
 
-    this.db.bumpLastModified();
     for (const entry of normalizedEntries) {
       this.emit("agent:log", {
         timestamp,
@@ -5426,6 +5570,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     taskId: string,
     options?: { limit?: number; offset?: number },
   ): Promise<AgentLogEntry[]> {
+    // Ensure buffered entries are visible before reading.
+    this.flushAgentLogBuffer();
     const limit = options?.limit !== undefined
       ? (Number.isFinite(options.limit) ? Math.max(0, Math.floor(options.limit)) : 0)
       : undefined;
@@ -5471,6 +5617,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
    * @returns Total number of log entries
    */
   async getAgentLogCount(taskId: string): Promise<number> {
+    this.flushAgentLogBuffer();
     const row = this.db.prepare(
       "SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
     ).get(taskId) as { count: number } | undefined;
@@ -5490,6 +5637,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     startIso: string,
     endIso: string | null,
   ): Promise<AgentLogEntry[]> {
+    // Ensure buffered entries are visible before reading.
+    this.flushAgentLogBuffer();
     const end = endIso ?? new Date().toISOString();
     const selectClause = this.getAgentLogSelectClause();
     const rows = this.db.prepare(`
@@ -6094,6 +6243,23 @@ ${stepsSection}`;
    */
   close(): void {
     this.stopWatching();
+    // Flush any remaining buffered agent log entries before closing.
+    // Wrap in try-catch because entries for already-deleted tasks will fail FK check.
+    if (this.agentLogBuffer.length > 0) {
+      try {
+        this.flushAgentLogBuffer();
+      } catch (err) {
+        // Best-effort flush — entries for deleted tasks will fail FK check.
+        // Log the error instead of silently swallowing it.
+        console.warn(`[fusion] Could not flush remaining agent log entries on close:`, err);
+      }
+    }
+    // Cancel any retry timer armed by a failed flush — the DB is about to close.
+    if (this.agentLogFlushTimer) {
+      clearTimeout(this.agentLogFlushTimer);
+      this.agentLogFlushTimer = null;
+    }
+    this.agentLogBuffer.length = 0;
     if (this._db) {
       this._db.close();
       this._db = null;