fix(memory): reject stale reindex publishes

extensions/memory-core/src/memory/manager-db.test.ts

@@ -5,7 +5,11 @@ import path from "node:path";
 import { DatabaseSync } from "node:sqlite";
 import { ensureMemoryIndexSchema } from "openclaw/plugin-sdk/memory-core-host-engine-storage";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
-import { cleanupAgedMemoryReindexTempFiles, publishMemoryDatabaseTables } from "./manager-db.js";
+import {
+  cleanupAgedMemoryReindexTempFiles,
+  publishMemoryDatabaseTables,
+  readMemoryDatabaseRevision,
+} from "./manager-db.js";
 import { acquireMemoryReindexLock } from "./manager-reindex-lock.js";
 
 function ensureTestMemorySchema(db: DatabaseSync): void {
@@ -49,6 +53,7 @@ describe("memory manager database publication", () => {
         targetDb,
         sourcePath,
         metaKey: "memory_index_meta",
+        expectedRevision: readMemoryDatabaseRevision(targetDb),
       });
 
       expect(
@@ -64,6 +69,51 @@ describe("memory manager database publication", () => {
     }
   });
 
+  it("rejects a stale shadow publish after a concurrent live memory update", () => {
+    const targetPath = path.join(fixtureRoot, "target.sqlite");
+    const sourcePath = path.join(fixtureRoot, "source.sqlite");
+    const targetDb = new DatabaseSync(targetPath);
+    const sourceDb = new DatabaseSync(sourcePath);
+    let concurrentDb: DatabaseSync | undefined;
+    try {
+      ensureTestMemorySchema(targetDb);
+      ensureTestMemorySchema(sourceDb);
+      targetDb
+        .prepare("INSERT INTO files (path, source, hash, mtime, size) VALUES (?, ?, ?, ?, ?)")
+        .run("memory.md", "memory", "published", 1, 1);
+      sourceDb
+        .prepare("INSERT INTO files (path, source, hash, mtime, size) VALUES (?, ?, ?, ?, ?)")
+        .run("memory.md", "memory", "shadow", 1, 1);
+      const expectedRevision = readMemoryDatabaseRevision(targetDb);
+      sourceDb.close();
+
+      concurrentDb = new DatabaseSync(targetPath);
+      concurrentDb.prepare("UPDATE files SET hash = ? WHERE path = ?").run("newer", "memory.md");
+      concurrentDb.close();
+      concurrentDb = undefined;
+
+      expect(() =>
+        publishMemoryDatabaseTables({
+          targetDb,
+          sourcePath,
+          metaKey: "memory_index_meta",
+          expectedRevision,
+        }),
+      ).toThrow(/changed while full reindex was building/);
+      expect(
+        targetDb.prepare("SELECT hash FROM files WHERE path = ?").get("memory.md"),
+      ).toEqual({ hash: "newer" });
+    } finally {
+      try {
+        concurrentDb?.close();
+      } catch {}
+      try {
+        sourceDb.close();
+      } catch {}
+      targetDb.close();
+    }
+  });
+
   it("removes aged orphan shadows but preserves young and locked shadows", async () => {
     const databasePath = path.join(fixtureRoot, "agent.sqlite");
     const database = new DatabaseSync(databasePath);

extensions/memory-core/src/memory/manager-db.ts

@@ -18,6 +18,7 @@ import {
 } from "./manager-reindex-lock.js";
 
 const MEMORY_REINDEX_SCHEMA = "memory_reindex";
+const MEMORY_INDEX_STATE_ID = 1;
 const MEMORY_DATABASE_FILE_SUFFIXES = ["", "-wal", "-shm", "-journal"] as const;
 const MEMORY_REINDEX_ENTRY_SUFFIXES = ["-wal", "-shm", "-journal", ""] as const;
 const MEMORY_REINDEX_UUID_PATTERN =
@@ -66,6 +67,16 @@ function readTableSql(db: DatabaseSync, schema: string, tableName: string): stri
   return typeof row?.sql === "string" && row.sql.trim() ? row.sql : null;
 }
 
+export function readMemoryDatabaseRevision(db: DatabaseSync): number {
+  const row = db
+    .prepare("SELECT revision FROM memory_index_state WHERE id = ?")
+    .get(MEMORY_INDEX_STATE_ID) as { revision?: unknown } | undefined;
+  if (typeof row?.revision !== "number" || !Number.isSafeInteger(row.revision)) {
+    throw new Error("Memory index revision is missing or invalid");
+  }
+  return row.revision;
+}
+
 function replaceVirtualTable(params: {
   db: DatabaseSync;
   tableName: "chunks_fts" | "chunks_vec";
@@ -97,10 +108,18 @@ export function publishMemoryDatabaseTables(params: {
   targetDb: DatabaseSync;
   sourcePath: string;
   metaKey: string;
+  expectedRevision: number;
 }): void {
   params.targetDb.prepare(`ATTACH DATABASE ? AS ${MEMORY_REINDEX_SCHEMA}`).run(params.sourcePath);
   try {
     runSqliteImmediateTransactionSync(params.targetDb, () => {
+      const liveRevision = readMemoryDatabaseRevision(params.targetDb);
+      if (liveRevision !== params.expectedRevision) {
+        throw new Error(
+          `Memory index changed while full reindex was building ` +
+            `(expected revision ${params.expectedRevision}, found ${liveRevision}); retry the full reindex.`,
+        );
+      }
       params.targetDb.prepare("DELETE FROM main.meta WHERE key = ?").run(params.metaKey);
       params.targetDb
         .prepare(

extensions/memory-core/src/memory/manager-sync-ops.ts

@@ -51,6 +51,7 @@ import {
   closeMemoryDatabase,
   openMemoryDatabaseAtPath,
   publishMemoryDatabaseTables,
+  readMemoryDatabaseRevision,
   removeMemoryDatabaseFiles,
 } from "./manager-db.js";
 import { isMemoryEmbeddingOperationError } from "./manager-embedding-errors.js";
@@ -2453,6 +2454,7 @@ export abstract class MemoryManagerSyncOps {
     try {
       cleanupAgedMemoryReindexTempFiles(dbPath);
       reindexLock = acquireMemoryReindexLock(dbPath);
+      const originalRevision = readMemoryDatabaseRevision(originalDb);
       tempDb = openMemoryDatabaseAtPath(tempDbPath, this.settings.store.vector.enabled);
       this.db = tempDb;
       this.lastMetaSerialized = null;
@@ -2532,6 +2534,7 @@ export abstract class MemoryManagerSyncOps {
         targetDb: originalDb,
         sourcePath: tempDbPath,
         metaKey: META_KEY,
+        expectedRevision: originalRevision,
       });
 
       this.db = originalDb;

packages/memory-host-sdk/src/host/memory-schema.ts

@@ -42,6 +42,45 @@ export function ensureMemoryIndexSchema(params: {
       updated_at INTEGER NOT NULL
     );
   `);
+  params.db.exec(`
+    CREATE TABLE IF NOT EXISTS memory_index_state (
+      id INTEGER PRIMARY KEY CHECK (id = 1),
+      revision INTEGER NOT NULL
+    );
+    INSERT OR IGNORE INTO memory_index_state (id, revision) VALUES (1, 0);
+
+    CREATE TRIGGER IF NOT EXISTS memory_files_revision_after_insert
+    AFTER INSERT ON files
+    BEGIN
+      UPDATE memory_index_state SET revision = revision + 1 WHERE id = 1;
+    END;
+    CREATE TRIGGER IF NOT EXISTS memory_files_revision_after_update
+    AFTER UPDATE ON files
+    BEGIN
+      UPDATE memory_index_state SET revision = revision + 1 WHERE id = 1;
+    END;
+    CREATE TRIGGER IF NOT EXISTS memory_files_revision_after_delete
+    AFTER DELETE ON files
+    BEGIN
+      UPDATE memory_index_state SET revision = revision + 1 WHERE id = 1;
+    END;
+
+    CREATE TRIGGER IF NOT EXISTS memory_chunks_revision_after_insert
+    AFTER INSERT ON chunks
+    BEGIN
+      UPDATE memory_index_state SET revision = revision + 1 WHERE id = 1;
+    END;
+    CREATE TRIGGER IF NOT EXISTS memory_chunks_revision_after_update
+    AFTER UPDATE ON chunks
+    BEGIN
+      UPDATE memory_index_state SET revision = revision + 1 WHERE id = 1;
+    END;
+    CREATE TRIGGER IF NOT EXISTS memory_chunks_revision_after_delete
+    AFTER DELETE ON chunks
+    BEGIN
+      UPDATE memory_index_state SET revision = revision + 1 WHERE id = 1;
+    END;
+  `);
   if (params.cacheEnabled) {
     params.db.exec(`
       CREATE TABLE IF NOT EXISTS ${params.embeddingCacheTable} (